Refactor chat app adapters to reduce duplicated scaffolding

chatapp_common.py

@@ -0,0 +1,178 @@
+import asyncio, glob, os, queue as Q, re, socket, sys, time
+
+HELP_TEXT = "📖 命令列表:\n/help - 显示帮助\n/status - 查看状态\n/stop - 停止当前任务\n/new - 清空当前上下文\n/restore - 恢复上次对话历史\n/llm [n] - 查看或切换模型"
+FILE_HINT = "If you need to show files to user, use [FILE:filepath] in your response."
+TAG_PATS = [r"<" + t + r">.*?</" + t + r">" for t in ("thinking", "summary", "tool_use", "file_content")]
+
+
+def clean_reply(text):
+    for pat in TAG_PATS:
+        text = re.sub(pat, "", text or "", flags=re.DOTALL)
+    return re.sub(r"\n{3,}", "\n\n", text).strip() or "..."
+
+
+def extract_files(text):
+    return re.findall(r"\[FILE:([^\]]+)\]", text or "")
+
+
+def strip_files(text):
+    return re.sub(r"\[FILE:[^\]]+\]", "", text or "").strip()
+
+
+def split_text(text, limit):
+    text, parts = (text or "").strip() or "...", []
+    while len(text) > limit:
+        cut = text.rfind("\n", 0, limit)
+        if cut < limit * 0.6:
+            cut = limit
+        parts.append(text[:cut].rstrip())
+        text = text[cut:].lstrip()
+    return parts + ([text] if text else []) or ["..."]
+
+
+def format_restore():
+    files = glob.glob("./temp/model_responses_*.txt")
+    if not files:
+        return None, "❌ 没有找到历史记录"
+    latest = max(files, key=os.path.getmtime)
+    with open(latest, "r", encoding="utf-8") as f:
+        content = f.read()
+    users = re.findall(r"=== USER ===\n(.+?)(?==== |$)", content, re.DOTALL)
+    resps = re.findall(r"=== Response ===.*?\n(.+?)(?==== Prompt|$)", content, re.DOTALL)
+    restored = []
+    for u, r in zip(users, resps):
+        u, r = u.strip(), r.strip()[:500]
+        if u and r:
+            restored.extend([f"[USER]: {u}", f"[Agent] {r}"])
+    if not restored:
+        return None, "❌ 历史记录里没有可恢复内容"
+    return (restored, os.path.basename(latest), len(restored) // 2), None
+
+
+def build_done_text(raw_text):
+    files = [p for p in extract_files(raw_text) if os.path.exists(p)]
+    body = strip_files(clean_reply(raw_text))
+    if files:
+        body = (body + "\n\n" if body else "") + "\n".join(f"生成文件: {p}" for p in files)
+    return body or "..."
+
+
+def public_access(allowed):
+    return not allowed or "*" in allowed
+
+
+def allowed_label(allowed):
+    return "public" if public_access(allowed) else sorted(allowed)
+
+
+def ensure_single_instance(port, label):
+    try:
+        lock_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        lock_sock.bind(("127.0.0.1", port))
+        return lock_sock
+    except OSError:
+        print(f"[{label}] Another instance is already running, skipping...")
+        sys.exit(1)
+
+
+def require_runtime(agent, label, **required):
+    missing = [k for k, v in required.items() if not v]
+    if missing:
+        print(f"[{label}] ERROR: please set {', '.join(missing)} in mykey.py or mykey.json")
+        sys.exit(1)
+    if agent.llmclient is None:
+        print(f"[{label}] ERROR: no usable LLM backend found in mykey.py or mykey.json")
+        sys.exit(1)
+
+
+def redirect_log(script_file, log_name, label, allowed):
+    log_dir = os.path.join(os.path.dirname(script_file), "temp")
+    os.makedirs(log_dir, exist_ok=True)
+    logf = open(os.path.join(log_dir, log_name), "a", encoding="utf-8", buffering=1)
+    sys.stdout = sys.stderr = logf
+    print(f"[NEW] {label} process starting, the above are history infos ...")
+    print(f"[{label}] allow list: {allowed_label(allowed)}")
+
+
+class AgentChatMixin:
+    label = "Chat"
+    source = "chat"
+    split_limit = 1500
+    ping_interval = 20
+
+    def __init__(self, agent, user_tasks):
+        self.agent, self.user_tasks = agent, user_tasks
+
+    async def send_text(self, chat_id, content, **ctx):
+        raise NotImplementedError
+
+    async def send_done(self, chat_id, raw_text, **ctx):
+        await self.send_text(chat_id, build_done_text(raw_text), **ctx)
+
+    async def handle_command(self, chat_id, cmd, **ctx):
+        parts = (cmd or "").split()
+        op = (parts[0] if parts else "").lower()
+        if op == "/stop":
+            state = self.user_tasks.get(chat_id)
+            if state:
+                state["running"] = False
+            self.agent.abort()
+            return await self.send_text(chat_id, "⏹️ 正在停止...", **ctx)
+        if op == "/status":
+            llm = self.agent.get_llm_name() if self.agent.llmclient else "未配置"
+            return await self.send_text(chat_id, f"状态: {'🔴 运行中' if self.agent.is_running else '🟢 空闲'}\nLLM: [{self.agent.llm_no}] {llm}", **ctx)
+        if op == "/llm":
+            if not self.agent.llmclient:
+                return await self.send_text(chat_id, "❌ 当前没有可用的 LLM 配置", **ctx)
+            if len(parts) > 1:
+                try:
+                    self.agent.next_llm(int(parts[1]))
+                    return await self.send_text(chat_id, f"✅ 已切换到 [{self.agent.llm_no}] {self.agent.get_llm_name()}", **ctx)
+                except Exception:
+                    return await self.send_text(chat_id, f"用法: /llm <0-{len(self.agent.list_llms()) - 1}>", **ctx)
+            lines = [f"{'→' if cur else '  '} [{i}] {name}" for i, name, cur in self.agent.list_llms()]
+            return await self.send_text(chat_id, "LLMs:\n" + "\n".join(lines), **ctx)
+        if op == "/restore":
+            try:
+                restored_info, err = format_restore()
+                if err:
+                    return await self.send_text(chat_id, err, **ctx)
+                restored, fname, count = restored_info
+                self.agent.abort()
+                self.agent.history.extend(restored)
+                return await self.send_text(chat_id, f"✅ 已恢复 {count} 轮对话\n来源: {fname}\n(仅恢复上下文，请输入新问题继续)", **ctx)
+            except Exception as e:
+                return await self.send_text(chat_id, f"❌ 恢复失败: {e}", **ctx)
+        if op == "/new":
+            self.agent.abort()
+            self.agent.history = []
+            return await self.send_text(chat_id, "🆕 已清空当前共享上下文", **ctx)
+        return await self.send_text(chat_id, HELP_TEXT, **ctx)
+
+    async def run_agent(self, chat_id, text, **ctx):
+        state = {"running": True}
+        self.user_tasks[chat_id] = state
+        try:
+            await self.send_text(chat_id, "思考中...", **ctx)
+            dq = self.agent.put_task(f"{FILE_HINT}\n\n{text}", source=self.source)
+            last_ping = time.time()
+            while state["running"]:
+                try:
+                    item = await asyncio.to_thread(dq.get, True, 3)
+                except Q.Empty:
+                    if self.agent.is_running and time.time() - last_ping > self.ping_interval:
+                        await self.send_text(chat_id, "⏳ 还在处理中，请稍等...", **ctx)
+                        last_ping = time.time()
+                    continue
+                if "done" in item:
+                    await self.send_done(chat_id, item.get("done", ""), **ctx)
+                    break
+            if not state["running"]:
+                await self.send_text(chat_id, "⏹️ 已停止", **ctx)
+        except Exception as e:
+            import traceback
+            print(f"[{self.label}] run_agent error: {e}")
+            traceback.print_exc()
+            await self.send_text(chat_id, f"❌ 错误: {e}", **ctx)
+        finally:
+            self.user_tasks.pop(chat_id, None)

dingtalkapp.py

@@ -1,8 +1,9 @@
-import os, sys, re, threading, asyncio, queue as Q, socket, time, glob, json
+import asyncio, json, os, sys, threading, time
 import requests
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 from agentmain import GeneraticAgent
+from chatapp_common import AgentChatMixin, ensure_single_instance, public_access, redirect_log, require_runtime, split_text
 from llmcore import mykeys
 
 try:
@@ -12,82 +13,26 @@ except Exception:
     print("Please install dingtalk-stream to use DingTalk: pip install dingtalk-stream")
     sys.exit(1)
 
-agent = GeneraticAgent()
+agent = GeneraticAgent(); agent.verbose = False
-agent.verbose = False
-
 CLIENT_ID = str(mykeys.get("dingtalk_client_id", "") or "").strip()
 CLIENT_SECRET = str(mykeys.get("dingtalk_client_secret", "") or "").strip()
 ALLOWED = {str(x).strip() for x in mykeys.get("dingtalk_allowed_users", []) if str(x).strip()}
-
+USER_TASKS = {}
-_TAG_PATS = [r"<" + t + r">.*?</" + t + r">" for t in ("thinking", "summary", "tool_use", "file_content")]
-_USER_TASKS = {}
 
 
-def _clean(text):
+class DingTalkApp(AgentChatMixin):
-    for pat in _TAG_PATS:
+    label, source, split_limit = "DingTalk", "dingtalk", 1800
-        text = re.sub(pat, "", text, flags=re.DOTALL)
-    return re.sub(r"\n{3,}", "\n\n", text).strip() or "..."
 
-
-def _extract_files(text):
-    return re.findall(r"\[FILE:([^\]]+)\]", text or "")
-
-
-def _strip_files(text):
-    return re.sub(r"\[FILE:[^\]]+\]", "", text or "").strip()
-
-
-def _split_text(text, limit=1800):
-    text = (text or "").strip() or "..."
-    parts = []
-    while len(text) > limit:
-        cut = text.rfind("\n", 0, limit)
-        if cut < limit * 0.6:
-            cut = limit
-        parts.append(text[:cut].rstrip())
-        text = text[cut:].lstrip()
-    if text:
-        parts.append(text)
-    return parts or ["..."]
-
-
-def _format_restore():
-    files = glob.glob("./temp/model_responses_*.txt")
-    if not files:
-        return None, "❌ 没有找到历史记录"
-    latest = max(files, key=os.path.getmtime)
-    with open(latest, "r", encoding="utf-8") as f:
-        content = f.read()
-    users = re.findall(r"=== USER ===\n(.+?)(?==== |$)", content, re.DOTALL)
-    resps = re.findall(r"=== Response ===.*?\n(.+?)(?==== Prompt|$)", content, re.DOTALL)
-    count, restored = 0, []
-    for u, r in zip(users, resps):
-        u, r = u.strip(), r.strip()[:500]
-        if u and r:
-            restored.extend([f"[USER]: {u}", f"[Agent] {r}"])
-            count += 1
-    if not restored:
-        return None, "❌ 历史记录里没有可恢复内容"
-    return (restored, os.path.basename(latest), count), None
-
-
-class DingTalkApp:
     def __init__(self):
-        self.client = None
+        super().__init__(agent, USER_TASKS)
-        self.access_token = None
+        self.client, self.access_token, self.token_expiry, self.background_tasks = None, None, 0, set()
-        self.token_expiry = 0
-        self.background_tasks = set()
 
     async def _get_access_token(self):
         if self.access_token and time.time() < self.token_expiry:
             return self.access_token
 
         def _fetch():
-            resp = requests.post(
+            resp = requests.post("https://api.dingtalk.com/v1.0/oauth2/accessToken", json={"appKey": CLIENT_ID, "appSecret": CLIENT_SECRET}, timeout=20)
-                "https://api.dingtalk.com/v1.0/oauth2/accessToken",
-                json={"appKey": CLIENT_ID, "appSecret": CLIENT_SECRET},
-                timeout=20,
-            )
             resp.raise_for_status()
             return resp.json()
 
@@ -107,30 +52,17 @@ class DingTalkApp:
         headers = {"x-acs-dingtalk-access-token": token}
         if chat_id.startswith("group:"):
             url = "https://api.dingtalk.com/v1.0/robot/groupMessages/send"
-            payload = {
+            payload = {"robotCode": CLIENT_ID, "openConversationId": chat_id[6:], "msgKey": msg_key, "msgParam": json.dumps(msg_param, ensure_ascii=False)}
-                "robotCode": CLIENT_ID,
-                "openConversationId": chat_id[6:],
-                "msgKey": msg_key,
-                "msgParam": json.dumps(msg_param, ensure_ascii=False),
-            }
         else:
             url = "https://api.dingtalk.com/v1.0/robot/oToMessages/batchSend"
-            payload = {
+            payload = {"robotCode": CLIENT_ID, "userIds": [chat_id], "msgKey": msg_key, "msgParam": json.dumps(msg_param, ensure_ascii=False)}
-                "robotCode": CLIENT_ID,
-                "userIds": [chat_id],
-                "msgKey": msg_key,
-                "msgParam": json.dumps(msg_param, ensure_ascii=False),
-            }
 
         def _post():
             resp = requests.post(url, json=payload, headers=headers, timeout=20)
             body = resp.text
             if resp.status_code != 200:
                 raise RuntimeError(f"HTTP {resp.status_code}: {body[:300]}")
-            try:
+            result = resp.json() if "json" in resp.headers.get("content-type", "") else {}
-                result = resp.json()
-            except Exception:
-                result = {}
             errcode = result.get("errcode")
             if errcode not in (None, 0):
                 raise RuntimeError(f"API errcode={errcode}: {body[:300]}")
@@ -143,119 +75,32 @@ class DingTalkApp:
             return False
 
     async def send_text(self, chat_id, content):
-        for part in _split_text(content):
+        for part in split_text(content, self.split_limit):
             await self._send_batch_message(chat_id, "sampleMarkdown", {"text": part, "title": "Agent Reply"})
 
-    async def send_done(self, chat_id, raw_text):
-        files = [p for p in _extract_files(raw_text) if os.path.exists(p)]
-        body = _strip_files(_clean(raw_text))
-        if files:
-            body = (body + "\n\n" if body else "") + "\n".join([f"生成文件: {p}" for p in files])
-        await self.send_text(chat_id, body or "...")
-
-    async def handle_command(self, chat_id, cmd):
-        parts = (cmd or "").split()
-        op = (parts[0] if parts else "").lower()
-        if op == "/stop":
-            state = _USER_TASKS.get(chat_id)
-            if state:
-                state["running"] = False
-            agent.abort()
-            await self.send_text(chat_id, "⏹️ 正在停止...")
-        elif op == "/status":
-            llm = agent.get_llm_name() if agent.llmclient else "未配置"
-            await self.send_text(chat_id, f"状态: {'🔴 运行中' if agent.is_running else '🟢 空闲'}\nLLM: [{agent.llm_no}] {llm}")
-        elif op == "/llm":
-            if not agent.llmclient:
-                return await self.send_text(chat_id, "❌ 当前没有可用的 LLM 配置")
-            if len(parts) > 1:
-                try:
-                    n = int(parts[1])
-                    agent.next_llm(n)
-                    await self.send_text(chat_id, f"✅ 已切换到 [{agent.llm_no}] {agent.get_llm_name()}")
-                except Exception:
-                    await self.send_text(chat_id, f"用法: /llm <0-{len(agent.list_llms()) - 1}>")
-            else:
-                lines = [f"{'→' if cur else '  '} [{i}] {name}" for i, name, cur in agent.list_llms()]
-                await self.send_text(chat_id, "LLMs:\n" + "\n".join(lines))
-        elif op == "/restore":
-            try:
-                restored_info, err = _format_restore()
-                if err:
-                    return await self.send_text(chat_id, err)
-                restored, fname, count = restored_info
-                agent.abort()
-                agent.history.extend(restored)
-                await self.send_text(chat_id, f"✅ 已恢复 {count} 轮对话\n来源: {fname}\n(仅恢复上下文，请输入新问题继续)")
-            except Exception as e:
-                await self.send_text(chat_id, f"❌ 恢复失败: {e}")
-        elif op == "/new":
-            agent.abort()
-            agent.history = []
-            await self.send_text(chat_id, "🆕 已清空当前共享上下文")
-        else:
-            await self.send_text(
-                chat_id,
-                "📖 命令列表:\n/help - 显示帮助\n/status - 查看状态\n/stop - 停止当前任务\n/new - 清空当前上下文\n/restore - 恢复上次对话历史\n/llm [n] - 查看或切换模型",
-            )
-
-    async def run_agent(self, chat_id, text):
-        state = {"running": True}
-        _USER_TASKS[chat_id] = state
-        try:
-            await self.send_text(chat_id, "思考中...")
-            prompt = f"If you need to show files to user, use [FILE:filepath] in your response.\n\n{text}"
-            dq = agent.put_task(prompt, source="dingtalk")
-            last_ping = time.time()
-            while state["running"]:
-                try:
-                    item = await asyncio.to_thread(dq.get, True, 3)
-                except Q.Empty:
-                    if agent.is_running and time.time() - last_ping > 20:
-                        await self.send_text(chat_id, "⏳ 还在处理中，请稍等...")
-                        last_ping = time.time()
-                    continue
-                if "done" in item:
-                    await self.send_done(chat_id, item.get("done", ""))
-                    break
-            if not state["running"]:
-                await self.send_text(chat_id, "⏹️ 已停止")
-        except Exception as e:
-            import traceback
-
-            print(f"[DingTalk] run_agent error: {e}")
-            traceback.print_exc()
-            await self.send_text(chat_id, f"❌ 错误: {e}")
-        finally:
-            _USER_TASKS.pop(chat_id, None)
-
     async def on_message(self, content, sender_id, sender_name, conversation_type=None, conversation_id=None):
         try:
             if not content:
                 return
-            public_access = not ALLOWED or "*" in ALLOWED
+            if not public_access(ALLOWED) and sender_id not in ALLOWED:
-            if not public_access and sender_id not in ALLOWED:
                 print(f"[DingTalk] unauthorized user: {sender_id}")
                 return
             is_group = conversation_type == "2" and conversation_id
             chat_id = f"group:{conversation_id}" if is_group else sender_id
             print(f"[DingTalk] message from {sender_name} ({sender_id}): {content}")
             if content.startswith("/"):
-                await self.handle_command(chat_id, content)
+                return await self.handle_command(chat_id, content)
-                return
             task = asyncio.create_task(self.run_agent(chat_id, content))
             self.background_tasks.add(task)
             task.add_done_callback(self.background_tasks.discard)
         except Exception:
             import traceback
-
             print("[DingTalk] handle_message error")
             traceback.print_exc()
 
     async def start(self):
-        handler = _DingTalkHandler(self)
         self.client = DingTalkStreamClient(Credential(CLIENT_ID, CLIENT_SECRET))
-        self.client.register_callback_handler(ChatbotMessage.TOPIC, handler)
+        self.client.register_callback_handler(ChatbotMessage.TOPIC, _DingTalkHandler(self))
         print("[DingTalk] bot starting...")
         while True:
             try:
@@ -274,44 +119,22 @@ class _DingTalkHandler(CallbackHandler):
     async def process(self, message):
         try:
             chatbot_msg = ChatbotMessage.from_dict(message.data)
-            text = ""
+            text = getattr(getattr(chatbot_msg, "text", None), "content", "") or ""
-            if getattr(getattr(chatbot_msg, "text", None), "content", None):
-                text = chatbot_msg.text.content.strip()
             extensions = getattr(chatbot_msg, "extensions", None) or {}
             recognition = ((extensions.get("content") or {}).get("recognition") or "").strip() if isinstance(extensions, dict) else ""
-            if not text:
+            if not (text := text.strip()):
                 text = recognition or str((message.data.get("text", {}) or {}).get("content", "") or "").strip()
-            sender_id = getattr(chatbot_msg, "sender_staff_id", None) or getattr(chatbot_msg, "sender_id", None) or "unknown"
+            sender_id = str(getattr(chatbot_msg, "sender_staff_id", None) or getattr(chatbot_msg, "sender_id", None) or "unknown")
             sender_name = getattr(chatbot_msg, "sender_nick", None) or "Unknown"
-            conversation_type = message.data.get("conversationType")
+            await self.app.on_message(text, sender_id, sender_name, message.data.get("conversationType"), message.data.get("conversationId") or message.data.get("openConversationId"))
-            conversation_id = message.data.get("conversationId") or message.data.get("openConversationId")
-            await self.app.on_message(text, str(sender_id), sender_name, conversation_type, conversation_id)
-            return AckMessage.STATUS_OK, "OK"
         except Exception as e:
             print(f"[DingTalk] callback error: {e}")
-            return AckMessage.STATUS_OK, "Error"
+        return AckMessage.STATUS_OK, "OK"
 
 
 if __name__ == "__main__":
-    try:
+    _LOCK_SOCK = ensure_single_instance(19530, "DingTalk")
-        _lock_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    require_runtime(agent, "DingTalk", dingtalk_client_id=CLIENT_ID, dingtalk_client_secret=CLIENT_SECRET)
-        _lock_sock.bind(("127.0.0.1", 19530))
+    redirect_log(__file__, "dingtalkapp.log", "DingTalk", ALLOWED)
-    except OSError:
-        print("[DingTalk] Another instance is already running, skipping...")
-        sys.exit(1)
-
-    if not CLIENT_ID or not CLIENT_SECRET:
-        print("[DingTalk] ERROR: please set dingtalk_client_id and dingtalk_client_secret in mykey.py or mykey.json")
-        sys.exit(1)
-    if agent.llmclient is None:
-        print("[DingTalk] ERROR: no usable LLM backend found in mykey.py or mykey.json")
-        sys.exit(1)
-
-    log_dir = os.path.join(os.path.dirname(__file__), "temp")
-    os.makedirs(log_dir, exist_ok=True)
-    _logf = open(os.path.join(log_dir, "dingtalkapp.log"), "a", encoding="utf-8", buffering=1)
-    sys.stdout = sys.stderr = _logf
-    print("[NEW] DingTalk process starting, the above are history infos ...")
-    print(f"[DingTalk] allow list: {'public' if not ALLOWED or '*' in ALLOWED else sorted(ALLOWED)}")
     threading.Thread(target=agent.run, daemon=True).start()
     asyncio.run(DingTalkApp().start())

qqapp.py

@@ -1,8 +1,9 @@
-import os, sys, re, threading, asyncio, queue as Q, socket, time, glob
+import asyncio, os, sys, threading, time
 from collections import deque
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 from agentmain import GeneraticAgent
+from chatapp_common import AgentChatMixin, ensure_single_instance, public_access, redirect_log, require_runtime, split_text
 from llmcore import mykeys
 
 try:
@@ -12,73 +13,19 @@ except Exception:
     print("Please install qq-botpy to use QQ module: pip install qq-botpy")
     sys.exit(1)
 
-agent = GeneraticAgent()
+agent = GeneraticAgent(); agent.verbose = False
-agent.verbose = False
-
 APP_ID = str(mykeys.get("qq_app_id", "") or "").strip()
 APP_SECRET = str(mykeys.get("qq_app_secret", "") or "").strip()
 ALLOWED = {str(x).strip() for x in mykeys.get("qq_allowed_users", []) if str(x).strip()}
-
+PROCESSED_IDS, USER_TASKS = deque(maxlen=1000), {}
-_TAG_PATS = [r"<" + t + r">.*?</" + t + r">" for t in ("thinking", "summary", "tool_use", "file_content")]
+SEQ_LOCK, MSG_SEQ = threading.Lock(), 1
-_PROCESSED_IDS = deque(maxlen=1000)
-_USER_TASKS = {}
-_SEQ_LOCK = threading.Lock()
-_MSG_SEQ = 1
 
 
 def _next_msg_seq():
-    global _MSG_SEQ
+    global MSG_SEQ
-    with _SEQ_LOCK:
+    with SEQ_LOCK:
-        _MSG_SEQ += 1
+        MSG_SEQ += 1
-        return _MSG_SEQ
+        return MSG_SEQ
-
-
-def _clean(text):
-    for pat in _TAG_PATS:
-        text = re.sub(pat, "", text, flags=re.DOTALL)
-    return re.sub(r"\n{3,}", "\n\n", text).strip() or "..."
-
-
-def _extract_files(text):
-    return re.findall(r"\[FILE:([^\]]+)\]", text or "")
-
-
-def _strip_files(text):
-    return re.sub(r"\[FILE:[^\]]+\]", "", text or "").strip()
-
-
-def _split_text(text, limit=1500):
-    text = (text or "").strip() or "..."
-    parts = []
-    while len(text) > limit:
-        cut = text.rfind("\n", 0, limit)
-        if cut < limit * 0.6:
-            cut = limit
-        parts.append(text[:cut].rstrip())
-        text = text[cut:].lstrip()
-    if text:
-        parts.append(text)
-    return parts or ["..."]
-
-
-def _format_restore():
-    files = glob.glob("./temp/model_responses_*.txt")
-    if not files:
-        return None, "❌ 没有找到历史记录"
-    latest = max(files, key=os.path.getmtime)
-    with open(latest, "r", encoding="utf-8") as f:
-        content = f.read()
-    users = re.findall(r"=== USER ===\n(.+?)(?==== |$)", content, re.DOTALL)
-    resps = re.findall(r"=== Response ===.*?\n(.+?)(?==== Prompt|$)", content, re.DOTALL)
-    count, restored = 0, []
-    for u, r in zip(users, resps):
-        u, r = u.strip(), r.strip()[:500]
-        if u and r:
-            restored.extend([f"[USER]: {u}", f"[Agent] {r}"])
-            count += 1
-    if not restored:
-        return None, "❌ 历史记录里没有可恢复内容"
-    return (restored, os.path.basename(latest), count), None
 
 
 def _build_intents():
@@ -86,16 +33,7 @@ def _build_intents():
         return botpy.Intents(public_messages=True, direct_message=True)
     except Exception:
         intents = botpy.Intents.none() if hasattr(botpy.Intents, "none") else botpy.Intents()
-        for attr in (
+        for attr in ("public_messages", "public_guild_messages", "direct_message", "direct_messages", "c2c_message", "c2c_messages", "group_at_message", "group_at_messages"):
-            "public_messages",
-            "public_guild_messages",
-            "direct_message",
-            "direct_messages",
-            "c2c_message",
-            "c2c_messages",
-            "group_at_message",
-            "group_at_messages",
-        ):
             if hasattr(intents, attr):
                 try:
                     setattr(intents, attr, True)
@@ -105,15 +43,12 @@ def _build_intents():
 
 
 def _make_bot_class(app):
-    intents = _build_intents()
+    class QQBot(botpy.Client):
-
-    class _QQBot(botpy.Client):
         def __init__(self):
-            super().__init__(intents=intents, ext_handlers=False)
+            super().__init__(intents=_build_intents(), ext_handlers=False)
 
         async def on_ready(self):
-            name = getattr(getattr(self, "robot", None), "name", "QQBot")
+            print(f"[QQ] bot ready: {getattr(getattr(self, 'robot', None), 'name', 'QQBot')}")
-            print(f"[QQ] bot ready: {name}")
 
         async def on_c2c_message_create(self, message: C2CMessage):
             await app.on_message(message, is_group=False)
@@ -124,154 +59,50 @@ def _make_bot_class(app):
         async def on_direct_message_create(self, message):
             await app.on_message(message, is_group=False)
 
-    return _QQBot
+    return QQBot
 
 
-class QQApp:
+class QQApp(AgentChatMixin):
+    label, source, split_limit = "QQ", "qq", 1500
+
     def __init__(self):
+        super().__init__(agent, USER_TASKS)
         self.client = None
 
     async def send_text(self, chat_id, content, *, msg_id=None, is_group=False):
         if not self.client:
             return
-        for part in _split_text(content):
+        api = self.client.api.post_group_message if is_group else self.client.api.post_c2c_message
-            seq = _next_msg_seq()
+        key = "group_openid" if is_group else "openid"
-            if is_group:
+        for part in split_text(content, self.split_limit):
-                await self.client.api.post_group_message(
+            await api(**{key: chat_id, "msg_type": 0, "content": part, "msg_id": msg_id, "msg_seq": _next_msg_seq()})
-                    group_openid=chat_id,
-                    msg_type=0,
-                    content=part,
-                    msg_id=msg_id,
-                    msg_seq=seq,
-                )
-            else:
-                await self.client.api.post_c2c_message(
-                    openid=chat_id,
-                    msg_type=0,
-                    content=part,
-                    msg_id=msg_id,
-                    msg_seq=seq,
-                )
-
-    async def send_done(self, chat_id, raw_text, *, msg_id=None, is_group=False):
-        files = [p for p in _extract_files(raw_text) if os.path.exists(p)]
-        body = _strip_files(_clean(raw_text))
-        if files:
-            body = (body + "\n\n" if body else "") + "\n".join([f"生成文件: {p}" for p in files])
-        await self.send_text(chat_id, body or "...", msg_id=msg_id, is_group=is_group)
-
-    async def handle_command(self, chat_id, cmd, *, msg_id=None, is_group=False):
-        parts = (cmd or "").split()
-        op = (parts[0] if parts else "").lower()
-        if op == "/stop":
-            state = _USER_TASKS.get(chat_id)
-            if state:
-                state["running"] = False
-            agent.abort()
-            await self.send_text(chat_id, "⏹️ 正在停止...", msg_id=msg_id, is_group=is_group)
-        elif op == "/status":
-            llm = agent.get_llm_name() if agent.llmclient else "未配置"
-            await self.send_text(chat_id, f"状态: {'🔴 运行中' if agent.is_running else '🟢 空闲'}\nLLM: [{agent.llm_no}] {llm}", msg_id=msg_id, is_group=is_group)
-        elif op == "/llm":
-            if not agent.llmclient:
-                return await self.send_text(chat_id, "❌ 当前没有可用的 LLM 配置", msg_id=msg_id, is_group=is_group)
-            if len(parts) > 1:
-                try:
-                    n = int(parts[1])
-                    agent.next_llm(n)
-                    await self.send_text(chat_id, f"✅ 已切换到 [{agent.llm_no}] {agent.get_llm_name()}", msg_id=msg_id, is_group=is_group)
-                except Exception:
-                    await self.send_text(chat_id, f"用法: /llm <0-{len(agent.list_llms()) - 1}>", msg_id=msg_id, is_group=is_group)
-            else:
-                lines = [f"{'→' if cur else '  '} [{i}] {name}" for i, name, cur in agent.list_llms()]
-                await self.send_text(chat_id, "LLMs:\n" + "\n".join(lines), msg_id=msg_id, is_group=is_group)
-        elif op == "/restore":
-            try:
-                restored_info, err = _format_restore()
-                if err:
-                    return await self.send_text(chat_id, err, msg_id=msg_id, is_group=is_group)
-                restored, fname, count = restored_info
-                agent.abort()
-                agent.history.extend(restored)
-                await self.send_text(chat_id, f"✅ 已恢复 {count} 轮对话\n来源: {fname}\n(仅恢复上下文，请输入新问题继续)", msg_id=msg_id, is_group=is_group)
-            except Exception as e:
-                await self.send_text(chat_id, f"❌ 恢复失败: {e}", msg_id=msg_id, is_group=is_group)
-        elif op == "/new":
-            agent.abort()
-            agent.history = []
-            await self.send_text(chat_id, "🆕 已清空当前共享上下文", msg_id=msg_id, is_group=is_group)
-        else:
-            await self.send_text(
-                chat_id,
-                "📖 命令列表:\n/help - 显示帮助\n/status - 查看状态\n/stop - 停止当前任务\n/new - 清空当前上下文\n/restore - 恢复上次对话历史\n/llm [n] - 查看或切换模型",
-                msg_id=msg_id,
-                is_group=is_group,
-            )
-
-    async def run_agent(self, chat_id, text, *, msg_id=None, is_group=False):
-        state = {"running": True}
-        _USER_TASKS[chat_id] = state
-        try:
-            await self.send_text(chat_id, "思考中...", msg_id=msg_id, is_group=is_group)
-            prompt = f"If you need to show files to user, use [FILE:filepath] in your response.\n\n{text}"
-            dq = agent.put_task(prompt, source="qq")
-            last_ping = time.time()
-            while state["running"]:
-                try:
-                    item = await asyncio.to_thread(dq.get, True, 3)
-                except Q.Empty:
-                    if agent.is_running and time.time() - last_ping > 20:
-                        await self.send_text(chat_id, "⏳ 还在处理中，请稍等...", msg_id=msg_id, is_group=is_group)
-                        last_ping = time.time()
-                    continue
-                if "done" in item:
-                    await self.send_done(chat_id, item.get("done", ""), msg_id=msg_id, is_group=is_group)
-                    break
-            if not state["running"]:
-                await self.send_text(chat_id, "⏹️ 已停止", msg_id=msg_id, is_group=is_group)
-        except Exception as e:
-            import traceback
-
-            print(f"[QQ] run_agent error: {e}")
-            traceback.print_exc()
-            await self.send_text(chat_id, f"❌ 错误: {e}", msg_id=msg_id, is_group=is_group)
-        finally:
-            _USER_TASKS.pop(chat_id, None)
 
     async def on_message(self, data, is_group=False):
         try:
             msg_id = getattr(data, "id", None)
-            if msg_id in _PROCESSED_IDS:
+            if msg_id in PROCESSED_IDS:
                 return
-            _PROCESSED_IDS.append(msg_id)
+            PROCESSED_IDS.append(msg_id)
             content = (getattr(data, "content", "") or "").strip()
             if not content:
                 return
             author = getattr(data, "author", None)
-            if is_group:
+            user_id = str(getattr(author, "member_openid" if is_group else "user_openid", "") or getattr(author, "id", "") or "unknown")
-                chat_id = str(getattr(data, "group_openid", "") or "")
+            chat_id = str(getattr(data, "group_openid", "") or user_id) if is_group else user_id
-                user_id = str(getattr(author, "member_openid", "") or getattr(author, "id", "") or "unknown")
+            if not public_access(ALLOWED) and user_id not in ALLOWED:
-            else:
-                user_id = str(getattr(author, "user_openid", "") or getattr(author, "id", "") or "unknown")
-                chat_id = user_id
-            public_access = not ALLOWED or "*" in ALLOWED
-            if not public_access and user_id not in ALLOWED:
                 print(f"[QQ] unauthorized user: {user_id}")
                 return
             print(f"[QQ] message from {user_id} ({'group' if is_group else 'c2c'}): {content}")
             if content.startswith("/"):
-                await self.handle_command(chat_id, content, msg_id=msg_id, is_group=is_group)
+                return await self.handle_command(chat_id, content, msg_id=msg_id, is_group=is_group)
-                return
             asyncio.create_task(self.run_agent(chat_id, content, msg_id=msg_id, is_group=is_group))
         except Exception:
             import traceback
-
             print("[QQ] handle_message error")
             traceback.print_exc()
 
     async def start(self):
-        BotClass = _make_bot_class(self)
+        self.client = _make_bot_class(self)()
-        self.client = BotClass()
         while True:
             try:
                 print(f"[QQ] bot starting... {time.strftime('%m-%d %H:%M')}")
@@ -283,25 +114,8 @@ class QQApp:
 
 
 if __name__ == "__main__":
-    try:
+    _LOCK_SOCK = ensure_single_instance(19528, "QQ")
-        _lock_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    require_runtime(agent, "QQ", qq_app_id=APP_ID, qq_app_secret=APP_SECRET)
-        _lock_sock.bind(("127.0.0.1", 19528))
+    redirect_log(__file__, "qqapp.log", "QQ", ALLOWED)
-    except OSError:
-        print("[QQ] Another instance is already running, skipping...")
-        sys.exit(1)
-
-    if not APP_ID or not APP_SECRET:
-        print("[QQ] ERROR: please set qq_app_id and qq_app_secret in mykey.py or mykey.json")
-        sys.exit(1)
-    if agent.llmclient is None:
-        print("[QQ] ERROR: no usable LLM backend found in mykey.py or mykey.json")
-        sys.exit(1)
-
-    log_dir = os.path.join(os.path.dirname(__file__), "temp")
-    os.makedirs(log_dir, exist_ok=True)
-    _logf = open(os.path.join(log_dir, "qqapp.log"), "a", encoding="utf-8", buffering=1)
-    sys.stdout = sys.stderr = _logf
-    print("[NEW] QQ process starting, the above are history infos ...")
-    print(f"[QQ] allow list: {'public' if not ALLOWED or '*' in ALLOWED else sorted(ALLOWED)}")
     threading.Thread(target=agent.run, daemon=True).start()
     asyncio.run(QQApp().start())

wecomapp.py

@@ -1,8 +1,9 @@
-import os, sys, re, threading, asyncio, queue as Q, socket, time, glob
+import asyncio, os, sys, threading
 from collections import deque
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 from agentmain import GeneraticAgent
+from chatapp_common import AgentChatMixin, ensure_single_instance, public_access, redirect_log, require_runtime, split_text
 from llmcore import mykeys
 
 try:
@@ -11,207 +12,60 @@ except Exception:
     print("Please install wecom_aibot_sdk to use WeCom: pip install wecom_aibot_sdk")
     sys.exit(1)
 
-agent = GeneraticAgent()
+agent = GeneraticAgent(); agent.verbose = False
-agent.verbose = False
-
 BOT_ID = str(mykeys.get("wecom_bot_id", "") or "").strip()
 SECRET = str(mykeys.get("wecom_secret", "") or "").strip()
 WELCOME = str(mykeys.get("wecom_welcome_message", "") or "").strip()
 ALLOWED = {str(x).strip() for x in mykeys.get("wecom_allowed_users", []) if str(x).strip()}
-
+PROCESSED_IDS, USER_TASKS = deque(maxlen=1000), {}
-_TAG_PATS = [r"<" + t + r">.*?</" + t + r">" for t in ("thinking", "summary", "tool_use", "file_content")]
-_PROCESSED_IDS = deque(maxlen=1000)
-_USER_TASKS = {}
 
 
-def _clean(text):
+class WeComApp(AgentChatMixin):
-    for pat in _TAG_PATS:
+    label, source, split_limit = "WeCom", "wecom", 1200
-        text = re.sub(pat, "", text, flags=re.DOTALL)
-    return re.sub(r"\n{3,}", "\n\n", text).strip() or "..."
 
-
-def _extract_files(text):
-    return re.findall(r"\[FILE:([^\]]+)\]", text or "")
-
-
-def _strip_files(text):
-    return re.sub(r"\[FILE:[^\]]+\]", "", text or "").strip()
-
-
-def _split_text(text, limit=1200):
-    text = (text or "").strip() or "..."
-    parts = []
-    while len(text) > limit:
-        cut = text.rfind("\n", 0, limit)
-        if cut < limit * 0.6:
-            cut = limit
-        parts.append(text[:cut].rstrip())
-        text = text[cut:].lstrip()
-    if text:
-        parts.append(text)
-    return parts or ["..."]
-
-
-def _format_restore():
-    files = glob.glob("./temp/model_responses_*.txt")
-    if not files:
-        return None, "❌ 没有找到历史记录"
-    latest = max(files, key=os.path.getmtime)
-    with open(latest, "r", encoding="utf-8") as f:
-        content = f.read()
-    users = re.findall(r"=== USER ===\n(.+?)(?==== |$)", content, re.DOTALL)
-    resps = re.findall(r"=== Response ===.*?\n(.+?)(?==== Prompt|$)", content, re.DOTALL)
-    count, restored = 0, []
-    for u, r in zip(users, resps):
-        u, r = u.strip(), r.strip()[:500]
-        if u and r:
-            restored.extend([f"[USER]: {u}", f"[Agent] {r}"])
-            count += 1
-    if not restored:
-        return None, "❌ 历史记录里没有可恢复内容"
-    return (restored, os.path.basename(latest), count), None
-
-
-class WeComApp:
     def __init__(self):
-        self.client = None
+        super().__init__(agent, USER_TASKS)
-        self.chat_frames = {}
+        self.client, self.chat_frames = None, {}
-
-    def _body(self, frame):
-        if hasattr(frame, "body"):
-            return frame.body or {}
-        if isinstance(frame, dict):
-            return frame.get("body", frame)
-        return {}
 
     async def send_text(self, chat_id, content):
-        if not self.client:
+        if not self.client or chat_id not in self.chat_frames:
-            return
+            if chat_id not in self.chat_frames:
-        frame = self.chat_frames.get(chat_id)
-        if not frame:
                 print(f"[WeCom] no frame found for chat: {chat_id}")
             return
-        for part in _split_text(content):
+        frame = self.chat_frames[chat_id]
-            stream_id = generate_req_id("stream")
+        for part in split_text(content, self.split_limit):
-            await self.client.reply_stream(frame, stream_id, part, finish=True)
+            await self.client.reply_stream(frame, generate_req_id("stream"), part, finish=True)
-
-    async def send_done(self, chat_id, raw_text):
-        files = [p for p in _extract_files(raw_text) if os.path.exists(p)]
-        body = _strip_files(_clean(raw_text))
-        if files:
-            body = (body + "\n\n" if body else "") + "\n".join([f"生成文件: {p}" for p in files])
-        await self.send_text(chat_id, body or "...")
-
-    async def handle_command(self, chat_id, cmd):
-        parts = (cmd or "").split()
-        op = (parts[0] if parts else "").lower()
-        if op == "/stop":
-            state = _USER_TASKS.get(chat_id)
-            if state:
-                state["running"] = False
-            agent.abort()
-            await self.send_text(chat_id, "⏹️ 正在停止...")
-        elif op == "/status":
-            llm = agent.get_llm_name() if agent.llmclient else "未配置"
-            await self.send_text(chat_id, f"状态: {'🔴 运行中' if agent.is_running else '🟢 空闲'}\nLLM: [{agent.llm_no}] {llm}")
-        elif op == "/llm":
-            if not agent.llmclient:
-                return await self.send_text(chat_id, "❌ 当前没有可用的 LLM 配置")
-            if len(parts) > 1:
-                try:
-                    n = int(parts[1])
-                    agent.next_llm(n)
-                    await self.send_text(chat_id, f"✅ 已切换到 [{agent.llm_no}] {agent.get_llm_name()}")
-                except Exception:
-                    await self.send_text(chat_id, f"用法: /llm <0-{len(agent.list_llms()) - 1}>")
-            else:
-                lines = [f"{'→' if cur else '  '} [{i}] {name}" for i, name, cur in agent.list_llms()]
-                await self.send_text(chat_id, "LLMs:\n" + "\n".join(lines))
-        elif op == "/restore":
-            try:
-                restored_info, err = _format_restore()
-                if err:
-                    return await self.send_text(chat_id, err)
-                restored, fname, count = restored_info
-                agent.abort()
-                agent.history.extend(restored)
-                await self.send_text(chat_id, f"✅ 已恢复 {count} 轮对话\n来源: {fname}\n(仅恢复上下文，请输入新问题继续)")
-            except Exception as e:
-                await self.send_text(chat_id, f"❌ 恢复失败: {e}")
-        elif op == "/new":
-            agent.abort()
-            agent.history = []
-            await self.send_text(chat_id, "🆕 已清空当前共享上下文")
-        else:
-            await self.send_text(
-                chat_id,
-                "📖 命令列表:\n/help - 显示帮助\n/status - 查看状态\n/stop - 停止当前任务\n/new - 清空当前上下文\n/restore - 恢复上次对话历史\n/llm [n] - 查看或切换模型",
-            )
-
-    async def run_agent(self, chat_id, text):
-        state = {"running": True}
-        _USER_TASKS[chat_id] = state
-        try:
-            await self.send_text(chat_id, "思考中...")
-            prompt = f"If you need to show files to user, use [FILE:filepath] in your response.\n\n{text}"
-            dq = agent.put_task(prompt, source="wecom")
-            last_ping = time.time()
-            while state["running"]:
-                try:
-                    item = await asyncio.to_thread(dq.get, True, 3)
-                except Q.Empty:
-                    if agent.is_running and time.time() - last_ping > 20:
-                        await self.send_text(chat_id, "⏳ 还在处理中，请稍等...")
-                        last_ping = time.time()
-                    continue
-                if "done" in item:
-                    await self.send_done(chat_id, item.get("done", ""))
-                    break
-            if not state["running"]:
-                await self.send_text(chat_id, "⏹️ 已停止")
-        except Exception as e:
-            import traceback
-
-            print(f"[WeCom] run_agent error: {e}")
-            traceback.print_exc()
-            await self.send_text(chat_id, f"❌ 错误: {e}")
-        finally:
-            _USER_TASKS.pop(chat_id, None)
 
     async def on_text(self, frame):
         try:
-            body = self._body(frame)
+            body = frame.body if hasattr(frame, "body") else frame.get("body", frame) if isinstance(frame, dict) else {}
             if not isinstance(body, dict):
                 return
             msg_id = body.get("msgid") or f"{body.get('chatid', '')}_{body.get('sendertime', '')}"
-            if msg_id in _PROCESSED_IDS:
+            if msg_id in PROCESSED_IDS:
                 return
-            _PROCESSED_IDS.append(msg_id)
+            PROCESSED_IDS.append(msg_id)
             from_info = body.get("from", {}) if isinstance(body.get("from", {}), dict) else {}
             sender_id = str(from_info.get("userid", "") or "unknown")
             chat_id = str(body.get("chatid", "") or sender_id)
             content = str((body.get("text", {}) or {}).get("content", "") or "").strip()
             if not content:
                 return
-            public_access = not ALLOWED or "*" in ALLOWED
+            if not public_access(ALLOWED) and sender_id not in ALLOWED:
-            if not public_access and sender_id not in ALLOWED:
                 print(f"[WeCom] unauthorized user: {sender_id}")
                 return
             self.chat_frames[chat_id] = frame
             print(f"[WeCom] message from {sender_id}: {content}")
             if content.startswith("/"):
-                await self.handle_command(chat_id, content)
+                return await self.handle_command(chat_id, content)
-                return
             asyncio.create_task(self.run_agent(chat_id, content))
         except Exception:
             import traceback
-
             print("[WeCom] handle_message error")
             traceback.print_exc()
 
     async def on_enter_chat(self, frame):
-        if not WELCOME or not self.client:
+        if WELCOME and self.client:
-            return
             try:
                 await self.client.reply_welcome(frame, {"msgtype": "text", "text": {"content": WELCOME}})
             except Exception as e:
@@ -230,19 +84,16 @@ class WeComApp:
         print(f"[WeCom] error: {frame}")
 
     async def start(self):
-        self.client = WSClient({
+        self.client = WSClient({"bot_id": BOT_ID, "secret": SECRET, "reconnect_interval": 1000, "max_reconnect_attempts": -1, "heartbeat_interval": 30000})
-            "bot_id": BOT_ID,
+        for event, handler in {
-            "secret": SECRET,
+            "connected": self.on_connected,
-            "reconnect_interval": 1000,
+            "authenticated": self.on_authenticated,
-            "max_reconnect_attempts": -1,
+            "disconnected": self.on_disconnected,
-            "heartbeat_interval": 30000,
+            "error": self.on_error,
-        })
+            "message.text": self.on_text,
-        self.client.on("connected", self.on_connected)
+            "event.enter_chat": self.on_enter_chat,
-        self.client.on("authenticated", self.on_authenticated)
+        }.items():
-        self.client.on("disconnected", self.on_disconnected)
+            self.client.on(event, handler)
-        self.client.on("error", self.on_error)
-        self.client.on("message.text", self.on_text)
-        self.client.on("event.enter_chat", self.on_enter_chat)
         print("[WeCom] bot starting...")
         await self.client.connect_async()
         while True:
@@ -250,25 +101,8 @@ class WeComApp:
 
 
 if __name__ == "__main__":
-    try:
+    _LOCK_SOCK = ensure_single_instance(19529, "WeCom")
-        _lock_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    require_runtime(agent, "WeCom", wecom_bot_id=BOT_ID, wecom_secret=SECRET)
-        _lock_sock.bind(("127.0.0.1", 19529))
+    redirect_log(__file__, "wecomapp.log", "WeCom", ALLOWED)
-    except OSError:
-        print("[WeCom] Another instance is already running, skipping...")
-        sys.exit(1)
-
-    if not BOT_ID or not SECRET:
-        print("[WeCom] ERROR: please set wecom_bot_id and wecom_secret in mykey.py or mykey.json")
-        sys.exit(1)
-    if agent.llmclient is None:
-        print("[WeCom] ERROR: no usable LLM backend found in mykey.py or mykey.json")
-        sys.exit(1)
-
-    log_dir = os.path.join(os.path.dirname(__file__), "temp")
-    os.makedirs(log_dir, exist_ok=True)
-    _logf = open(os.path.join(log_dir, "wecomapp.log"), "a", encoding="utf-8", buffering=1)
-    sys.stdout = sys.stderr = _logf
-    print("[NEW] WeCom process starting, the above are history infos ...")
-    print(f"[WeCom] allow list: {'public' if not ALLOWED or '*' in ALLOWED else sorted(ALLOWED)}")
     threading.Thread(target=agent.run, daemon=True).start()
     asyncio.run(WeComApp().start())