#!/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)