# -*- coding: utf-8 -*- import asyncio import threading import time import tomllib import traceback import uuid from collections import deque import toml from loguru import logger import wechat_ipad from base.plugin_common.message_plugin_interface import MessagePluginInterface from base.plugin_common.plugin_interface import PluginStatus from base.plugin_common.plugin_manager import PluginManager from base.plugin_common.plugin_registry import PluginRegistry from configuration import Config from db.connection import DBConnectionManager from db.contacts_db import ContactsDBOperator from db.group_plugin_config_db import GroupPluginConfigDBOperator from db.llm_catalog_db import LLMCatalogDBOperator from db.plugin_schedule_db import PluginScheduleDBOperator from db.system_job_db import SystemJobDBOperator from utils.system_jobs import SystemJobLoader from utils.email_util import EmailSender from utils.group_plugin_config_service import GroupPluginConfigService from utils.plugin_schedule_manager import PluginScheduleManager from utils.revoke.message_auto_revoke import MessageAutoRevoke from utils.robot_cmd.robot_command import GroupBotManager, Feature, PermissionStatus from utils.wechat.contact_manager import ContactManager from utils.wechat.member_monitor import ChatroomMemberMonitor from utils.wechat.message_to_db import MessageStorage from utils.ai.llm_registry import LLMRegistry from utils.trace_context import set_current_trace_id, reset_current_trace_id from wechat_ipad import WechatAPIClient from wechat_ipad.models.message import WxMessage, MessageType # 定义全局信号量，限制最大并发 10 sem = asyncio.Semaphore(20) class Robot: """个性化自己的机器人 """ def __init__(self, config: Config) -> None: super().__init__() self.config = config self.LOG = logger self.LOG.info(f"=" * 50) # wechat_ipad 相关属性 self.ipad_bot: WechatAPIClient self.ipad_config = None self.ipad_running = False self.ipad_thread = None self.ipad_loop = None self.wxid = None self.nickname = None self.alias = None self.phone = None self.message_auto_revoke: MessageAutoRevoke = None self.LOG.debug(f"DB+REDIS 连接池开始初始化") # 使用单例模式获取实例 self.db_manager = DBConnectionManager.get_instance( mysql_config=self.config.mariadb, redis_config=self.config.redis ) self.LOG.debug(f"数据库连接管理器初始化完成") # 为了兼容现有代码，保留原有的连接池 self.db_pool = self.db_manager.mysql_pool self.redis_pool = self.db_manager.redis_pool self.contacts_db = ContactsDBOperator(self.db_manager) self.group_plugin_config_db = GroupPluginConfigDBOperator(self.db_manager) self.llm_catalog_db = LLMCatalogDBOperator(self.db_manager) self.plugin_schedule_db = PluginScheduleDBOperator(self.db_manager) self.system_job_db = SystemJobDBOperator(self.db_manager) self.group_plugin_config_db.init_tables() # 新版 LLM 目录模型（Provider 模板 / Dify 应用 / Scene）初始化。 self.llm_catalog_db.init_tables() self.llm_catalog_db.bootstrap_from_legacy_llm(self.config.llm) self.group_plugin_config_service = GroupPluginConfigService( db_operator=self.group_plugin_config_db, redis_client=self.db_manager.get_redis_connection(), ) # 初始化联系人管理器 self.contact_manager = ContactManager.get_instance() self.allContacts = {} # 将在登录后填充 # 提前初始化消息存储： # 1. DashboardServer 会在主线程里较早启动，并直接读取 robot.message_storage； # 2. 旧逻辑要等 iPad 登录成功后才赋值，导致后台在启动竞态下拿不到这个属性； # 3. 这里先给一个可用的默认实例，后续登录成功后再注入真实 bot 覆盖即可。 self.message_storage = MessageStorage() self.groups = {} # 存储按group_id分组的消息列表，每个group_id最多保留10条消息 # 初始化插件系统 self.LOG.debug("开始初始化插件系统...") self.plugin_registry = PluginRegistry() self.plugin_modules = {} # 存储已加载的插件模块 self.plugins = {} # 存储已加载的插件实例 # 设置插件系统上下文 self.system_context = { "config": config, "plugin_registry": self.plugin_registry, "db_manager": self.db_manager, "db_pool": self.db_pool, "redis_pool": self.redis_pool, "group_plugin_config_service": self.group_plugin_config_service, } self.plugin_manager = PluginManager(plugin_dir=getattr(self.config, "plugin_dir", "plugins")) self.plugin_manager.set_system_context(self.system_context) self.plugins = self.plugin_manager.load_all_plugins() # 热加载改为低频扫描：每 60 秒检查一次插件文件变动 self.plugin_manager.start_hot_reload_watcher(interval_seconds=60.0) self.system_job_loader = SystemJobLoader(self, self.system_job_db) self.system_job_loader.init_and_load() self.plugin_schedule_manager = PluginScheduleManager(self.plugin_manager, self.plugin_schedule_db) self.plugin_schedule_manager.init_and_load() # 将历史业务型系统任务迁移到插件调度配置，避免升级后出现“任务丢失”。 migration_result = self.plugin_schedule_manager.migrate_from_system_jobs(self.system_job_db) if migration_result.get("migrated", 0) > 0: self.LOG.info(f"系统任务迁移到插件任务完成: {migration_result}") self.plugin_schedule_manager.reload_from_db() # 迁移完成后，清理已下沉到插件层的系统任务，避免后台重复维护两套配置。 self._cleanup_migrated_system_jobs() # 加载插件 self.LOG.debug("插件系统初始化完成") GroupBotManager.load_local_cache() # 权限模块加载 self.gbm = GroupBotManager() self.email_sender = EmailSender( smtp_server=self.config.email.get("smtp_server", "smtp.163.com"), smtp_port=self.config.email.get("smtp_port", 465), sender_email=self.config.email.get("sender_email", "bovine_liu@163.com"), sender_password=self.config.email.get("sender_password", "LTS9BhmX9XhS36QS") ) # 通过类属性设置 admin_list，而不是实例属性 GroupBotManager.admin_list = self.config.wx_config.get("admin", []) self.recent_msg_ids = deque(maxlen=20) def apply_runtime_config(self, reload_catalog: bool = False) -> None: """把最新全局配置应用到当前运行中的关键对象。 说明： 1. `self.config.reload()` 只会刷新 Config 实例中的字段，不会自动更新启动时已构造的依赖对象； 2. 这里集中处理“保存配置后希望立刻生效”的轻量刷新动作，避免为大多数改动走整进程重启； 3. 该方法刻意不去重建 DB 连接、微信登录态、插件实例，尽量把影响范围控制在可热刷的配置项。 """ # 邮件发送器在初始化时会拷贝 SMTP 参数，因此这里需要按最新配置重建一份实例。 self.email_sender = EmailSender( smtp_server=self.config.email.get("smtp_server", "smtp.163.com"), smtp_port=self.config.email.get("smtp_port", 465), sender_email=self.config.email.get("sender_email", "bovine_liu@163.com"), sender_password=self.config.email.get("sender_password", "LTS9BhmX9XhS36QS") ) # 管理员列表走 GroupBotManager 的类级缓存；只 reload Config 不会自动回写到这里。 GroupBotManager.admin_list = self.config.wx_config.get("admin", []) # system_context 中保存的是 config 对象引用，reload 后插件读取到的是最新字段。 # 但 LLMRegistry 自己还有一层短 TTL 缓存，因此保存全局 LLM 配置后需要显式清掉。 if reload_catalog: self.llm_catalog_db.bootstrap_from_legacy_llm(self.config.llm) LLMRegistry.invalidate_cache() self.LOG.info( "运行时配置已应用: " f"admin_count={len(GroupBotManager.admin_list)}, " f"email_sender={'ready' if self.email_sender else 'missing'}, " f"llm_cache_reloaded={reload_catalog}" ) def _cleanup_migrated_system_jobs(self): """清理已经迁移到插件层的历史系统任务键。""" migrated_keys = [ "news_baidu_report_auto", "epic_free_games", "message_ranking_push", "sehuatang_pdf_push", "xiuren_download", "shenshi_r15_download", "update_image_cache", # 联系人头像缓存任务在 2026-05-06 调整过命名： # 1. 旧键 `sync_contact_avatar_cache` 只存在于历史数据库配置； # 2. 新键统一使用 `contact_avatar_cache_sync`，避免命名风格前后不一致； # 3. 这里在启动期顺手清理旧键，避免后台任务页长期出现“数据库有记录、运行态无处理器”的幽灵任务。 "sync_contact_avatar_cache", ] removed = 0 for job_key in migrated_keys: try: row = self.system_job_db.get_job(job_key) if not row: continue if self.system_job_db.delete_job(job_key): removed += 1 except Exception as e: self.LOG.warning(f"清理迁移系统任务失败: job_key={job_key}, error={e}") if removed > 0: self.LOG.info(f"已清理 {removed} 个历史系统任务配置（迁移至插件任务）") def init_wechat_ipad(self): """初始化wechat_ipad客户端""" try: # 读取config.toml文件 with open("wechat_ipad/config.toml", "rb") as f: self.ipad_config = tomllib.load(f) self.LOG.debug("正在初始化wechat_ipad客户端...") # 检查必要的配置 server_url = self.ipad_config.get("server_url", "") if server_url == "": self.LOG.error("server_url不能为空，wechat_ipad初始化失败") return False server_ip = self.ipad_config.get("server_ip", "") server_port = self.ipad_config.get("server_port", 8059) # 创建事件循环 self.ipad_loop = asyncio.new_event_loop() # 在新线程中启动wechat_ipad客户端 self.ipad_thread = threading.Thread( target=self._run_wechat_ipad_client, args=(server_ip, server_port), daemon=True ) self.ipad_thread.start() self.LOG.debug("wechat_ipad客户端初始化完成") return True except Exception as e: self.LOG.error(f"初始化wechat_ipad客户端失败: {e}") return False def _run_wechat_ipad_client(self, server_ip, server_port): """在新线程中运行wechat_ipad客户端""" asyncio.set_event_loop(self.ipad_loop) self.ipad_loop.run_until_complete(self._wechat_ipad_core(server_ip, server_port)) async def _wechat_ipad_core(self, server_ip, server_port): """wechat_ipad核心逻辑，基于bot-core.py""" try: self.LOG.debug("启动wechat_ipad bot") # 调用登录接口 self.ipad_bot = wechat_ipad.WechatAPIClient(server_ip, server_port) self.message_auto_revoke = MessageAutoRevoke(self.ipad_bot) wxid = self.ipad_config.get("wxid", "") device_name = self.ipad_config.get("device_name", "") device_id = self.ipad_config.get("device_id", "") if device_name == "": device_name = self.ipad_bot.create_device_name() if device_id == "": device_id = self.ipad_bot.create_device_id() # 登录逻辑 if not await self.ipad_bot.is_logged_in(wxid): await self._handle_ipad_login(wxid, device_name, device_id) else: # 已登录 self.ipad_bot.wxid = wxid profile = await self.ipad_bot.get_profile() self.ipad_bot.nickname = profile.get("NickName").get("string") self.ipad_bot.alias = profile.get("Alias") self.ipad_bot.phone = profile.get("BindMobile").get("string") self.ipad_bot.signature = profile.get("Signature", "") # 更新Robot类的属性 self.wxid = self.ipad_bot.wxid self.nickname = self.ipad_bot.nickname self.alias = self.ipad_bot.alias self.phone = self.ipad_bot.phone self.signature = self.ipad_bot.signature self.LOG.info( f"wechat_ipad登录账号信息: wxid: {self.wxid} 昵称: {self.nickname} 微信号: {self.alias} 手机号: {self.phone}") # 注入加载完成的bot self.plugin_manager.inject_bot(self.ipad_bot) self.LOG.info(f"wechat_ipad登录设备信息: device_name: {device_name} device_id: {device_id}") self.LOG.info("wechat_ipad登录成功") # 登录成功后加载联系人信息 self.allContacts = self.get_all_contacts() friends = await self.ipad_bot.get_contract_list() self.head_images = self.get_all_head_images() self.all_chatroom_members = self.contacts_db.get_chatroom_member_list_name_all() # self.LOG.debug(f"all_chatroom_members:{self.all_chatroom_members}") self.contact_manager.set_contacts(self.allContacts, friends, self.head_images, self.all_chatroom_members) self.message_storage = MessageStorage(self.ipad_bot) self.member_monitor = ChatroomMemberMonitor(self.ipad_bot) # # 获取扩展信息，显示相关内容 ext_profile = await self.ipad_bot.get_profile_info_ext() self.ipad_bot.profile_ext = ext_profile self.head_image = ext_profile.get("SmallHeadImgUrl") # 先接受堆积消息 self.LOG.info("处理堆积消息中") # await self.ipad_bot.send_text_message("filehelper", "ipad客户端启动成功") count = 0 while True: data = await self.ipad_bot.sync_message() data = data.get("AddMsgs") if not data: if count > 2: break else: count += 1 continue self.LOG.debug(f"接受到 {len(data)} 条消息") await asyncio.sleep(1) self.LOG.info("处理堆积消息完毕") # 标记为运行中 self.ipad_running = True # 开启自动心跳（作为后台任务） heartbeat_task = asyncio.create_task(self._heartbeat_task()) heartbeat_task_long = asyncio.create_task(self._heartbeat_task_long()) # 开始处理消息 self.LOG.info("开始处理wechat_ipad消息") while self.ipad_running: try: data_temp = await self.ipad_bot.sync_message() except Exception as e: self.LOG.error(f"获取新消息失败 {e}") if "用户可能退出" in str(e): self.LOG.error(f"用户可能退出: {e}") self.email_sender.send_wechat_alert(self.config.email.get("alert_recipient"), f"用户可能退出: {e}", self.wxid, self.nickname) await self.login_twice_auto_auth() await asyncio.sleep(5) continue data = data_temp.get("AddMsgs") if data: for message in data: # self.LOG.debug(f"sync_message.处理消息消息内容: {message}") # 处理消息 try: wxmsg: WxMessage = WxMessage.from_json(message) self._attach_trace_id(wxmsg) # 判断是否已经收到过。处理。存储最近20个msg_id，处理之前判断是否在清单里面，如果在，这不重新处理了。 msg_id = wxmsg.msg_id if msg_id in self.recent_msg_ids: self.LOG.info(self._trace_message(wxmsg, f"出现重复ID消息: {msg_id}")) continue # 已处理，跳过 self.recent_msg_ids.append(msg_id) self.LOG.debug( self._trace_message( wxmsg, f"收到消息 type={getattr(wxmsg.msg_type, 'name', wxmsg.msg_type)} " f"sender={wxmsg.sender} room={wxmsg.roomid or '-'}" ) ) except Exception as e: self.LOG.error(f"WxMessage.from_json 解析失败，消息内容: {message}，错误: {e}") continue # 跳过本条消息，继续处理下一条 # 创建独立任务，不阻塞下一条消息 # 并发执行，限制最大并发数 xx = asyncio.create_task(self._process_with_semaphore(wxmsg)) else: # 只有当 Ret 不等于 0 或者 不包含 KeyBuf 时才打印 if not (isinstance(data_temp, dict) and data_temp.get("Ret") == 0 and "KeyBuf" in data_temp): self.LOG.debug(f"MESSAGE：{data_temp}") changed_groups = self.member_monitor.parse_mod_contacts_msg(data_temp) if changed_groups: self.LOG.info(f"监测到群成员变动消息，涉及群: {changed_groups}") for group_id in changed_groups: if self.gbm.get_group_permission(group_id, Feature.GROUP_MEMBER_CHANGE) == PermissionStatus.ENABLED: xx = asyncio.create_task(self.member_monitor.check_and_handle_changes(group_id)) # 使用异步睡眠替代忙等待循环 await asyncio.sleep(2) except Exception as e: self.LOG.exception(f"wechat_ipad客户端运行出错: {e}") self.ipad_running = False # 在类里直接写一个内联 async 方法（不额外抽取新的对外方法） async def _process_with_semaphore(self, wxmsg): async with sem: # 进入单条消息处理前，把 trace_id 放入当前异步上下文： # 1. 后续插件中的 AI 调用、消息发送、子协程都可以自动继承这个 trace_id； # 2. 这样不需要给大量现有方法额外加 trace_id 参数，侵入性更小； # 3. finally 中会回滚 token，避免把当前消息的 trace_id 泄漏到下一条消息。 trace_token = set_current_trace_id(self._get_trace_id(wxmsg)) try: await self._process_ipad_message(wxmsg) except Exception as e: self.LOG.error(self._trace_message(wxmsg, f"处理消息失败 msg_id={wxmsg.msg_id}, 错误: {e}")) finally: reset_current_trace_id(trace_token) async def _handle_ipad_login(self, wxid, device_name, device_id): """处理wechat_ipad登录""" while not await self.ipad_bot.is_logged_in(wxid): # 需要登录 try: if await self.ipad_bot.get_cached_info(wxid): # 尝试唤醒登录 uuid = await self.ipad_bot.awaken_login(wxid) self.LOG.info(f"获取到登录uuid: {uuid}") else: # 二维码登录 if not device_name: device_name = self.ipad_bot.create_device_name() if not device_id: device_id = self.ipad_bot.create_device_id() uuid, url = await self.ipad_bot.get_qr_code(device_id=device_id, device_name=device_name, print_qr=True) self.LOG.info(f"获取到登录uuid: {uuid}") self.LOG.info(f"获取到登录二维码: {url}") except Exception as e: self.LOG.error(f"登录过程出错: {e}") # 二维码登录 if not device_name: device_name = self.ipad_bot.create_device_name() if not device_id: device_id = self.ipad_bot.create_device_id() uuid, url = await self.ipad_bot.get_qr_code(device_id=device_id, device_name=device_name, print_qr=True) self.LOG.info(f"获取到登录uuid: {uuid}") self.LOG.info(f"获取到登录二维码: {url}") while True: self.LOG.info(f"uuid: {uuid}, url: {url}") stat, data = await self.ipad_bot.check_login_uuid(uuid, device_id=device_id) if stat: break self.LOG.info(f"等待登录中，过期倒计时：{data}") await asyncio.sleep(5) # 保存登录信息 self.ipad_config["wxid"] = self.ipad_bot.wxid self.ipad_config["device_name"] = device_name self.ipad_config["device_id"] = device_id self.ipad_config["login_time"] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) with open("wechat_ipad/config.toml", "w", encoding="utf-8") as f: toml.dump(self.ipad_config, f) # 获取登录账号信息 self.ipad_bot.wxid = data.get("acctSectResp").get("userName") self.ipad_bot.nickname = data.get("acctSectResp").get("nickName") self.ipad_bot.alias = data.get("acctSectResp").get("alias") self.ipad_bot.phone = data.get("acctSectResp").get("bindMobile") self.ipad_bot.signature = data.get("Signature", "") # 更新Robot类的属性 self.wxid = self.ipad_bot.wxid self.nickname = self.ipad_bot.nickname self.alias = self.ipad_bot.alias self.phone = self.ipad_bot.phone self.signature = self.ipad_bot.signature self.LOG.info( f"wechat_ipad登录账号信息: wxid: {self.wxid} 昵称: {self.nickname} 微信号: {self.alias} 手机号: {self.phone}") break async def _heartbeat_task(self): """wechat_ipad心跳任务""" self.LOG.info("开启wechat_ipad心跳！") while self.ipad_running: try: success = await self.ipad_bot.heartbeat() if success: self.LOG.debug("心跳进行中") else: self.LOG.warning("心跳失败") except Exception as e: self.LOG.error(f"wechat_ipad heartbeat: {e}") if "用户可能退出" in str(e): self.LOG.error(f"用户可能退出: {e}") self.email_sender.send_wechat_alert(self.config.email.get("alert_recipient"), f"用户可能退出: {e}", self.wxid, self.nickname) await self.login_twice_auto_auth() await asyncio.sleep(60) async def _heartbeat_task_long(self): """wechat_ipad心跳任务""" self.LOG.info("开启wechat_ipad长连接心跳！") while self.ipad_running: try: success = await self.ipad_bot.heartbeat_long() if success: self.LOG.debug("长连接心跳进行中") else: self.LOG.warning("长连接心跳失败") except Exception as e: self.LOG.error(f"wechat_ipad heartbeat long: {e}") if "用户可能退出" in str(e): self.LOG.error(f"用户可能退出: {e}") self.email_sender.send_wechat_alert(self.config.email.get("alert_recipient"), f"用户可能退出: {e}", self.wxid, self.nickname) await self.login_twice_auto_auth() await asyncio.sleep(120) async def _process_ipad_message(self, message: WxMessage): """处理wechat_ipad消息""" try: # self.LOG.debug(f"message: {message}") # 消息已经是WxMessage对象，直接使用其属性和方法 # 判断是否为群消息 is_group = message.from_group() group_id = message.roomid # 检测群聊是否已加入机器人管理 if is_group and group_id not in GroupBotManager.local_cache["group_list"]: self.LOG.info(f"检测到新群聊: {group_id}，自动添加到机器人管理列表并启用机器人功能") # 添加群组到列表 GroupBotManager.local_cache["group_list"].add(group_id) # 保存到Redis redis_conn = self.db_manager.get_redis_connection() redis_conn.sadd("group:list", group_id) # 设置ROBOT功能为启用状态 GroupBotManager.set_group_permission(group_id, Feature.ROBOT, PermissionStatus.ENABLED) # 获取群成员信息并更新数据库 try: chatroom_info = await self.ipad_bot.get_chatroom_info(group_id) self.LOG.debug(f"获取到群信息: {chatroom_info}") self.allContacts[group_id] = chatroom_info.get('NickName').get("string", "未知群名") if chatroom_info: # 保存群信息到数据库 self.contacts_db.save_chatroom_info(chatroom_info) members = await self.ipad_bot.get_chatroom_member_list(group_id) # 保存群成员信息 if members: # 兼容逻辑已放到save_chatroom_member_simple内部 self.contacts_db.save_chatroom_member_simple(group_id, members) self.LOG.info(f"member_list: {members}") # 更新联系人缓存 for member in members: wxid = member.get("UserName", "") nick_name = member.get("NickName", "") displayName = member.get("DisplayName", "") small_head_img_url = member.get("SmallHeadImgUrl", "") # 如果displayName不为空，使用displayName if displayName: nick_name = displayName if wxid: self.allContacts[wxid] = nick_name self.head_images[wxid] = small_head_img_url friends = await self.ipad_bot.get_contract_list() self.all_chatroom_members = self.contacts_db.get_chatroom_member_list_name_all() self.contact_manager.set_contacts(self.allContacts, friends, self.head_images, self.all_chatroom_members) self.LOG.info(f"已更新群 {group_id} 的成员信息") except Exception as e: self.LOG.error(f"获取群成员信息失败: {e}") # 尝试使用插件处理消息 await self.process_plugin_message(message) if is_group: self.LOG.debug(f"入库和记录群消息: {message}") # 调用统计逻辑进行聊天数据统计： try: if message.sender != self.wxid: self.message_storage.process_message(message) except Exception as e: self.LOG.error(self._trace_message(message, f"process_message error: {e}")) # # 聊天记录入库动作： try: self.message_storage.archive_message(message) # 单独处理图片消息 后续写定时任务自动完成下载。延时处理。 if message.msg_type == MessageType.IMAGE: # 图片消息类型 self.message_storage.process_image(message) except Exception as e: self.LOG.error(self._trace_message(message, f"archive_message error: {e}")) except Exception as e: self.LOG.error(self._trace_message(message, f"处理wechat_ipad消息出错: {e}")) def stop_wechat_ipad(self): """停止wechat_ipad客户端""" self.ipad_running = False if self.ipad_loop: self.ipad_loop.stop() self.LOG.info("wechat_ipad客户端已停止") def keep_running_and_block_process(self) -> None: """ 保持机器人运行，不让进程退出 """ while True: time.sleep(1) async def process_plugin_message(self, msg) -> bool: """使用插件处理消息""" # 获取所有消息处理插件 # 关闭00:30-05:00的系统交互，降低被风控风险 current_hour = time.localtime().tm_hour current_minute = time.localtime().tm_min is_sleep_time = (current_hour == 0 and current_minute >= 30) or (1 <= current_hour < 5) if is_sleep_time: # 只处理特定消息，如管理员消息或紧急消息 self.LOG.info(self._trace_message(msg, f"夜间休眠时间(00:30-05:00)，忽略消息: {msg}")) return False message_plugins = self.plugin_registry.get_plugins_by_type(MessagePluginInterface) message_plugins = self._sort_message_plugins(message_plugins) if not message_plugins: return False # 依次尝试处理消息 for plugin in message_plugins: if plugin.status != PluginStatus.RUNNING: continue # 这里在进入插件前统一准备统计上下文： # 1. 事件系统删除后，插件调用统计需要直接在主链路埋点； # 2. 提前抽出 room_id / sender / command，后续无论成功还是异常都能复用； # 3. 这样可以保证观测逻辑收口在一处，避免每个插件自己重复埋点。 room_id = msg.roomid if msg.from_group() else "" sender = msg.sender command_name = self._extract_plugin_command(msg) started_at = time.perf_counter() try: # 转换消息为插件可处理的格式 plugin_msg = { "type": msg.msg_type, "content": msg.content.clean_content, "sender": sender, "roomid": room_id, "is_at": msg.is_at(self.wxid), "timestamp": time.time(), "trace_id": self._get_trace_id(msg), "all_contacts": self.allContacts, "full_wx_msg": msg, "gbm": self.gbm, "bot": self.ipad_bot, "revoke": self.message_auto_revoke } # 检查插件是否可以处理该消息 if plugin.can_process(plugin_msg): processed, _ = await plugin.process_message(plugin_msg) self._record_plugin_call_result( plugin=plugin, msg=msg, command_name=command_name, # 这里把“无异常执行完成”视为统计意义上的成功： # 1. 很多插件返回 False 只是表示“本次不拦截”或“异步排队后继续放行”； # 2. 若直接把 processed=False 记成失败，会把成功率统计严重拉低； # 3. 真正的失败已经会走异常分支，因此统计层这里按“未抛错即成功”更合理。 process_result=True, process_time_ms=self._elapsed_ms(started_at), ) if processed: self.LOG.info( self._trace_message( msg, f"插件命中 plugin={plugin.name} command={command_name} " f"cost_ms={self._elapsed_ms(started_at)}" ) ) return True except Exception as e: self._record_plugin_call_error( plugin=plugin, msg=msg, command_name=command_name, error=e, ) self.LOG.error(self._trace_message(msg, f"插件 {plugin.name} 处理消息失败: {e}")) return False def _attach_trace_id(self, msg: WxMessage) -> str: """为消息对象附加稳定 trace_id，便于后续全链路关联。""" trace_id = self._get_trace_id(msg) if trace_id: return trace_id msg_id = str(getattr(msg, "msg_id", "") or "0") create_time = str(getattr(msg, "create_time", "") or "0") sender_tail = str(getattr(msg, "sender", "") or "")[-6:] or "unknown" random_tail = uuid.uuid4().hex[:6] trace_id = f"wx-{msg_id}-{create_time}-{sender_tail}-{random_tail}" setattr(msg, "trace_id", trace_id) return trace_id @staticmethod def _get_trace_id(msg: WxMessage) -> str: """读取消息对象上的 trace_id；若不存在则返回空字符串。""" return str(getattr(msg, "trace_id", "") or "").strip() def _trace_message(self, msg: WxMessage, message: str) -> str: """为日志消息统一追加 trace_id 前缀。""" trace_id = self._get_trace_id(msg) if not trace_id: return message return f"[trace_id={trace_id}] {message}" @staticmethod def _elapsed_ms(started_at: float) -> float: """把 monotonic 起始时间转换为毫秒耗时。""" return round((time.perf_counter() - started_at) * 1000, 2) @staticmethod def _extract_plugin_command(msg: WxMessage) -> str: """尽力从消息内容中提取一个可读的“触发命令”。""" # 这里不追求把所有命令解析得非常精确，只要能满足后台统计可读性即可： # 1. 文本消息优先取第一段词，避免把整句长文本都记成 command； # 2. 非文本消息统一落到消息类型名，便于区分“文本触发”和“链接触发”等场景； # 3. 空内容时返回通用占位，避免统计表出现 NULL / 空字符串。 raw_content = str(getattr(getattr(msg, "content", None), "clean_content", "") or "").strip() if raw_content: first_token = raw_content.split()[0].strip() return first_token[:50] if first_token else "[文本消息]" msg_type = getattr(getattr(msg, "msg_type", None), "name", "") return f"[{msg_type or 'UNKNOWN'}]" def _get_stats_collector_plugin(self): """获取运行中的统计收集插件实例。""" # 统计插件已经从“事件订阅”切到“主链路直接回调”， # 因此每次埋点前都需要安全地确认插件实例是否存在且处于运行态。 plugin = self.plugin_manager.plugins.get("指令记录") if not plugin: return None if getattr(plugin, "status", None) != PluginStatus.RUNNING: return None return plugin def _record_plugin_call_result( self, *, plugin, msg: WxMessage, command_name: str, process_result: bool, process_time_ms: float, ) -> None: """将插件执行结果直接写入统计插件。""" stats_plugin = self._get_stats_collector_plugin() if not stats_plugin or not hasattr(stats_plugin, "record_plugin_call"): return try: stats_plugin.record_plugin_call( plugin_name=plugin.name, command=command_name, user_id=msg.sender, group_id=msg.roomid if msg.from_group() else None, is_group=msg.from_group(), process_result=process_result, process_time_ms=process_time_ms, trace_id=self._get_trace_id(msg), ) except Exception as stats_error: self.LOG.error(self._trace_message(msg, f"记录插件调用统计失败: plugin={plugin.name}, error={stats_error}")) def _record_plugin_call_error( self, *, plugin, msg: WxMessage, command_name: str, error: Exception, ) -> None: """将插件执行异常直接写入统计插件。""" stats_plugin = self._get_stats_collector_plugin() if not stats_plugin or not hasattr(stats_plugin, "record_plugin_error"): return try: stats_plugin.record_plugin_error( plugin_name=plugin.name, command=command_name, user_id=msg.sender, group_id=msg.roomid if msg.from_group() else None, is_group=msg.from_group(), error_message=str(error), trace_id=self._get_trace_id(msg), # 这里保留完整堆栈，便于后台直接查看异常上下文，而不必只看摘要日志。 stack_trace=traceback.format_exc(), ) except Exception as stats_error: self.LOG.error(self._trace_message(msg, f"记录插件异常统计失败: plugin={plugin.name}, error={stats_error}")) @staticmethod def _sort_message_plugins(message_plugins): """将兜底型插件放到最后执行，避免影响其他插件命中。""" if not message_plugins: return [] def is_fallback_plugin(plugin): feature_key = str(getattr(plugin, "feature_key", "") or "").strip().upper() module_name = str(getattr(plugin.__class__, "__module__", "") or "").lower() plugin_name = str(getattr(plugin, "name", "") or "").strip().lower() return ( feature_key == "AI_AUTO_RESPONSE" or "plugins.ai_auto_response" in module_name or plugin_name in {"小牛群聊bot", "ai_auto_response"} ) normal_plugins = [plugin for plugin in message_plugins if not is_fallback_plugin(plugin)] fallback_plugins = [plugin for plugin in message_plugins if is_fallback_plugin(plugin)] return normal_plugins + fallback_plugins def get_all_contacts(self) -> dict: """获取所有联系人信息并返回字典格式 {wxid: nickname}""" try: # 从数据库获取联系人信息 contacts = self.contacts_db.get_all_contacts() return contacts except Exception as e: self.LOG.error(f"获取联系人信息失败: {e}") return {} def get_all_head_images(self) -> dict: """获取所有的联系人头像信息""" try: # 从数据库获取所有联系人的头像信息 head_images = self.contacts_db.get_all_contacts_avatar() return head_images except Exception as e: self.LOG.error(f"获取所有联系人头像信息失败: {e}") return {} async def refresh_contacts_db(self): """刷新联系人信息""" self.LOG.info("开始刷新联系人信息") contacts = await self.ipad_bot.get_contract_list() self.LOG.debug(f"获取到的联系人：{contacts}") batch_size = 20 discovered_groups = set() for i in range(0, len(contacts), batch_size): batch_contacts = contacts[i:i + batch_size] contact_info = await self.ipad_bot.get_contract_detail(batch_contacts) self.LOG.debug(f"获取到的联系人详细信息数量：{len(contact_info)}") friend_contacts = [] official_contacts = [] for contact in contact_info: user_name = contact.get("UserName") if isinstance(user_name, dict): user_name = user_name.get("string", "") user_name = user_name or "" if not user_name: continue if user_name.endswith("@chatroom"): discovered_groups.add(user_name) # 群资料这里不能只在“首次发现”时写入： # 1. 群头像、小群名、公告等字段都可能在微信侧发生变化； # 2. 如果只插入不更新，后续头像缓存拿到的仍然会是旧 URL； # 3. 因此每次刷新通讯录都做一次 upsert，确保群资料是最新的。 self.contacts_db.save_chatroom_info(contact) continue if user_name.startswith("gh_"): official_contacts.append(contact) else: friend_contacts.append(contact) # 联系人详情这里必须允许覆盖更新： # 1. get_contract_detail 已经重新向远端拿到了最新昵称、签名、头像 URL； # 2. 如果 still only_insert=True，库里旧联系人将永远保留历史头像地址； # 3. 改成 upsert 后，后续头像缓存同步才能真正拿到最新 URL 并下载新头像。 if friend_contacts: self.contacts_db.save_contacts(friend_contacts, "friends", only_insert=False) if official_contacts: self.contacts_db.save_contacts(official_contacts, "ghs", only_insert=False) groups = self.contacts_db.get_chatroom_list() for group in groups: group_id = group["chatroom_id"] discovered_groups.add(group_id) chatroom_info = await self.ipad_bot.get_chatroom_info(group_id) self.LOG.debug(f"获取到的群成员信息：{chatroom_info}") if chatroom_info.get("UserName", ""): members = await self.ipad_bot.get_chatroom_member_list(group_id) if members: active_member_wxids = [] for member in members: wxid = member.get("UserName", "") if isinstance(wxid, dict): wxid = wxid.get("string", "") if wxid: active_member_wxids.append(wxid) self.contacts_db.mark_chatroom_members_active(group_id, active_member_wxids) self.contacts_db.mark_chatroom_members_left(group_id, active_member_wxids) # 群成员头像 URL 同样需要覆盖更新： # 1. 群成员换头像后，成员表里的 small_head_img_url 会变； # 2. 若只做 INSERT IGNORE，则历史记录不会被刷新； # 3. 这里改成 upsert，保证后台通讯录与头像缓存都能感知到最新头像地址。 self.contacts_db.save_chatroom_member_simple(group_id, members, only_insert=False) self.LOG.info(f"已增量同步群 {group_id} 的成员信息") else: self.contacts_db.mark_chatroom_members_left(group_id, []) self.LOG.warning(f"群 {group_id} 当前未获取到成员列表，已将历史成员标记为已退群") else: self.LOG.warning(f"获取群 {group_id} 信息失败，保留群资料并将成员标记为已退群。") self.contacts_db.mark_chatroom_members_left(group_id, []) for group_id in discovered_groups: if not self.contacts_db.get_chatroom_info(group_id): chatroom_info = await self.ipad_bot.get_chatroom_info(group_id) if chatroom_info.get("UserName", ""): self.contacts_db.save_chatroom_info(chatroom_info) members = await self.ipad_bot.get_chatroom_member_list(group_id) if members: self.contacts_db.save_chatroom_member_simple(group_id, members, only_insert=False) friends = await self.ipad_bot.get_contract_list() self.allContacts = self.get_all_contacts() self.head_images = self.get_all_head_images() self.all_chatroom_members = self.contacts_db.get_chatroom_member_list_name_all() self.contact_manager.set_contacts(self.allContacts, friends, self.head_images, self.all_chatroom_members) self.LOG.info("联系人信息刷新完成") async def login_twice_auto_auth(self) -> None: try: self.LOG.info(f"定时进行二次登录动作") resp = await self.ipad_bot.twice_auto_auth() if resp: self.LOG.info(f"定时二次登录成功！") if self.ipad_running: self.LOG.info(f"ipad_wechat running:{self.ipad_running}") else: self.ipad_running = True self.LOG.info(f"ipad_wechat stopped change running:{self.ipad_running}") else: self.LOG.error(f"定时二次登录失败！") self.ipad_running = False except Exception as e: self.LOG.error(f"login_twice_auto_auth error: {e}") # ============================================== 系统级任务（刚需）========================================================== async def message_count_to_db(self): try: self.message_storage.write_to_db() except Exception as e: self.LOG.error(f"write_to_db error：{e}")