# -*- coding: utf-8 -*- import time import asyncio import threading import tomllib import toml import wechat_ipad from loguru import logger from base.func_epic import is_friday, get_free from base.func_news import News from configuration import Config from plugin_common.event_system import EventType, EventSystem from plugin_common.message_plugin_interface import MessagePluginInterface from plugin_common.plugin_interface import PluginStatus from plugin_common.plugin_manager import PluginManager from plugin_common.plugin_registry import PluginRegistry from sehuatang.shehuatang import pdf_file_path from utils.revoke.message_auto_revoke import MessageAutoRevoke from utils.robot_cmd.robot_command import GroupBotManager, Feature, PermissionStatus from db.connection import DBConnectionManager from db.contacts_db import ContactsDBOperator from utils.wechat.contact_manager import ContactManager from utils.wechat.message_to_db import MessageStorage from wechat_ipad import WechatAPIClient from wechat_ipad.models.message import WxMessage, MessageType from plugins.xiuren_image.meitu_dl import meitu_dowload_pub_pic class Robot: """个性化自己的机器人 """ def __init__(self, config: Config) -> None: super().__init__() self.config = config self.LOG = logger # 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.info(f"DB+REDIS 连接池开始初始化") # 使用单例模式获取实例 self.db_manager = DBConnectionManager.get_instance( mysql_config=self.config.mariadb, redis_config=self.config.redis ) self.LOG.info(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.contact_manager = ContactManager.get_instance() self.allContacts = {} # 将在登录后填充 self.groups = {} # 存储按group_id分组的消息列表，每个group_id最多保留10条消息 GroupBotManager.load_local_cache() # 权限模块加载 self.gbm = GroupBotManager() # 初始化插件系统 self.LOG.info("开始初始化插件系统...") self.plugin_registry = PluginRegistry() self.event_system = EventSystem() self.plugin_modules = {} # 存储已加载的插件模块 self.plugins = {} # 存储已加载的插件实例 # 设置插件系统上下文 self.system_context = { "config": config, "event_system": self.event_system, "plugin_registry": self.plugin_registry, "db_pool": self.db_pool, "redis_pool": self.redis_pool } 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() # 加载插件 self.LOG.info("插件系统初始化完成") 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.info("正在初始化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", 8058) # 创建事件循环 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.info("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.info("启动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") # 更新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.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() self.head_images = self.get_all_head_images() self.contact_manager.set_contacts(self.allContacts, self.head_images) self.message_storage = MessageStorage(self.ipad_bot) # 先接受堆积消息 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 = await self.ipad_bot.sync_message() except Exception as e: self.LOG.error(f"获取新消息失败 {e}") if "用户可能退出" in str(e): self.LOG.error(f"用户可能退出: {e}") await self.login_twice_auto_auth() await asyncio.sleep(5) continue data = data.get("AddMsgs") if data: tasks = [] for message in data: # self.LOG.debug(f"sync_message.处理消息消息内容: {message}") # 处理消息 try: wxmsg: WxMessage = WxMessage.from_json(message) except Exception as e: self.LOG.error(f"WxMessage.from_json 解析失败，消息内容: {message}，错误: {e}") continue # 跳过本条消息，继续处理下一条 tasks.append(self._process_ipad_message(wxmsg)) if tasks: await asyncio.gather(*tasks) # 使用异步睡眠替代忙等待循环 await asyncio.sleep(0.5) except Exception as e: self.LOG.error(f"wechat_ipad客户端运行出错: {e}") self.ipad_running = False 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") # 更新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.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}") 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}") 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对象，直接使用其属性和方法 from_user = message.sender to_user = message.to_user content = message.content msg_type = message.msg_type # 判断是否为群消息 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}") 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 self.contact_manager.set_contacts(self.allContacts, self.head_images) self.LOG.info(f"已更新群 {group_id} 的成员信息") except Exception as e: self.LOG.error(f"获取群成员信息失败: {e}") # 发布消息接收事件 self.event_system.publish(EventType.MESSAGE_RECEIVED, {"message": message}) # 尝试使用插件处理消息 plugin_processed = await self.process_plugin_message(message) if is_group: self.LOG.debug(f"入库和记录群消息: {message}") # 调用统计逻辑进行聊天数据统计： try: self.message_storage.process_message(message) except Exception as e: self.LOG.error(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(f"archive_message error: {e}") except Exception as e: self.LOG.error(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: # self.runPendingJobs() time.sleep(1) # 添加一个方法用于刷新联系人信息 def refresh_contacts(self): """刷新联系人信息""" self.allContacts = self.get_all_contacts() self.contact_manager.refresh_contacts(self.allContacts) self.LOG.info("联系人信息已刷新") async def send_group_txt_message(self, msg: str, feature: Feature): """向所有启用了特定功能的群发送文本消息""" try: receivers = self.gbm.get_group_list() if not receivers: return for r in receivers: if self.gbm.get_group_permission(r, feature) == PermissionStatus.ENABLED: await self.ipad_bot.send_text_message(r, msg) except Exception as e: self.LOG.error(f"send_group_txt_message:{feature.description} error：{e}") async def send_group_file_message(self, path: str, feature: Feature): try: receivers = self.gbm.get_group_list() if not receivers: return for r in receivers: if self.gbm.get_group_permission(r, feature) == PermissionStatus.ENABLED: await self.ipad_bot.send_image_message(r, path) except Exception as e: self.LOG.error(f"send_group_file_message:{feature.description} error：{e}") 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(f"夜间休眠时间(00:30-05:00)，忽略消息: {msg}") return False message_plugins = self.plugin_registry.get_plugins_by_type(MessagePluginInterface) if not message_plugins: return False # 依次尝试处理消息 for plugin in message_plugins: if plugin.status != PluginStatus.RUNNING: continue try: # 转换消息为插件可处理的格式 plugin_msg = { "type": msg.msg_type, "content": msg.content.clean_content, "sender": msg.sender, "roomid": msg.roomid if msg.from_group() else "", "is_at": msg.is_at(self.wxid), "timestamp": time.time(), "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) if processed: # 发布消息处理事件 self.event_system.publish(EventType.MESSAGE_PROCESSED, { "message": msg, "plugin": plugin.name }) return True except Exception as e: self.LOG.error(f"插件 {plugin.name} 处理消息失败: {e}") return False 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("开始刷新联系人信息") # 删除所有的联系人信息 self.contacts_db.delete_all_contacts() self.LOG.debug("已删除所有的联系人信息") contacts = await self.ipad_bot.get_contract_list() self.LOG.debug(f"获取到的联系人：{contacts}") # 获取联系人详细信息,get_contract_detail每次可以获取20个 # 每次获取20个，需要循环获取 # 将联系人列表分成每组20个 batch_size = 20 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)}") self.contacts_db.save_contacts(contact_info, "friends") # 获取群聊列表 groups = self.contacts_db.get_chatroom_list() # 调用接口完成群成员信息获取与保存逻辑 for group in groups: # 调用接口获取群成员信息 group_id = group["chatroom_id"] chatroom_info = await self.ipad_bot.get_chatroom_info(group_id) self.LOG.debug(f"获取到的群成员信息：{chatroom_info}") 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: # 兼容逻辑已放到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 self.contact_manager.set_contacts(self.allContacts, self.head_images) self.LOG.info(f"已更新群 {group_id} 的成员信息") else: self.LOG.error(f"获取群 {group_id} 信息失败，证明用户无该群信息，删除群的相关资料。") # 删除群数据库中的群信息 self.contacts_db.delete_chatroom_all_info(group_id) 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"定时二次登录失败！") except Exception as e: self.LOG.error(f"login_twice_auto_auth error: {e}") # ============================================== 业务内容========================================================== async def news_baidu_report_auto(self) -> None: try: news = News().get_baidu_news() await self.send_group_txt_message(news, Feature.DAILY_NEWS) except Exception as e: self.LOG.error(f"newsBaiduReportAuto error：{e}") async def news_en_report(self, website, sender: str = None) -> None: try: news = News().get_eng_news(website) await self.ipad_bot.send_text_message(sender, news) except Exception as e: self.LOG.error(f"newsEnReport error：{e}") # 使用装饰器标记定时任务 星期五 10:30 执行 async def send_epic_free_games(self): try: if is_friday(): games = get_free() await self.send_group_txt_message(games, Feature.EPIC) except Exception as e: self.LOG.error(f"sendEpicFreeGames 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}") async def generate_sehuatang_pdf(self): try: self.LOG.info("开始生成PDF,generate_sehuatang_pdf") path = pdf_file_path() # 暂时只发4K群 await self.send_group_file_message(path, Feature.PDF_CAPABILITY) except Exception as e: self.LOG.error(f"generateSehuatangPdf error：{e}") async def xiu_ren_download_task(self): try: # 每天下载10组图，然后发一个帖子PDF meitu_dowload_pub_pic() except Exception as e: self.LOG.error(f"xiu_ren_download_task error：{e}") async def generate_and_send_ranking(self): try: receivers = self.gbm.get_group_list() if not receivers: return for r in receivers: if self.gbm.get_group_permission(r, Feature.DAILY_SUMMARY) == PermissionStatus.ENABLED: output = self.message_storage.generate_and_send_ranking(r, self.allContacts) await self.ipad_bot.send_text_message(r, output) except Exception as e: self.LOG.error(f"SendRanking error：{e}") async def send_ranking_task(self): result = await self.ipad_bot.get_chatroom_nickname("Jyunere", "43687793133@chatroom") self.LOG.info(f"send_ranking_task：{result}")