# -*- coding: utf-8 -*- import logging import re import time import dacite from base.func_epic import is_friday, get_free from base.func_news import News from configuration import Config from gewechat.call_back_message.message import WxMessage, MessageType from gewechat.client import gewe_client from gewechat.response.model.group.chatroom_info import ChatroomInfo from gewechat.response.model.personal.profile import Profile from utils.json_converter import json_to_object from utils.wechat.message_to_db import MessageStorage 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 utils.robot_cmd.robot_command import GroupBotManager from job_mgmt import Job from utils.robot_cmd.robot_command import Feature from utils.robot_cmd.robot_command import PermissionStatus from sehuatang.shehuatang import pdf_file_path from utils.wechat.contact_manager import ContactManager from xiuren.meitu_dl import meitu_dowload_pub_pic from xiuren.xiuren_pdf import generate_pdf_from_images from db.connection import DBConnectionManager from message_util import MessageUtil from db.contacts_db import ContactsDBOperator class Robot(Job): """个性化自己的机器人 """ def __init__(self, config: Config) -> None: self.client = gewe_client.client if not self.client: logging.getLogger("Robot").error("gewe_client.client 不存在，Robot 初始化失败，程序退出。") return self.config = config self.app_id = gewe_client.app_id self.LOG = logging.getLogger("Robot") 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.get_all_contacts() self.contact_manager.set_contacts(self.allContacts) # 获取个人信息 profile_dict = self.client.get_profile(self.app_id) try: profile: Profile = dacite.from_dict(Profile, profile_dict) except Exception as e: self.LOG.info(f"Profile dict 转换失败: {e}") return if profile.data.wxid is None: self.LOG.info(f"获取个人信息失败，退出程序！") return self.wxid = profile.data.wxid # 初始化消息工具类 - 使用联系人管理器 self.message_util = MessageUtil() 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, "client": gewe_client, "event_system": self.event_system, "plugin_registry": self.plugin_registry, "db_pool": self.db_pool, "redis_pool": self.redis_pool, "message_util": self.message_util } 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("插件系统初始化完成") # 消息存档模块初始化，自动完成入库动作 self.message_storage = MessageStorage(self.client) @staticmethod def value_check(args: dict) -> bool: if args: return all(value is not None for key, value in args.items() if key != 'proxy') return False def toChitchat(self, msg: WxMessage) -> bool: """闲聊，接入 ChatGPT """ # 去除@的人和空格等字符 q = re.sub(r"@.*?[\u2005|\s]", "", msg.content.raw_content).replace(" ", "") if q == "#今日百度新闻": self.news_baidu_report((msg.roomid if msg.from_group() else msg.sender)) return True elif q in ["nbc", "cnn", "abc", "fox", "bbc"]: self.news_en_report(q, (msg.roomid if msg.from_group() else msg.sender)) return True else: # 如果是群消息，并且群没开启AI，则不处理该动作 if msg.from_group() and self.gbm.get_group_permission(msg.roomid, Feature.AI_CAPABILITY) == PermissionStatus.ENABLED: resp = self.gbm.get_enabled_features(msg.roomid) self.message_util.send_text(resp, (msg.roomid if msg.from_group() else msg.sender), msg.sender) return True else: return True def processMsg(self, msg: WxMessage) -> None: """当接收到消息的时候，会调用本方法。如果不实现本方法，则打印原始消息。 此处可进行自定义发送的内容,如通过 msg.content 关键字自动获取当前天气信息，并发送到对应的群组@发送者 群号：msg.roomid 微信ID：msg.sender 消息内容：msg.content content = "xx天气信息为：" receivers = msg.roomid self.sendTextMsg(content, receivers, msg.sender) """ try: # 检测群聊是否已加入机器人管理，如果没有则自动添加并开启机器人功能 if msg.from_group() and msg.roomid not in GroupBotManager.local_cache["group_list"]: self.LOG.info(f"检测到新群聊: {msg.roomid}，自动添加到机器人管理列表并启用机器人功能") # 添加群组到列表 GroupBotManager.local_cache["group_list"].add(msg.roomid) # 保存到Redis redis_conn = self.db_manager.get_redis_connection() redis_conn.sadd("group:list", msg.roomid) # 设置ROBOT功能为启用状态 GroupBotManager.set_group_permission(msg.roomid, Feature.ROBOT, PermissionStatus.ENABLED) # 更新联系人信息 # 群第一次加入机器人管理，自动添加并开启机器人功能，需要进行群成员信息初始化。请完成写入数据库，并更新联系人信息 try: chatroom_info = self.client.get_chatroom_info(self.app_id, msg.roomid) self.LOG.info(f"chatroom_info: {chatroom_info}") self.contacts_db.save_chatroom_info(chatroom_info.get('data', {})) self.LOG.info(f"添加新的群信息到数据库成功：{chatroom_info}") except Exception as e: self.LOG.error(f"chatroom_info save error: {e}") return except Exception as e: self.LOG.error(f"加入新群，自动添加并开启机器人功能 error: {e}") # 如果用户信息缓存里面没有这个用户昵称，则添加用户信息，并且维护该用户信息 # 以 wxid 作为唯一标识 try: if msg.from_group(): wxid = msg.sender if wxid and wxid not in self.allContacts: # 添加到数据库 # 这里假设 contacts_db 有 save_contact_info 方法，参数为 dict resp = self.client.get_chatroom_member_detail(self.app_id, msg.roomid, [wxid]) resp_obj = json_to_object(resp) infos = resp_obj.data for info in infos: self.LOG.info(f"已添加新用户信息到数据库: {wxid}") # 更新缓存 self.allContacts[wxid] = info.get("nickName", "nickName") self.contact_manager.set_contacts(self.allContacts) self.LOG.info(f"已维护新用户信息到缓存: {wxid}") self.contacts_db.save_chatroom_member_detail(msg.roomid, infos) except Exception as e: self.LOG.error(f"添加新用户信息到数据库失败: {e}") # # # 发布消息接收事件 # self.event_system.publish(EventType.MESSAGE_RECEIVED, {"message": msg}) # # # 标记插件是否处理了消息 # plugin_processed = False # # # 尝试使用插件处理消息 # if self.process_plugin_message(msg): # plugin_processed = True # # # 群聊消息处理 - 无论插件是否处理过，都执行数据存储 # if msg.from_group(): # # 调用统计逻辑进行聊天数据统计： # try: # self.message_storage.process_message(msg) # except Exception as e: # self.LOG.error(f"process_message error: {e}") # # # 聊天记录入库动作： # try: # self.message_storage.archive_message(msg) # # 单独处理图片消息 # if msg.msg_type == 3: # 图片消息类型 # self.message_storage.process_image(msg) # except Exception as e: # self.LOG.error(f"archive_message error: {e}") # # # 如果插件已处理消息，则不再执行后续的业务逻辑 # if plugin_processed: # return # # # 记录在群里发的最新消息，可以通过撤回指令撤回 # try: # if msg.from_self(): # rsp = self.gbm.handle_command(msg.roomid, msg.content) # # 不在群里发送，防止被骚扰 # if rsp is not None: # self.message_util.send_text(rsp, msg.roomid, msg.sender) # return # except Exception as e: # self.LOG.error(f"revoke_receive_message error: {e}") # # return # 处理完群聊信息，后面就不需要处理了 # # # 如果插件已处理消息，则不再执行后续的业务逻辑 # if plugin_processed: # return # # elif msg.msg_type == MessageType.TEXT: # 文本消息 # # 让配置加载更灵活，自己可以更新配置。也可以利用定时任务更新。 # if msg.from_self(): # if msg.content.clean_content == "^更新$": # self.config.reload() # self.gbm.load_local_cache() # self.LOG.info("已更新") # if msg.content.clean_content == "今日百度新闻": # self.news_baidu_report() # if msg.content.clean_content == "TO_DB": # self.message_count_to_db() # if msg.content.clean_content == "PDF": # self.generate_sehuatang_pdf() # if msg.content.raw_content.startswith("清除群-"): # self.gbm.handle_command(msg.roomid, msg.content.clean_content) # else: # self.toChitchat(msg) # 闲聊 def onMsg(self, msg: WxMessage) -> int: try: self.LOG.info(msg) # 打印信息 self.processMsg(msg) except Exception as e: self.LOG.error(e) return 0 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("联系人信息已刷新") 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: self.message_util.send_text(msg, r) except Exception as e: self.LOG.error(f"send_group_txt_message:{feature.description} error：{e}") 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: self.message_util.send_file(path, r) except Exception as e: self.LOG.error(f"send_group_file_message:{feature.description} error：{e}") def process_plugin_message(self, msg: WxMessage) -> bool: """使用插件处理消息""" # 获取所有消息处理插件 message_plugins = self.plugin_registry.get_plugins_by_type(MessagePluginInterface) # 依次尝试处理消息 for plugin in message_plugins: if plugin.status != PluginStatus.RUNNING: continue try: # 转换WxMessage为插件可处理的格式 plugin_msg = { "type": msg.msg_type, "content": msg.content.clean_content, "sender": msg.sender, "roomid": msg.roomid if msg.from_group() else "", "xml": msg.content.xml_content, "is_at": msg.is_at(self.wxid), # 确保正确设置is_at标志 "timestamp": time.time(), "message_util": self.message_util, # 提供消息工具类 "gbm": self.gbm, # 每次从程序变量中取，保证最新 "all_contacts": self.allContacts, "full_wx_msg": msg } # 检查插件是否可以处理该消息 if plugin.can_process(plugin_msg): processed, _ = 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 news_baidu_report_auto(self) -> None: try: news = News().get_baidu_news() self.send_group_txt_message(news, Feature.DAILY_NEWS) except Exception as e: self.LOG.error(f"newsBaiduReportAuto error：{e}") def news_baidu_report(self, sender: str = None) -> None: try: news = News().get_baidu_news() if news and isinstance(news, str): self.message_util.send_text(news, sender) else: self.LOG.error("获取百度新闻返回值异常") except Exception as e: self.LOG.error(f"newsBaiduReport error：{e}") # 发送错误信息给用户，让用户知道发生了什么 def news_en_report(self, website, sender: str = None) -> None: try: news = News().get_eng_news(website) self.message_util.send_text(news, sender) except Exception as e: self.LOG.error(f"newsEnReport error：{e}") # 使用装饰器标记定时任务 星期五 10:30 执行 def send_epic_free_games(self): try: if is_friday(): games = get_free() self.send_group_txt_message(games, Feature.EPIC) except Exception as e: self.LOG.error(f"sendEpicFreeGames error：{e}") # 使用装饰器标记定时任务 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}") def generate_sehuatang_pdf(self): try: self.LOG.info("开始生成PDF,generate_sehuatang_pdf") path = pdf_file_path() # 暂时只发4K群 self.send_group_file_message(path, Feature.PDF_CAPABILITY) except Exception as e: self.LOG.error(f"generateSehuatangPdf error：{e}") 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) self.message_util.send_text(output, r) except Exception as e: self.LOG.error(f"SendRanking error：{e}") # # # 设置定时任务 # def game_auto_tasks(self): # try: # group_ids = get_group_ids() # for gid in group_ids: # if self.gbm.get_group_permission(gid, Feature.TASK_GAME) == PermissionStatus.ENABLED: # rep = run_random_task_assignment(group_id=gid) # message = rep["message"] # player_id = rep["player_id"] # print(f"消息: {message}") # print(f"玩家ID: {player_id}") # self.send_text_msg(message, gid, player_id) # except Exception as e: # self.LOG.error(f"message_summary_robot error：{e}") 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}") def xiu_ren_pdf_send(self): try: pub_path = generate_pdf_from_images("xiuren") self.message_util.send_file(pub_path, "45317011307@chatroom") except Exception as e: self.LOG.error(f"xiu_ren_pdf_send error：{e}") # 本逻辑主要解决加载联系人信息的问题，只从数据库里面提取，不完成下载行为。 def get_all_contacts(self) -> dict: """获取所有联系人信息并返回字典格式 {wxid: nickname}""" # 从数据库提取信息，如果数据库没内容，则完成第一次初始化。 try: # 先尝试从数据库获取联系人信息 contacts_dict = self.contacts_db.get_all_contacts() # 获取群成员列表 return contacts_dict except Exception as e: self.LOG.error(f"获取联系人信息失败: {e}") return {} def sync_all_contacts(self): """同步所有联系人信息""" try: # 数据库中没有联系人信息，需要初始化 self.LOG.info("数据库中没有联系人信息，开始初始化...") contacts_dict = {} # 获取所有联系人列表 response = self.client.fetch_contacts_list(self.app_id) self.LOG.info(f"获取联系人列表响应: {response}") if not response or response.get("ret") != 200: self.LOG.warning(f"获取联系人列表失败: {response}") return contacts_dict # 从响应中提取联系人数据 contact_data = response.get("data", {}) # 处理好友列表 friends = contact_data.get("friends", []) for wxid in friends: contacts_dict[wxid] = wxid # 默认使用wxid作为昵称 # 处理群聊列表 chatrooms = contact_data.get("chatrooms", []) for chatroom_id in chatrooms: contacts_dict[chatroom_id] = chatroom_id # 如果是群聊，则获取群成员信息 self.update_chatroom_member_details(chatroom_id) # 处理公众号列表 ghs = contact_data.get("ghs", []) for gh_id in ghs: contacts_dict[gh_id] = gh_id # 获取联系人详细信息（昵称等） self.update_contact_details(contacts_dict) self.LOG.info(f"成功获取并保存{len(contacts_dict)}个联系人信息") return contacts_dict except Exception as e: self.LOG.error(f"获取联系人信息失败: {e}") return {} def update_contact_details(self, contacts_dict): """更新联系人详细信息（昵称等）""" try: # 将wxid列表分批处理，每批50个 batch_size = 10 wxids = list(contacts_dict.keys()) for i in range(0, len(wxids), batch_size): batch_wxids = wxids[i:i + batch_size] # 批量获取联系人详细信息 contact_info = self.client.get_detail_info(self.app_id, batch_wxids) self.LOG.info(f"获取联系人详细信息响应: {contact_info}") # 处理返回的数据 if contact_info and contact_info.get("ret") == 200 and "data" in contact_info: contact_data = contact_info.get("data", []) if contact_data: for contact in contact_data: user_name = contact.get("userName") if not user_name or user_name not in contacts_dict: continue # 更新昵称 contacts_dict[user_name] = contact.get("nickName") or user_name try: # 判断联系人类型 contact_type = "friends" # 默认为好友类型 if user_name.endswith("@chatroom"): contact_type = "chatrooms" elif user_name.startswith("gh_"): contact_type = "ghs" # 保存到数据库 self.contacts_db.save_contacts([contact], contact_type) except Exception as e: self.LOG.error(f"处理联系人 {user_name} 失败: {e}") continue else: self.LOG.error(f"获取联系人详情失败: {contact_info}") except Exception as e: self.LOG.error(f"更新联系人详细信息失败: {e}") def update_chatroom_member_details(self, chatroom_id): """更新群成员详细信息""" try: # 首先获取群成员列表 members_response = self.client.get_chatroom_member_list(self.app_id, chatroom_id) if members_response and members_response.get('ret') == 200: member_list = members_response.get('data', {}).get('memberList', []) # 提取成员wxid列表 member_wxids = [member.get('wxid') for member in member_list if member.get('wxid')] if member_wxids: # 按照官方接口格式传递参数 details_response = self.client.get_chatroom_member_detail( self.app_id, chatroom_id, member_wxids # 直接传递列表，不需要转换为集合 ) success = self.contacts_db.process_chatroom_member_detail_response(chatroom_id, details_response) if success: self.LOG.info(f"成功更新群聊{chatroom_id}的成员详细信息") else: self.LOG.error(f"更新群聊{chatroom_id}的成员详细信息失败") return success else: self.LOG.warning(f"群聊{chatroom_id}没有成员") return False else: self.LOG.error(f"获取群聊{chatroom_id}成员列表失败") return False except Exception as e: self.LOG.error(f"更新群聊成员详细信息出错: {e}")