zalo-hermes-bot/bot_app.py

#!/usr/bin/env python3
"""Zalo Bot → Hermes Agent (API Server). Không dùng support KB :8766."""

from __future__ import annotations

import asyncio
import logging
import os
import warnings
from concurrent.futures import ThreadPoolExecutor

from dotenv import load_dotenv
from pathlib import Path

# Load .env TRƯỚC mọi import đọc os.environ (hermes_client / BOT_TOKEN)
load_dotenv(Path(__file__).resolve().parent / ".env")

import zalo_bot

from format_zalo import format_for_zalo
from hermes_client import (
    HERMES_API_BASE,
    HERMES_BACKEND,
    ask_hermes,
    clear_history,
    hermes_health,
)

logging.getLogger("zalo_bot").setLevel(logging.ERROR)
warnings.filterwarnings("ignore")

BOT_TOKEN = os.getenv("BOT_TOKEN")
if not BOT_TOKEN:
    raise ValueError("BOT_TOKEN not found in .env")

_HERMES_WORKERS = int(os.getenv("HERMES_WORKERS", "2"))
_executor = ThreadPoolExecutor(max_workers=max(1, _HERMES_WORKERS))
_busy: set[str] = set()


def split_zalo_text(text: str, limit: int = 1800) -> list[str]:
    text = (text or "").strip()
    if not text:
        return ["(không có nội dung)"]
    if len(text) <= limit:
        return [text]
    parts: list[str] = []
    rest = text
    while rest:
        if len(rest) <= limit:
            parts.append(rest)
            break
        cut = rest.rfind("\n", 0, limit)
        if cut < limit // 3:
            cut = rest.rfind(" ", 0, limit)
        if cut < limit // 3:
            cut = limit
        parts.append(rest[:cut].rstrip())
        rest = rest[cut:].lstrip()
    return parts or [text[:limit]]


def get_answer(chat_id: str, question: str) -> str:
    q = (question or "").strip()
    if not q:
        return "Bạn vui lòng gửi nội dung cần hỏi nhé."

    if q.lower() in {"/reset", "/new", "reset", "xóa lịch sử", "xoa lich su"}:
        clear_history(chat_id)
        return "Đã xóa lịch sử hội thoại với Hermes cho chat này."

    result = ask_hermes(chat_id, q)
    answer = (result.get("answer") or "").strip()
    plain = format_for_zalo(answer)
    return plain or answer


async def send_long_message(bot: zalo_bot.Bot, chat_id: str, text: str) -> None:
    for part in split_zalo_text(text):
        await bot.send_message(chat_id, part)


async def handle_user_message(bot: zalo_bot.Bot, chat_id: str, question: str) -> None:
    """Chạy nền: Hermes → trả Zalo (không chặn long-poll)."""
    key = str(chat_id)
    if key in _busy:
        await bot.send_message(
            chat_id,
            "Mình đang xử lý câu trước của bạn. Vui lòng đợi một chút…",
        )
        return
    _busy.add(key)
    try:
        try:
            await bot.send_chat_action(chat_id, "typing")
        except Exception:
            pass

        loop = asyncio.get_running_loop()
        answer = await loop.run_in_executor(_executor, get_answer, chat_id, question)
        await send_long_message(bot, chat_id, answer)
        print(f"[zalo] Sent to {chat_id} ({len(answer)} chars)")
    except Exception as e:
        print(f"[zalo] handle error for {chat_id}: {e}")
        try:
            await bot.send_message(
                chat_id,
                "Có lỗi khi nhờ Hermes trả lời. Bạn thử lại sau nhé.",
            )
        except Exception:
            pass
    finally:
        _busy.discard(key)


async def main() -> None:
    health = hermes_health()
    print(f"[hermes] backend={HERMES_BACKEND} base={HERMES_API_BASE} health={health}")
    if not health.get("ok"):
        print(
            "[hermes] WARNING: API Server chưa sẵn sàng. "
            "Bật API_SERVER_ENABLED=true + API_SERVER_KEY trong ~/.hermes/.env "
            "rồi systemctl --user restart hermes-gateway "
            "(listen http://127.0.0.1:8642)."
        )

    bot = zalo_bot.Bot(BOT_TOKEN)
    async with bot:
        me = await bot.get_me()
        print(
            f"[zalo] started as {getattr(me, 'account_name', None) or me} "
            f"id={getattr(me, 'id', None)}; listening…"
        )

        while True:
            try:
                update = await bot.get_update(timeout=30)
                if not (update and update.message and update.message.text):
                    continue

                chat_id = update.message.chat.id
                question = update.message.text
                print(f"[zalo] Message from {chat_id}: {question}")
                asyncio.create_task(handle_user_message(bot, chat_id, question))

            except asyncio.CancelledError:
                raise
            except KeyboardInterrupt:
                raise
            except zalo_bot.error.ZaloError:
                await asyncio.sleep(1)
            except Exception as e:
                print(f"[zalo] Unexpected error: {e}")
                await asyncio.sleep(2)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n[zalo] stopped (Ctrl+C)")
    finally:
        _executor.shutdown(wait=False, cancel_futures=True)