# -*- coding: utf-8 -*- import logging import re import time import xml.etree.ElementTree as ET from queue import Empty from threading import Thread import random from base.func_doubao import Doubao from base.func_epic import is_friday, get_free from base.func_zhipu import ZhiPu from wcferry import Wcf, WxMsg from base.func_bard import BardAssistant from base.func_chatglm import ChatGLM from base.func_chatgpt import ChatGPT from base.func_news import News from base.func_tigerbot import TigerBot from base.func_xinghuo_web import XinghuoWeb from base.func_claude import Claude from configuration import Config from constants import ChatType from game_task.game_task_encyclopedia import game_process_message, get_group_ids,run_random_task_assignment from group_add.main import GroupAdd from group_auto.group_auto_invite import get_first_group_id, process_command from group_auto.group_member_change import GroupMemberChange from message_storage.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 robot_cmd.robot_command import GroupBotManager from job_mgmt import Job from robot_cmd.robot_command import Feature from robot_cmd.robot_command import PermissionStatus __version__ = "39.2.4.0" from sehuatang.shehuatang import pdf_file_path from xiuren.meitu_dl import meitu_dowload_pub_pic from xiuren.random_pic import get_xiuren_heisi_pic from xiuren.xiuren_pdf import generate_pdf_from_images from db.connection import DBConnectionManager from message_util import MessageUtil # 在文件顶部导入装饰器 from job_decorators import scheduled_job, register_scheduled_jobs class Robot(Job): """个性化自己的机器人 """ def __init__(self, config: Config, wcf: Wcf, chat_type: int) -> None: self.wcf = wcf self.config = config self.LOG = logging.getLogger("Robot") self.wxid = self.wcf.get_self_wxid() self.allContacts = self.get_all_contacts() # 修改初始化方法中的这一部分 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.message_util = MessageUtil(wcf, 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, "wcf": wcf, "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.gmc = GroupMemberChange(wcf, self.redis_pool) # 加群测试 self.group_add = GroupAdd(wcf, self.gbm) # 在初始化结束时注册所有被装饰的定时任务 register_scheduled_jobs(self) if ChatType.is_in_chat_types(chat_type): if chat_type == ChatType.TIGER_BOT.value and TigerBot.value_check(self.config.TIGERBOT): self.chat = TigerBot(self.config.TIGERBOT) elif chat_type == ChatType.CHATGPT.value and ChatGPT.value_check(self.config.CHATGPT): self.chat = ChatGPT(self.config.CHATGPT) elif chat_type == ChatType.XINGHUO_WEB.value and XinghuoWeb.value_check(self.config.XINGHUO_WEB): self.chat = XinghuoWeb(self.config.XINGHUO_WEB) elif chat_type == ChatType.CHATGLM.value and ChatGLM.value_check(self.config.CHATGLM): self.chat = ChatGLM(self.config.CHATGLM) elif chat_type == ChatType.BardAssistant.value and BardAssistant.value_check(self.config.BardAssistant): self.chat = BardAssistant(self.config.BardAssistant) elif chat_type == ChatType.ZhiPu.value and ZhiPu.value_check(self.config.ZhiPu): self.chat = ZhiPu(self.config.ZhiPu) elif chat_type == ChatType.CLAUDE.value and Claude.value_check(self.config.CLAUDE): self.chat = Claude(self.config.CLAUDE) elif chat_type == ChatType.DOUBAO.value and Claude.value_check(self.config.DOUBAO): self.chat = Doubao(self.config.DOUBAO) else: self.LOG.warning("未配置模型") self.chat = None else: if TigerBot.value_check(self.config.TIGERBOT): self.chat = TigerBot(self.config.TIGERBOT) elif ChatGPT.value_check(self.config.CHATGPT): self.chat = ChatGPT(self.config.CHATGPT) elif XinghuoWeb.value_check(self.config.XINGHUO_WEB): self.chat = XinghuoWeb(self.config.XINGHUO_WEB) elif ChatGLM.value_check(self.config.CHATGLM): self.chat = ChatGLM(self.config.CHATGLM) elif BardAssistant.value_check(self.config.BardAssistant): self.chat = BardAssistant(self.config.BardAssistant) elif ZhiPu.value_check(self.config.ZhiPu): self.chat = ZhiPu(self.config.ZhiPu) elif Claude.value_check(self.config.CLAUDE): self.chat = Claude(self.config.CLAUDE) elif Doubao.value_check(self.config.DOUBAO): self.chat = Doubao(self.config.DOUBAO) else: self.LOG.warning("未配置模型") self.chat = None self.LOG.info(f"已选择: {self.chat}") @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 toAt(self, msg: WxMsg) -> bool: """处理被 @ 消息 :param msg: 微信消息结构 :return: 处理状态，`True` 成功，`False` 失败 """ return self.toChitchat(msg) def toChitchat(self, msg: WxMsg) -> bool: """闲聊，接入 ChatGPT """ if not self.chat: # 没接 ChatGPT，固定回复 rsp = "你@我干嘛？" else: # 接了 ChatGPT，智能回复 # 去除@的人和空格等字符 q = re.sub(r"@.*?[\u2005|\s]", "", msg.content).replace(" ", "") # 使用正则表达式匹配加群指令 pattern = r'#加群:\[(.*?)\]' # 匹配 #加群:[<任何内容>] match = re.match(pattern, q) # 所有人员都可以要求他撤回刚刚的信息 if msg.from_group() and q.startswith("/"): # 进行权限判断 加入权限，防止tokens浪费 if self.gbm.get_group_permission(msg.roomid, Feature.TASK_GAME) == PermissionStatus.DISABLED: return True else: try: # 因为内容中存在空格指令，所以不能使用q game_message = re.sub(r"@.*?[\u2005|\s]", "", msg.content) self.LOG.info(f"msg.content:{msg.content}\n game_message: {game_message}") resp = game_process_message(group_id=msg.roomid, player_id=msg.sender, message=game_message, player_name=self.allContacts.get(msg.sender, msg.sender)) message = resp["message"] player_id = resp["player_id"] print(f"消息: {message}") print(f"玩家ID: {player_id}") self.send_text_msg(message, msg.roomid, msg.sender) except Exception as e: self.LOG.error(f"game_message_load error：{e}") return True 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 # 如果正则匹配到时加群指令，则从库中提取第一个群ID elif match: try: group_id = get_first_group_id(match.group(1)) self.LOG.info(f"邀请加入{match.group(1)}群，ID：{group_id}{msg.sender}") self.wcf.invite_chatroom_members(group_id, msg.sender) except Exception as e: self.LOG.error(f"邀请加入群出错：{e}") return True else: # 如果是群消息，并且群没开启AI，则不处理该动作 if msg.from_group() and self.gbm.get_group_permission(msg.roomid, Feature.AI_CAPABILITY) == PermissionStatus.DISABLED: return True else: if msg.type == 1: # 只处理类型为1的消息提供的问题，引用@不予以对话 rsp = self.chat.get_answer(q, (msg.roomid if msg.from_group() else msg.sender)) else: return True if rsp: if msg.from_group(): self.send_text_msg(rsp, msg.roomid, msg.sender) else: self.send_text_msg(rsp, msg.sender) return True else: self.LOG.error(f"无法从 ChatGPT 获得答案") return False def processMsg(self, msg: WxMsg) -> None: """当接收到消息的时候，会调用本方法。如果不实现本方法，则打印原始消息。 此处可进行自定义发送的内容,如通过 msg.content 关键字自动获取当前天气信息，并发送到对应的群组@发送者 群号：msg.roomid 微信ID：msg.sender 消息内容：msg.content content = "xx天气信息为：" receivers = msg.roomid self.sendTextMsg(content, receivers, msg.sender) """ # 发布消息接收事件 self.event_system.publish(EventType.MESSAGE_RECEIVED, {"message": msg}) # 尝试使用插件处理消息 if self.process_plugin_message(msg): return # 如果没有插件处理，使用原有逻辑处理消息 # 群聊消息 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) # self.LOG.info(f"msg.xml:{msg.xml}") except Exception as e: self.LOG.error(f"archive_message error: {e}") # 记录在群里发的最新消息，可以通过撤回指令撤回 try: if msg.from_self(): rsp = self.gbm.handle_command(msg.roomid, msg.content) # 不在群里发送，防止被骚扰 if rsp is not None: self.send_text_msg(rsp, msg.sender) return except Exception as e: self.LOG.error(f"revoke_receive_message error: {e}") # 兼容不@ 直接/触发指令，回答问题。 try: if msg.content.startswith("/"): # 进行权限判断 加入权限，防止tokens浪费 if self.gbm.get_group_permission(msg.roomid, Feature.TASK_GAME) == PermissionStatus.DISABLED: return else: # 因为内容中存在空格指令，所以不能使用 self.LOG.info(f"msg.content:{msg.content}\n game_message: {msg.content}") resp = game_process_message(group_id=msg.roomid, player_id=msg.sender, message=msg.content, player_name=self.allContacts.get(msg.sender, msg.sender)) message = resp["message"] player_id = resp["player_id"] print(f"消息: {message}") print(f"玩家ID: {player_id}") self.send_text_msg(message, msg.roomid, msg.sender) return except Exception as e: self.LOG.error(f"game_message_load error：{e}") try: # if msg.type == 10000: # self.LOG.info(msg) # self.group_add.handle_message(msg) # return result = self.gmc.process_message(msg.roomid, msg.xml) # 判断是否没有变化 if "$NO_CHANGE$" not in result: self.LOG.info(f"检测到群成员变化，进行相关内容输出:{result}") self.send_text_msg(result, msg.roomid) except Exception as e: self.LOG.error(f"group_member_change error: {e}") if msg.is_at(self.wxid): # 被@ self.toAt(msg) return # 处理完群聊信息，后面就不需要处理了 # 非群聊信息，按消息类型进行处理 if msg.type == 37: # 好友请求 self.LOG.info(f"收到好友请求：{msg}") self.auto_accept_friend_request(msg) elif msg.type == 10000: # 系统信息 self.say_hi_to_new_friend(msg) elif msg.type == 0x01: # 文本消息 # 让配置加载更灵活，自己可以更新配置。也可以利用定时任务更新。 if msg.from_self(): if msg.content == "^更新$": self.config.reload() self.gbm.load_local_cache() self.LOG.info("已更新") if msg.content == "今日百度新闻": self.news_baidu_report() if msg.content.startswith("清除群-"): self.gbm.handle_command(msg.roomid, msg.content) if msg.content == 'GROUP_LIST': self.send_text_msg(self.gbm.get_group_list(), msg.sender) if msg.content.startswith('#加群配置'): # msg_content = "# 加群配置|add 原生鱼 xxx@room" parts = msg.content.split('|') resp = "" if len(parts) > 1: after_pipe = parts[1] resp = process_command(after_pipe) else: resp = process_command("help") self.send_text_msg(resp, msg.sender) self.send_text_msg(f"指令：{msg.content} 已执行", msg.sender) else: self.toChitchat(msg) # 闲聊 def onMsg(self, msg: WxMsg) -> int: try: self.LOG.debug(msg) # 打印信息 self.processMsg(msg) except Exception as e: self.LOG.error(e) return 0 def enableRecvMsg(self) -> None: self.wcf.enable_recv_msg(self.onMsg) def enableReceivingMsg(self) -> None: def innerProcessMsg(wcf: Wcf): while wcf.is_receiving_msg(): try: msg = wcf.get_msg() self.LOG.debug(msg) self.processMsg(msg) except Empty: continue # Empty message except Exception as e: self.LOG.error(f"Receiving message error: {e}") self.wcf.enable_receiving_msg() Thread(target=innerProcessMsg, name="GetMessage", args=(self.wcf,), daemon=True).start() def send_text_msg(self, msg: str, receiver: str, at_list: str = "") -> None: """ 发送消息 :param msg: 消息字符串 :param receiver: 接收人wxid或者群id :param at_list: 要@的wxid, @所有人的wxid为：notify@all """ # msg 中需要有 @ 名单中一样数量的 @ # 风控处理，随机延迟发送，解决群消息高频发送导致的微信风险 time.sleep(random.uniform(0.3, 1.0)) ats = "" if at_list: if at_list == "notify@all": # @所有人 ats = " @所有人" else: wxids = at_list.split(",") for wxid in wxids: # 根据 wxid 查找群昵称 ats += f" @{self.wcf.get_alias_in_chatroom(wxid, receiver)}" # {msg}{ats} 表示要发送的消息内容后面紧跟@，例如 北京天气情况为：xxx @张三 if ats == "": self.LOG.info(f"To {receiver}: {msg}") self.wcf.send_text(f"{msg}", receiver, at_list) else: self.LOG.info(f"To {receiver}: {ats}\r{msg}") self.wcf.send_text(f"{ats}\n\n{msg}", receiver, at_list) def get_all_contacts(self) -> dict: """ 获取联系人（包括好友、公众号、服务号、群成员……） 格式: {"wxid": "NickName"} """ contacts = self.wcf.query_sql("MicroMsg.db", "SELECT UserName, NickName FROM Contact;") return {contact["UserName"]: contact["NickName"] for contact in contacts} def keep_running_and_block_process(self) -> None: """ 保持机器人运行，不让进程退出 """ while True: self.runPendingJobs() time.sleep(1) def auto_accept_friend_request(self, msg: WxMsg) -> None: try: xml = ET.fromstring(msg.content) v3 = xml.attrib["encryptusername"] v4 = xml.attrib["ticket"] scene = int(xml.attrib["scene"]) res = self.wcf.accept_new_friend(v3, v4, scene) self.LOG.info(f"同意好友请求：{res}") except Exception as e: self.LOG.error(f"同意好友出错：{e}") def say_hi_to_new_friend(self, msg: WxMsg) -> None: nickName = re.findall(r"你已添加了(.*)，现在可以开始聊天了。", msg.content) if nickName: # 添加了好友，更新好友列表 self.allContacts[msg.sender] = nickName[0] self.send_text_msg(f"Hi {nickName[0]}，我自动通过了你的好友请求。", msg.sender) 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.send_text_msg(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.wcf.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: WxMsg) -> bool: """使用插件处理消息""" # 获取所有消息处理插件 message_plugins = self.plugin_registry.get_plugins_by_type(MessagePluginInterface) # 依次尝试处理消息 for plugin in message_plugins: if plugin.status != PluginStatus.RUNNING: continue try: # 转换WxMsg为插件可处理的格式 plugin_msg = { "type": msg.type, "content": msg.content, "sender": msg.sender, "roomid": msg.roomid if msg.from_group() else "", "xml": msg.xml, "is_at": msg.is_at(self.wxid), "timestamp": time.time(), "wcf": self.wcf, # 提供wcf对象，让插件可以直接发送消息 "message_util": self.message_util, # 提供消息工具类 "gbm": self.gbm, # 每次从程序变量中取，保证最新 "all_contacts": self.allContacts } # 检查插件是否可以处理该消息 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 # ============================================== 业务内容========================================================== @scheduled_job(cron="0 0 8 * * *", name="每日新闻推送") 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() self.send_text_msg(news, sender) 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.send_text_msg(news, sender) except Exception as e: self.LOG.error(f"newsEnReport error：{e}") # 使用装饰器标记定时任务 星期五 10:30 执行 @scheduled_job(cron="0 30 10 * * 5", name="Epic免费游戏推送") 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}") # 使用装饰器标记定时任务 @scheduled_job(cron="0 0 * * * *", name="消息统计入库") 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}") @scheduled_job(cron="0 0 15 * * *", name="发送色花堂") def generate_sehuatang_pdf(self): try: 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}") @scheduled_job(cron="0 30 9 * * *", name="发送消息排行榜") 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.send_text_msg(output, r) except Exception as e: self.LOG.error(f"SendRanking error：{e}") # 设置定时任务 @scheduled_job(cron="0 0 18 * * *", name="每天发一个游戏任务") 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}") @scheduled_job(cron="0 30 1 * * *", name="每天下载10组图") 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}") @scheduled_job(cron="0 30 17 * * *", name="发送秀人PDF") def xiu_ren_pdf_send(self): try: pub_path = generate_pdf_from_images("xiuren") self.wcf.send_file(pub_path, "45317011307@chatroom") except Exception as e: self.LOG.error(f"xiu_ren_pdf_send error：{e}")