190 lines
8.7 KiB
Python
190 lines
8.7 KiB
Python
"""长期运行、连接监督与本地控制台命令。"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import queue
|
|
import shlex
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
from .backend import CAPABILITIES, TEXT_CAPABILITIES, NativeFileHelperBackend, NativeTextBackend
|
|
from .plugins import PluginManager
|
|
from .sdk import ServiceFacade
|
|
from .members import MemberStore
|
|
from .personality import prepare_switch
|
|
|
|
logger = logging.getLogger("Bot")
|
|
|
|
|
|
class BotApplication:
|
|
def __init__(self, root, store, *, backend_factory=None):
|
|
self.root, self.store = root, store
|
|
self.config = store.load()
|
|
self.service = ServiceFacade(text_delay=self.config.get("text_send_delay"))
|
|
self.members = MemberStore.from_config(self.config, self.store.path.parent)
|
|
generic = self.config.get("backend") == "native_text"
|
|
capabilities = TEXT_CAPABILITIES if generic else CAPABILITIES
|
|
self.plugins = PluginManager(root / "plugins", store, self.service, capabilities,
|
|
self.config.get("plugin_queue_size", 128), member_store=self.members)
|
|
backend_type = NativeTextBackend if generic else NativeFileHelperBackend
|
|
self.backend_factory = backend_factory or (lambda: backend_type(root, self.config))
|
|
self.backend = None
|
|
self.stop_event = threading.Event()
|
|
self.commands = queue.Queue()
|
|
self.last_cleanup = None
|
|
self._connection_ready = False
|
|
self._last_connection_log = 0
|
|
self._last_connection_error = ""
|
|
|
|
def _log_connection_error(self, error):
|
|
text = f"{type(error).__name__}: {error}"
|
|
now = time.monotonic()
|
|
if text != self._last_connection_error or now - self._last_connection_log >= 30:
|
|
logger.warning("等待微信连接：%s", text)
|
|
self._last_connection_error, self._last_connection_log = text, now
|
|
|
|
def _connect(self):
|
|
try:
|
|
candidate = self.backend_factory()
|
|
self.backend = candidate
|
|
self.service.bind(candidate)
|
|
candidate.start(self._emit)
|
|
self._connection_ready = True
|
|
self._last_connection_error = ""
|
|
logger.info("微信已连接，PID=%s；收发范围：%s", candidate.pid,
|
|
getattr(candidate, "scope", "文本"))
|
|
except Exception as error:
|
|
self._connection_ready = False
|
|
self._log_connection_error(error)
|
|
self._disconnect()
|
|
|
|
def _emit(self, message):
|
|
allowed = self.config.get("chat_allowlist", [])
|
|
if not allowed or message.chat_id in allowed:
|
|
if self.members is not None:
|
|
try:
|
|
self.members.record(message)
|
|
except Exception:
|
|
logger.exception("群文本保存失败，消息继续分发到插件")
|
|
if message.data.get("dispatch_to_plugins", True):
|
|
if self.plugins.is_active("personality"):
|
|
message = prepare_switch(message, self.members)
|
|
self.plugins.dispatch(message)
|
|
|
|
def _disconnect(self):
|
|
self._connection_ready = False
|
|
try:
|
|
self.last_cleanup = self.service.close()
|
|
self.backend = None
|
|
return True
|
|
except Exception as error:
|
|
# 清理忙时保留实例，只重试清理，不覆盖 DLL、不再次注入。
|
|
self._log_connection_error(error)
|
|
return False
|
|
|
|
def execute(self, command):
|
|
words = shlex.split(command)
|
|
if not words:
|
|
return
|
|
action = words[0].lower()
|
|
if action in ("quit", "exit", "stop") and len(words) == 1:
|
|
self.stop_event.set()
|
|
elif action == "help" and len(words) == 1:
|
|
logger.info("命令：status / chats / members [群ID] / history 称呼 [条数] / plugins / enable 名称 / disable 名称 / reload [名称] / quit")
|
|
elif action == "status" and len(words) == 1:
|
|
logger.info("状态：%s", json.dumps({"connected": self._connection_ready,
|
|
"pid": self.backend.pid if self.backend else None,
|
|
"native": self.backend.diagnostics() if self.backend and hasattr(self.backend, "diagnostics") else {},
|
|
"plugins": self.plugins.status()}, ensure_ascii=False))
|
|
elif action == "chats" and len(words) == 1:
|
|
rows = self.backend.chats() if self.backend and hasattr(self.backend, "chats") else []
|
|
logger.info("本次连接已接收的会话：%s", json.dumps(rows, ensure_ascii=False))
|
|
elif action == "members" and len(words) in (1, 2):
|
|
if self.members is None:
|
|
raise ValueError("尚未配置成员目录")
|
|
rows = [{"name": m.name, "display_name": m.display_name, "wxid": m.wxid,
|
|
"room": m.room, "bound": bool(m.wxid)} for m in self.members.list_members(words[1] if len(words) == 2 else None)]
|
|
logger.info("成员目录：%s", json.dumps(rows, ensure_ascii=False))
|
|
elif action == "history" and len(words) in (2, 3):
|
|
if self.members is None:
|
|
raise ValueError("尚未配置成员目录")
|
|
rows = self.members.history(words[1], limit=int(words[2]) if len(words) == 3 else 10)
|
|
logger.info("成员聊天记录：%s", json.dumps(rows, ensure_ascii=False))
|
|
elif action == "plugins" and len(words) == 1:
|
|
for row in self.plugins.status():
|
|
state = "运行" if row["active"] else "停用"
|
|
logger.info("%s [%s] 已处理=%s 异常=%s 丢弃=%s %s", row["name"], state,
|
|
row["handled"], row["errors"], row["dropped"], row["error"])
|
|
elif action in ("enable", "disable") and len(words) == 2:
|
|
getattr(self.plugins, action)(words[1])
|
|
elif action == "reload" and len(words) in (1, 2):
|
|
self.plugins.reload(words[1] if len(words) == 2 else None)
|
|
else:
|
|
raise ValueError("未知命令；输入 help 查看用法")
|
|
|
|
def _read_console(self):
|
|
# stdin 关闭时仍继续运行，适合后台启动。
|
|
try:
|
|
for line in sys.stdin:
|
|
if self.stop_event.is_set():
|
|
break
|
|
self.commands.put(line.strip())
|
|
except (OSError, ValueError):
|
|
pass
|
|
|
|
def run(self, *, seconds=0):
|
|
started = time.monotonic()
|
|
next_connect = 0
|
|
next_rollover = 0
|
|
try:
|
|
if self.members is not None:
|
|
self.members.initialize()
|
|
self.plugins.reload()
|
|
if self.config.get("console", True):
|
|
threading.Thread(target=self._read_console, name="BotConsole", daemon=True).start()
|
|
logger.info("框架已启动；输入 help 查看插件管理命令，Ctrl+C 退出")
|
|
while not self.stop_event.is_set():
|
|
now = time.monotonic()
|
|
if self.members is not None and now >= next_rollover:
|
|
self.members.rollover()
|
|
next_rollover = now + 0.5
|
|
if seconds and now - started >= seconds:
|
|
break
|
|
while True:
|
|
try:
|
|
command = self.commands.get_nowait()
|
|
except queue.Empty:
|
|
break
|
|
try:
|
|
self.execute(command)
|
|
except Exception:
|
|
logger.exception("控制台命令执行失败")
|
|
if self.stop_event.is_set():
|
|
break
|
|
if self._connection_ready and self.backend.last_error:
|
|
self._log_connection_error(self.backend.last_error)
|
|
self._disconnect()
|
|
next_connect = now + self.config.get("reconnect_seconds", 5)
|
|
elif not self._connection_ready and now >= next_connect:
|
|
if self.backend is None or self._disconnect():
|
|
self._connect()
|
|
next_connect = time.monotonic() + self.config.get("reconnect_seconds", 5)
|
|
self.stop_event.wait(0.1)
|
|
except KeyboardInterrupt:
|
|
logger.info("收到退出指令")
|
|
finally:
|
|
self.stop()
|
|
|
|
def stop(self):
|
|
self.stop_event.set()
|
|
self.service.cancel_pending()
|
|
self.plugins.shutdown()
|
|
cleaned = self._disconnect()
|
|
if self.members is not None:
|
|
self.members.close()
|
|
if not cleaned:
|
|
raise RuntimeError("原生连接未能清理；保留 DLL，请查看日志后再启动")
|
|
logger.info("框架已停止；连接清理：%s", self.last_cleanup)
|