diff --git a/docs/wechat_ipad多版本Server适配路线图.md b/docs/wechat_ipad多版本Server适配路线图.md index a623b61..3931013 100644 --- a/docs/wechat_ipad多版本Server适配路线图.md +++ b/docs/wechat_ipad多版本Server适配路线图.md @@ -20,33 +20,37 @@ - 已新增 [wechat_ipad/provider_base.py](/d:/learn/abot/wechat_ipad/provider_base.py:1) - 已新增 `providers/legacy_855/` 独立目录,并迁入当前 855/859 协议实现 - 已将 [robot.py](/d:/learn/abot/robot.py:1) 的接入实例化入口切换为 `WechatGateway` +- 已新增 [wechat_ipad/providers/legacy_855/runtime.py](/d:/learn/abot/wechat_ipad/providers/legacy_855/runtime.py:1) +- 已将 855 的登录、历史消息拉取、心跳、长心跳、消息轮询、掉线二次登录恢复迁入 `legacy_855` provider +- 已将 [robot.py](/d:/learn/abot/robot.py:1) 精简为“注册回调 + 业务处理”,不再直接维护 855 的运行时主循环 +- 已补上 `Legacy855WechatClient` 的显式初始化入口,避免 provider 多继承构造链不稳定 当前尚未完成的关键项: -- `Robot` 中的登录、心跳、长心跳、消息轮询、掉线恢复逻辑仍未完全迁入 855 provider -- `legacy_855` 目录内尚未补充独立的 `runtime.py` 收口运行时模型 - 855 provider 仍需完成一轮“当前项目实际依赖接口”的可上线回归验证 +- 855 provider 仍需继续梳理“项目真实使用到的接口覆盖面”,确认是否还有遗漏的旧能力残留在历史目录 +- 864 provider 尚未开始接入,当前统一接口仍主要围绕 855 第一阶段目标进行验证 因此,当前状态可以定义为: - “接入入口已收口” -- “运行时主链路迁移进行中” +- “855 运行时主链路已迁入 provider” - “尚未达到 855 可直接替换现网上线的最终状态” ## 2. 当前问题概览 ### 2.1 当前耦合点 -当前微信接入实现存在以下特点: +当前微信接入实现仍需关注以下历史耦合点与残留影响: - [robot.py](/d:/learn/abot/robot.py:221) 直接读取 `wechat_ipad/config.toml` -- [robot.py](/d:/learn/abot/robot.py:263) 直接实例化 `wechat_ipad.WechatAPIClient` -- `Robot` 自己承担了登录、心跳、长心跳、消息轮询、掉线恢复等运行时职责 -- `wechat_ipad/client/*.py` 直接面向当前 server 协议编写,接口路径、请求体、返回结构都写死 +- `Robot` 的实例化入口虽然已切到 `WechatGateway`,但配置读取与业务初始化仍在主程序中 +- 855 的运行时职责已经迁入 provider,但 864 尚未接入验证,统一抽象仍需继续收敛 +- `wechat_ipad/client/*.py` 仍作为历史目录存在,接口路径、请求体、返回结构都面向旧 server 编写 这导致一个结果: -- 一旦 server 版本变化,不只是 `client` 要改,`Robot` 主链路也要跟着改 +- 如果继续在历史目录或 `Robot` 主链路里堆版本判断,后续 server 版本变化时改动面仍会再次放大 ### 2.2 855 / 864 的核心差异 diff --git a/robot.py b/robot.py index 426e3ff..6611007 100644 --- a/robot.py +++ b/robot.py @@ -6,8 +6,6 @@ import tomllib import traceback import uuid from collections import deque - -import toml from loguru import logger import wechat_ipad @@ -271,145 +269,119 @@ class Robot: server_type = str(self.ipad_config.get("server_type", "legacy_855") or "legacy_855").strip() self.ipad_bot = WechatGateway(server_ip, server_port, server_type=server_type) 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)} 条历史消息,开始仅落库归档") - for raw_message in data: - await self._archive_startup_history_message(raw_message) - 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) + # 855 provider 现在自行承接运行时模型: + # 1. provider 内部负责登录、历史消息拉取、心跳、长心跳、掉线恢复与实时轮询; + # 2. Robot 只注册业务回调,继续处理联系人初始化、消息归档、插件调度等项目内逻辑; + # 3. 这样未来切到 864 时,主链路只需要替换 provider,而不是继续改这里的大循环。 + await self.ipad_bot.run_runtime( + ipad_config=self.ipad_config, + config_path="wechat_ipad/config.toml", + logger=self.LOG, + on_login_ready=self._on_ipad_login_ready, + on_history_message=self._archive_startup_history_message, + on_message=self._handle_runtime_message, + on_idle_payload=self._handle_runtime_idle_payload, + on_logout=self._handle_ipad_logout, + on_runtime_state_change=self._handle_runtime_state_change, + ) except Exception as e: self.LOG.exception(f"wechat_ipad客户端运行出错: {e}") self.ipad_running = False - # 在类里直接写一个内联 async 方法(不额外抽取新的对外方法) + async def _on_ipad_login_ready(self, login_identity: dict) -> None: + """处理 provider 登录成功后的项目侧初始化动作。 + + 这里保留在 Robot 的原因很明确: + 1. 联系人缓存、插件注入、消息归档器、成员监控器都属于项目业务层能力; + 2. provider 不应该知道本项目有哪些数据库表、后台缓存或插件系统; + 3. 因此登录“流程”放到 provider,登录后的“业务初始化”继续留在 Robot。 + """ + self.wxid = login_identity.get("wxid", "") + self.nickname = login_identity.get("nickname", "") + self.alias = login_identity.get("alias", "") + self.phone = login_identity.get("phone", "") + self.signature = login_identity.get("signature", "") + + # 这里同时把 Robot 侧的身份信息镜像回 bot,保证旧代码仍可从 `self.ipad_bot.xxx` 读取。 + self.ipad_bot.wxid = self.wxid + self.ipad_bot.nickname = self.nickname + self.ipad_bot.alias = self.alias + self.ipad_bot.phone = self.phone + self.ipad_bot.signature = self.signature + self.LOG.info( + f"wechat_ipad登录账号信息: wxid: {self.wxid} 昵称: {self.nickname} 微信号: {self.alias} 手机号: {self.phone}" + ) + + self.plugin_manager.inject_bot(self.ipad_bot) + 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.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") + + async def _handle_runtime_message(self, raw_message: dict) -> None: + """处理 provider 交付的单条实时原始消息。""" + try: + wxmsg: WxMessage = WxMessage.from_json(raw_message) + self._attach_trace_id(wxmsg) + msg_id = wxmsg.msg_id + if msg_id in self.recent_msg_ids: + self.LOG.info(self._trace_message(wxmsg, f"出现重复ID消息: {msg_id}")) + return + 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 解析失败,消息内容: {raw_message},错误: {e}") + return + + # 这里继续沿用“单条消息单独起任务 + 信号量限流”的项目策略: + # 1. 保持与现网处理吞吐一致; + # 2. 避免 provider 轮询被某条耗时消息阻塞; + # 3. 也不把并发控制职责再塞回 provider,边界更清楚。 + asyncio.create_task(self._process_with_semaphore(wxmsg)) + + async def _handle_runtime_idle_payload(self, data_temp: dict) -> None: + """处理 855 空轮询之外的补充同步负载,例如群成员变更通知。""" + if isinstance(data_temp, dict) and data_temp.get("Ret") == 0 and "KeyBuf" in data_temp: + return + + 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: + asyncio.create_task(self.member_monitor.check_and_handle_changes(group_id)) + + async def _handle_ipad_logout(self, reason: str) -> None: + """处理 provider 识别到的掉线事件,仅负责业务侧告警。""" + self.LOG.error(f"用户可能退出: {reason}") + self.email_sender.send_wechat_alert( + self.config.email.get("alert_recipient"), + f"用户可能退出: {reason}", + self.wxid, + self.nickname, + ) + + async def _handle_runtime_state_change(self, running: bool) -> None: + """镜像 provider 运行态到 Robot,供后台与运维逻辑读取。""" + self.ipad_running = running async def _archive_startup_history_message(self, raw_message: dict) -> None: """启动阶段只归档历史消息,不触发实时业务处理。 @@ -471,109 +443,6 @@ class Robot: 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: @@ -658,6 +527,8 @@ class Robot: def stop_wechat_ipad(self): """停止wechat_ipad客户端""" self.ipad_running = False + if hasattr(self, "ipad_bot") and self.ipad_bot and hasattr(self.ipad_bot, "stop_runtime"): + self.ipad_bot.stop_runtime() if self.ipad_loop: self.ipad_loop.stop() self.LOG.info("wechat_ipad客户端已停止") @@ -999,24 +870,6 @@ class Robot: 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): diff --git a/wechat_ipad/providers/legacy_855/provider.py b/wechat_ipad/providers/legacy_855/provider.py index 92d53a6..baa3b0e 100644 --- a/wechat_ipad/providers/legacy_855/provider.py +++ b/wechat_ipad/providers/legacy_855/provider.py @@ -5,8 +5,10 @@ from wechat_ipad.providers.legacy_855.friends import FriendMixin from wechat_ipad.providers.legacy_855.group import ChatroomMixin from wechat_ipad.providers.legacy_855.login import LoginMixin from wechat_ipad.providers.legacy_855.message import MessageMixin +from wechat_ipad.providers.legacy_855.runtime import Legacy855RuntimeMixin from wechat_ipad.providers.legacy_855.tools import ToolMixin from wechat_ipad.providers.legacy_855.user import UserMixin +from wechat_ipad.providers.legacy_855.base import WechatAPIClientBase class Legacy855WechatClient( @@ -17,6 +19,7 @@ class Legacy855WechatClient( ChatroomMixin, UserMixin, ToolMixin, + Legacy855RuntimeMixin, WechatProviderBase, ): """855/859 风格 server 的独立 Provider。 @@ -30,6 +33,19 @@ class Legacy855WechatClient( provider_name = "legacy_855" server_type = "legacy_855" + def __init__(self, ip: str, port: int, **kwargs): + """初始化 855 provider。 + + 说明: + 1. 旧 `wechat_ipad/client` 的多继承结构没有显式构造入口,迁移后这里补一个稳定初始化点; + 2. 基础连接信息仍写入 `WechatAPIClientBase`,消息发送队列继续沿用 `MessageMixin` 的实现; + 3. 运行时状态由 `Legacy855RuntimeMixin` 单独初始化,便于后续 864 provider 走自己的模型。 + """ + del kwargs + WechatAPIClientBase.__init__(self, ip, port) + MessageMixin.__init__(self) + self._init_runtime_state() + async def send_at_message(self, wxid: str, content: str, at: list[str]) -> tuple[int, int, int]: """发送 @ 消息,兼容现有插件调用方式。""" if not self.wxid: @@ -47,4 +63,3 @@ class Legacy855WechatClient( output = content return await self.send_text_message(wxid, output, at) - diff --git a/wechat_ipad/providers/legacy_855/runtime.py b/wechat_ipad/providers/legacy_855/runtime.py new file mode 100644 index 0000000..b01e21b --- /dev/null +++ b/wechat_ipad/providers/legacy_855/runtime.py @@ -0,0 +1,386 @@ +import asyncio +import time +from typing import Any, Awaitable, Callable + +import toml + + +AsyncCallback = Callable[..., Awaitable[None]] + + +class Legacy855RuntimeMixin: + """855/859 风格 server 的运行时编排实现。 + + 设计说明: + 1. 855 的差异不只是接口路径,而是“客户端自己负责保活和拉消息”的运行模型; + 2. 这里把登录、心跳、历史消息消化、实时轮询、掉线恢复集中收口到 provider 内部; + 3. `Robot` 只再关心“登录成功后做什么”“收到消息后怎么处理”,避免后续接 864 时继续改主链路。 + """ + + def _init_runtime_state(self) -> None: + """初始化运行时状态字段。 + + 这里不用 `__init__` 参与多继承链,而是由 provider 显式调用: + 1. 现有 mixin 组合里已经有消息队列初始化逻辑; + 2. 显式初始化更容易看清哪些状态只属于 855 runtime; + 3. 也能避免后续再因为 MRO 调用顺序引入隐蔽问题。 + """ + self._runtime_running = False + self._runtime_recovery_lock = asyncio.Lock() + + def stop_runtime(self) -> None: + """请求停止当前 provider 的运行时主循环。""" + self._runtime_running = False + + def is_runtime_running(self) -> bool: + """返回当前 provider 运行时是否处于运行态。""" + return bool(getattr(self, "_runtime_running", False)) + + async def run_runtime( + self, + *, + ipad_config: dict, + config_path: str, + logger, + on_login_ready: AsyncCallback, + on_history_message: AsyncCallback, + on_message: AsyncCallback, + on_idle_payload: AsyncCallback | None = None, + on_logout: AsyncCallback | None = None, + on_runtime_state_change: AsyncCallback | None = None, + ) -> None: + """启动 855 provider 的完整运行时。 + + 参数说明: + 1. `ipad_config` 与 `config_path` 由上层传入,provider 只负责更新和落盘登录态; + 2. `on_*` 回调保持尽量少,只暴露业务层真正需要接手的几个时机; + 3. 这样既避免 `Robot` 再写协议细节,也不额外引入复杂的事件总线或状态机层。 + """ + wxid = str(ipad_config.get("wxid", "") or "").strip() + device_name = str(ipad_config.get("device_name", "") or "").strip() + device_id = str(ipad_config.get("device_id", "") or "").strip() + + if not device_name: + device_name = self.create_device_name() + if not device_id: + device_id = self.create_device_id() + + await self._ensure_login( + wxid=wxid, + device_name=device_name, + device_id=device_id, + ipad_config=ipad_config, + config_path=config_path, + logger=logger, + ) + + # 登录后的项目初始化若失败,应直接中断启动: + # 1. 这里会初始化联系人缓存、插件注入、消息存储等关键依赖; + # 2. 如果仅记录异常继续运行,后续消息循环只会在半初始化状态下引发更多连锁问题; + # 3. 因此这里刻意不吞异常,让启动期问题尽早暴露。 + await on_login_ready(self.get_login_identity(device_name=device_name, device_id=device_id)) + logger.info(f"wechat_ipad登录设备信息: device_name: {device_name} device_id: {device_id}") + logger.info("wechat_ipad登录成功") + + logger.info("处理堆积消息中") + await self._drain_startup_history(on_history_message=on_history_message, logger=logger) + logger.info("处理堆积消息完毕") + + await self._set_runtime_running(True, on_runtime_state_change=on_runtime_state_change, logger=logger) + heartbeat_task = asyncio.create_task( + self._heartbeat_loop( + heartbeat_func=self.heartbeat, + interval_seconds=60, + logger=logger, + loop_name="wechat_ipad心跳", + on_logout=on_logout, + on_runtime_state_change=on_runtime_state_change, + ) + ) + heartbeat_long_task = asyncio.create_task( + self._heartbeat_loop( + heartbeat_func=self.heartbeat_long, + interval_seconds=120, + logger=logger, + loop_name="wechat_ipad长连接心跳", + on_logout=on_logout, + on_runtime_state_change=on_runtime_state_change, + ) + ) + + try: + logger.info("开始处理wechat_ipad消息") + while self.is_runtime_running(): + try: + data_temp = await self.sync_message() + except Exception as e: + logger.error(f"获取新消息失败 {e}") + recovered = await self._try_recover_from_logout( + reason=e, + logger=logger, + on_logout=on_logout, + on_runtime_state_change=on_runtime_state_change, + ) + if not recovered: + if not self.is_runtime_running(): + break + await asyncio.sleep(5) + continue + + data = data_temp.get("AddMsgs") + if data: + for message in data: + await self._safe_callback( + on_message, + message, + logger=logger, + callback_name="on_message", + ) + else: + # 对于 855 而言,`Ret=0 + KeyBuf` 只是正常空轮询返回,不需要额外刷屏日志。 + if on_idle_payload: + await self._safe_callback( + on_idle_payload, + data_temp, + logger=logger, + callback_name="on_idle_payload", + ) + + await asyncio.sleep(2) + finally: + heartbeat_task.cancel() + heartbeat_long_task.cancel() + await asyncio.gather(heartbeat_task, heartbeat_long_task, return_exceptions=True) + await self._set_runtime_running(False, on_runtime_state_change=on_runtime_state_change, logger=logger) + + async def _ensure_login( + self, + *, + wxid: str, + device_name: str, + device_id: str, + ipad_config: dict, + config_path: str, + logger, + ) -> None: + """保证当前 provider 已完成登录,并把登录结果写回配置。 + + 这里沿用现有 855 的行为: + 1. 优先复用缓存唤醒; + 2. 唤醒失败或无缓存时回退到二维码登录; + 3. 登录成功后继续把 wxid / device 信息写回 `config.toml`,保持现有部署习惯不变。 + """ + if await self.is_logged_in(wxid): + self.wxid = wxid + profile = await self.get_profile() + self.nickname = profile.get("NickName", {}).get("string", "") + self.alias = profile.get("Alias", "") + self.phone = profile.get("BindMobile", {}).get("string", "") + self.signature = profile.get("Signature", "") + logger.info( + f"wechat_ipad登录账号信息: wxid: {self.wxid} 昵称: {self.nickname} 微信号: {self.alias} 手机号: {self.phone}" + ) + return + + while not await self.is_logged_in(wxid): + uuid = "" + url = "" + try: + if await self.get_cached_info(wxid): + uuid = await self.awaken_login(wxid) + logger.info(f"获取到登录uuid: {uuid}") + else: + uuid, url = await self.get_qr_code(device_id=device_id, device_name=device_name, print_qr=True) + logger.info(f"获取到登录uuid: {uuid}") + logger.info(f"获取到登录二维码: {url}") + except Exception as e: + logger.error(f"登录过程出错: {e}") + uuid, url = await self.get_qr_code(device_id=device_id, device_name=device_name, print_qr=True) + logger.info(f"获取到登录uuid: {uuid}") + logger.info(f"获取到登录二维码: {url}") + + while True: + logger.info(f"uuid: {uuid}, url: {url}") + stat, data = await self.check_login_uuid(uuid, device_id=device_id) + if stat: + break + logger.info(f"等待登录中,过期倒计时:{data}") + await asyncio.sleep(5) + + self._apply_login_result(data=data, logger=logger) + ipad_config["wxid"] = self.wxid + ipad_config["device_name"] = device_name + ipad_config["device_id"] = device_id + ipad_config["login_time"] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + with open(config_path, "w", encoding="utf-8") as f: + toml.dump(ipad_config, f) + break + + def _apply_login_result(self, *, data: dict, logger) -> None: + """把登录接口返回的用户信息统一写回当前 provider。""" + acct_section = data.get("acctSectResp", {}) or {} + self.wxid = acct_section.get("userName", "") + self.nickname = acct_section.get("nickName", "") + self.alias = acct_section.get("alias", "") + self.phone = acct_section.get("bindMobile", "") + self.signature = data.get("Signature", "") + logger.info( + f"wechat_ipad登录账号信息: wxid: {self.wxid} 昵称: {self.nickname} 微信号: {self.alias} 手机号: {self.phone}" + ) + + async def _drain_startup_history(self, *, on_history_message: AsyncCallback, logger) -> None: + """在实时主循环前先消化堆积消息。 + + 这里保持旧逻辑的退出条件: + 1. 连续多次轮询不到 `AddMsgs` 才认为历史堆积已经处理完; + 2. 每批历史消息仍交给上层回调决定如何归档; + 3. provider 只负责拉取与调度,不把历史消息也混入实时业务处理。 + """ + empty_rounds = 0 + while True: + data = await self.sync_message() + add_msgs = data.get("AddMsgs") + if not add_msgs: + if empty_rounds > 2: + break + empty_rounds += 1 + continue + + logger.debug(f"接受到 {len(add_msgs)} 条历史消息,开始仅落库归档") + for raw_message in add_msgs: + await self._safe_callback( + on_history_message, + raw_message, + logger=logger, + callback_name="on_history_message", + ) + await asyncio.sleep(1) + + async def _heartbeat_loop( + self, + *, + heartbeat_func: Callable[[], Awaitable[bool]], + interval_seconds: int, + logger, + loop_name: str, + on_logout: AsyncCallback | None, + on_runtime_state_change: AsyncCallback | None, + ) -> None: + """统一承接心跳与长心跳循环,减少 855 provider 内部重复代码。""" + logger.info(f"开启{loop_name}!") + while self.is_runtime_running(): + try: + success = await heartbeat_func() + if success: + logger.debug(f"{loop_name}进行中") + else: + logger.warning(f"{loop_name}失败") + except Exception as e: + logger.error(f"{loop_name}: {e}") + recovered = await self._try_recover_from_logout( + reason=e, + logger=logger, + on_logout=on_logout, + on_runtime_state_change=on_runtime_state_change, + ) + if not recovered and not self.is_runtime_running(): + break + await asyncio.sleep(interval_seconds) + + async def _try_recover_from_logout( + self, + *, + reason: Exception | str, + logger, + on_logout: AsyncCallback | None, + on_runtime_state_change: AsyncCallback | None, + ) -> bool: + """处理 855 provider 的掉线恢复逻辑。 + + 关键点: + 1. 855 的掉线恢复是 provider 运行模型的一部分,因此也应该收口在 provider 内部; + 2. 这里用锁把恢复流程串行化,避免心跳线程与消息轮询线程同时触发二次登录; + 3. 上层只接收一个“检测到掉线”的通知,用于发告警或记录运维日志。 + """ + if not self._is_logout_reason(reason): + return False + + async with self._runtime_recovery_lock: + # 进入锁后再判断一次,避免并发恢复时第二个协程重复执行二次登录。 + if not self.is_runtime_running(): + return False + + await self._safe_callback( + on_logout, + str(reason), + logger=logger, + callback_name="on_logout", + ) + + try: + logger.info("定时进行二次登录动作") + resp = await self.twice_auto_auth() + if resp: + logger.info("定时二次登录成功!") + return True + logger.error("定时二次登录失败!") + except Exception as e: + logger.error(f"login_twice_auto_auth error: {e}") + + await self._set_runtime_running(False, on_runtime_state_change=on_runtime_state_change, logger=logger) + return False + + async def _set_runtime_running( + self, + running: bool, + *, + on_runtime_state_change: AsyncCallback | None, + logger, + ) -> None: + """同步 provider 运行态,并通知上层镜像状态。""" + self._runtime_running = running + if on_runtime_state_change: + await self._safe_callback( + on_runtime_state_change, + running, + logger=logger, + callback_name="on_runtime_state_change", + ) + + async def _safe_callback( + self, + callback: AsyncCallback | None, + *args: Any, + logger, + callback_name: str, + ) -> None: + """统一保护上层回调,避免单个业务异常直接打断 provider 主循环。""" + if callback is None: + return + try: + await callback(*args) + except Exception as e: + logger.exception(f"执行回调失败: {callback_name}, error: {e}") + + @staticmethod + def _is_logout_reason(reason: Exception | str) -> bool: + """判断当前异常是否属于 855 provider 约定的掉线场景。""" + return "用户可能退出" in str(reason) + + def get_login_identity(self, *, device_name: str = "", device_id: str = "") -> dict[str, Any]: + """返回当前登录身份的轻量归一化结构。 + + 第一阶段先继续使用 dict: + 1. 便于 `Robot` 直接消费,不额外引入 dataclass; + 2. 后续如果 864 也需要对齐结构,可以在 provider 内继续增量补字段; + 3. 这里同时把 device 信息带上,方便上层统一打印和展示。 + """ + return { + "wxid": self.wxid, + "nickname": self.nickname, + "alias": self.alias, + "phone": self.phone, + "signature": getattr(self, "signature", ""), + "device_name": device_name, + "device_id": device_id, + }