Files
abot/robot.py
2025-03-27 16:12:38 +08:00

574 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- 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 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 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
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.contact_manager = ContactManager.get_instance()
self.allContacts = self.get_all_contacts()
self.contact_manager.set_contacts(self.allContacts)
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.contact_manager)
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()
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(" ", "")
# 所有人员都可以要求他撤回刚刚的信息
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
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 微信IDmsg.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)
except Exception as e:
self.LOG.error(f"加入新群,自动添加并开启机器人功能 error: {e}")
# 发布消息接收事件
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)
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}")
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 == "TO_DB":
self.message_count_to_db()
if msg.content == "PDF":
self.generate_sehuatang_pdf()
if msg.content.startswith("清除群-"):
self.gbm.handle_command(msg.roomid, msg.content)
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.contact_manager.update_contact(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
# ============================================== 业务内容==========================================================
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 执行
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:
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.send_text_msg(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.wcf.send_file(pub_path, "45317011307@chatroom")
except Exception as e:
self.LOG.error(f"xiu_ren_pdf_send error{e}")
# 添加一个方法用于刷新联系人信息
def refresh_contacts(self):
"""刷新联系人信息"""
self.allContacts = self.get_all_contacts()
self.contact_manager.refresh_contacts(self.allContacts)
self.LOG.info("联系人信息已刷新")