main.py
! /usr/bin/env python3
-- coding: utf-8 --
import signal
from argparse import ArgumentParser
from base.func_report_reminder import ReportReminder
from configuration import Config
from constants import ChatType
from robot import Robot, version
from wcferry import Wcf
def weather_report(robot: Robot) -> None:
"""模拟发送天气预报
"""
获取接收人
receivers = ["filehelper"]
获取天气，需要自己实现，可以参考 https://gitee.com/lch0821/WeatherScrapy 获取天气。
report = "这就是获取到的天气情况了"
for r in receivers:
robot.sendTextMsg(report, r)
# robot.sendTextMsg(report, r, "notify@all") # 发送消息并@所有人
def main(chat_type: int):
config = Config()
wcf = Wcf(debug=True)
def handler(sig, frame):
wcf.cleanup() # 退出前清理环境
exit(0)
signal.signal(signal.SIGINT, handler)
robot = Robot(config, wcf, chat_type)
robot.LOG.info(f"WeChatRobot【{version}】成功启动···")
机器人启动发送测试消息
robot.sendTextMsg("机器人启动成功！", "filehelper")
接收消息
robot.enableRecvMsg() # 可能会丢消息？
robot.enableReceivingMsg() # 加队列
每天 7 点发送天气预报
robot.onEveryTime("07:00", weather_report, robot=robot)
每天 7:30 发送新闻
robot.onEveryTime("07:30", robot.newsReport)
每天 16:30 提醒发日报周报月报
robot.onEveryTime("16:30", ReportReminder.remind, robot=robot)
让机器人一直跑
robot.keepRunningAndBlockProcess()
if name == "main":
parser = ArgumentParser()
parser.add_argument('-c', type=int, default=0, help=f'选择模型参数序号: {ChatType.help_hint()}')
args = parser.parse_args().c
main(args)
robot.py
-- coding: utf-8 --
import logging # 导入日志模块，用于记录程序运行信息方便调试查看状态
import re # 导入正则表达式模块，用于文本基于特定模式的匹配等操作
import time # 导入时间模块，可获取时间信息、进行时间延迟等
import xml.etree.ElementTree as ET # 导入用于解析处理XML数据的模块并简称为ET
from queue import Empty # 从queue模块导入Empty类，用于判断队列是否为空
from threading import Thread # 从threading模块导入Thread类，用于创建管理线程实现多线程编程
from base.func_zhipu import ZhiPu # 从base.func_zhipu模块导入ZhiPu类（功能看对应模块定义，可能与某功能相关）
from wcferry import Wcf, WxMsg # 从wcferry模块导入Wcf和WxMsg类，可能涉及微信相关功能实现（如消息处理等）
from base.func_bard import BardAssistant # 从base.func_bard模块导入BardAssistant类（推测与调用Bard相关服务辅助功能有关）
from base.func_chatglm import ChatGLM # 从base.func_chatglm模块导入ChatGLM类（可能调用ChatGLM智能对话功能）
from base.func_chatgpt import ChatGPT # 从base.func_chatgpt模块导入ChatGPT类（大概率用于调用ChatGPT对话交互等功能）
from base.func_chengyu import cy # 从base.func_chengyu模块导入cy（可能与成语相关功能，看模块具体内容）
from base.func_news import News # 从base.func_news模块导入News类（可能用于获取新闻资讯等功能）
from base.func_tigerbot import TigerBot # 从base.func_tigerbot模块导入TigerBot类（可能调用TigerBot服务实现功能）
from base.func_xinghuo_web import XinghuoWeb # 从base.func_xinghuo_web模块导入XinghuoWeb类（推测和调用星火网页端相关功能有关）
from configuration import Config # 从configuration模块导入Config类（可能用于读取配置信息等操作）
from constants import ChatType # 从constants模块导入ChatType（通常定义聊天类型等常量）
from job_mgmt import Job # 从job_mgmt模块导入Job类（可能用于管理任务相关功能，如调度等）
version = "39.2.4.0"
class Robot(Job):
"""个性化自己的机器人
"""
def init(self, config: Config, wcf: Wcf, chat_type: int) -> None:
# 类的构造函数，用于初始化对象属性，参数config是配置对象，wcf可能用于外部交互，chat_type表示聊天类型
self.wcf = wcf # 将传入的wcf对象赋值给实例变量self.wcf，方便后续使用
self.config = config # 将传入的config对象赋值给self.config，用于获取配置参数
self.LOG = logging.getLogger("Robot") # 获取名为"Robot"的日志记录器用于记录日志
self.wxid = self.wcf.get_self_wxid() # 获取自身微信ID（假设和微信相关）并赋值给self.wxid
self.allContacts = self.getAllContacts() # 获取所有联系人信息赋值给self.allContacts
if ChatType.is_in_chat_types(chat_type): # 判断chat_type是否在合法聊天类型范围内
if chat_type == ChatType.TIGER_BOT.value and TigerBot.value_check(self.config.TIGERBOT):
self.chat = TigerBot(self.config.TIGERBOT) # 创建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)
else:
self.LOG.warning("未配置模型") # 若不满足上述情况，记录警告日志表示未配置模型
self.chat = None # 将self.chat赋值为None
else: # 如果chat_type不在合法范围
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)
else:
self.LOG.warning("未配置模型")
self.chat = None
self.LOG.info(f"已选择: {self.chat}") # 记录信息日志输出已选择的聊天模型情况
@staticmethod
# 定义一个静态方法，意味着可以通过类名直接调用该方法，而不需要实例化类对象
def value_check(args: dict) -> bool:
# 方法名为value_check，接收一个字典类型的参数args，返回值为布尔类型
if args:
# 如果传入的字典args不为空（即有元素）
return all(value is not None for key, value in args.items() if key != 'proxy')
# 使用生成器表达式和all函数进行判断，对于args字典中除了键为'proxy'之外的所有键值对，检查值是否不为None，若所有满足条件的值都不为None则返回True，否则返回False
return False
# 如果传入的字典args为空，则直接返回False
def toAt(self, msg: WxMsg) -> bool:
"""
函数名为toAt，是一个实例方法，用于处理被 @ 消息。
参数self表示类的实例本身，msg是WxMsg类型的参数，代表微信消息结构，通过它可以获取消息相关的各种信息，比如发送者、内容等。
返回值为布尔类型，True表示处理成功，False表示处理失败，以此来告知调用者该消息处理的最终状态。
"""
return self.toChitchat(msg)
# 在这个方法中，直接调用了实例的toChitchat方法，
# 并将接收到的微信消息msg作为参数传递过去，然后返回toChitchat方法的执行结果，
# 意味着处理被 @ 消息的逻辑委托给了toChitchat方法来执行。
def toChengyu(self, msg: WxMsg) -> bool:
"""
函数名为toChengyu，是一个实例方法，用于处理成语查询/接龙消息。接收WxMsg类型的msg参数代表微信消息结构，返回布尔类型表示处理状态，True为成功，False为失败。
"""
status = False
# 初始化处理状态为False，表示默认未成功处理
texts = re.findall(r"^([#|?|？])(.*)$", msg.content)
# 使用正则表达式从消息内容中查找匹配的内容，期望提取出以#、?或？开头的内容，结果格式类似[('#', '天天向上')]
if texts:
# 如果找到了匹配的内容
flag = texts[0][0]
# 获取匹配内容中的第一个元素，即开头的标志字符（#、?或？）
text = texts[0][1]
# 获取匹配内容中的第二个元素，即具体的文本内容（比如成语相关的文本）
if flag == "#": # 接龙
# 如果标志字符为#，表示是成语接龙操作
if cy.isChengyu(text):
# 调用cy模块（可能是处理成语相关功能的模块）的isChengyu方法判断text是否是成语
rsp = cy.getNext(text)
# 如果是成语，则调用cy模块的getNext方法获取下一个成语（实现成语接龙的逻辑）
if rsp:
# 如果获取到了下一个成语
self.sendTextMsg(rsp, msg.roomid)
# 调用实例的sendTextMsg方法将获取到的成语发送到对应的房间（msg.roomid指定房间）
status = True
# 将处理状态更新为True，表示此次成语接龙处理成功
elif flag in ["?", "？"]: # 查词
# 如果标志字符是?或？，表示是成语查询操作
if cy.isChengyu(text):
# 同样先判断是否是成语
rsp = cy.getMeaning(text)
# 如果是成语，调用cy模块的getMeaning方法获取成语的含义
if rsp:
# 如果获取到了含义
self.sendTextMsg(rsp, msg.roomid)
# 将含义发送到对应的房间
status = True
# 更新处理状态为True，表示成语查询处理成功
return status
# 最后返回处理状态，告知调用者成语相关处理是否成功
def toChitchat(self, msg: WxMsg) -> bool:
"""
函数名为toChitchat，是一个实例方法，用于处理闲聊场景，这里涉及接入ChatGPT来生成回复内容。接收WxMsg类型的msg参数，代表微信消息结构，返回布尔类型表示处理状态，True表示处理成功，False表示处理失败。
"""
if not self.chat: # 没接ChatGPT，固定回复
# 判断self.chat是否为None，如果为None意味着没有接入ChatGPT（可能配置等原因没启用相关服务）
rsp = "你@我干嘛？"
# 这种情况下设置一个固定的回复内容
else: # 接了ChatGPT，智能回复
# 如果self.chat不为None，说明接入了ChatGPT，走智能回复逻辑
q = re.sub(r"@.*?[\u2005|\s]", "", msg.content).replace(" ", "")
# 使用正则表达式去除消息内容中以@开头直到遇到空白字符或特定空白符（如\u2005）的部分，然后再去除剩余内容中的空格，得到处理后的提问内容q
rsp = self.chat.get_answer(q, (msg.roomid if msg.from_group() else msg.sender))
# 调用self.chat（应该是对应聊天服务的实例对象）的get_answer方法，传入处理后的提问q以及根据消息是否来自群组来决定传入群组ID或者发送者信息，获取回复内容rsp
if rsp:
# 如果获取到了回复内容（即rsp不为空）
if msg.from_group():
# 如果消息是来自群组的
self.sendTextMsg(rsp, msg.roomid, msg.sender)
# 调用实例的sendTextMsg方法，将回复内容rsp发送到对应的群组（msg.roomid指定群组），并指定消息的发送者信息（msg.sender），可能用于在群组中显示回复来源等情况
else:
self.sendTextMsg(rsp, msg.sender)
# 如果消息不是来自群组（即私聊消息），则将回复内容发送给对应的发送者（msg.sender）
return True
# 表示消息处理成功，成功发送了回复内容，返回True告知调用者
else:
self.LOG.error(f"无法从ChatGPT获得答案")
# 如果没有获取到回复内容，记录一条错误日志，表示无法从ChatGPT获取到答案
return False
# 返回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)
"""
群聊消息
if msg.from_group():
# 如果在群里被 @
if msg.roomid not in self.config.GROUPS: # 不在配置的响应的群列表里，忽略
return
if msg.is_at(self.wxid): # 被@
self.toAt(msg)
else: # 其他消息
self.toChengyu(msg)
return # 处理完群聊信息，后面就不需要处理了
非群聊信息，按消息类型进行处理
if msg.type == 37: # 好友请求
self.autoAcceptFriendRequest(msg)
elif msg.type == 10000: # 系统信息
self.sayHiToNewFriend(msg)
elif msg.type == 0x01: # 文本消息
# 让配置加载更灵活，自己可以更新配置。也可以利用定时任务更新。
if msg.from_self():
if msg.content == "^更新$":
self.config.reload()
self.LOG.info("已更新")
else:
self.toChitchat(msg) # 闲聊
def onMsg(self, msg: WxMsg) -> int:
try:
self.LOG.info(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.info(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 sendTextMsg(self, msg: str = "Hello World", receiver: str = "WANWUGY", at_list: str = "") -> None:
""" 发送消息
:param msg: 消息字符串
:param receiver: 接收人wxid或者群id
:param at_list: 要@的wxid, @所有人的wxid为：notify@all
"""
# msg 中需要有 @ 名单中一样数量的 @
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 getAllContacts(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 keepRunningAndBlockProcess(self) -> None:
"""
保持机器人运行，不让进程退出
"""
while True:
self.runPendingJobs()
time.sleep(1)
def autoAcceptFriendRequest(self, msg: WxMsg) -> None:
try:
xml = ET.fromstring(msg.content)
v3 = xml.attrib["encryptusername"]
v4 = xml.attrib["ticket"]
scene = int(xml.attrib["scene"])
self.wcf.accept_new_friend(v3, v4, scene)
except Exception as e:
self.LOG.error(f"同意好友出错：{e}")
def sayHiToNewFriend(self, msg: WxMsg) -> None:
nickName = re.findall(r"你已添加了(.*)，现在可以开始聊天了。", msg.content)
if nickName:
# 添加了好友，更新好友列表
self.allContacts[msg.sender] = nickName[0]
self.sendTextMsg(f"Hi {nickName[0]}，我自动通过了你的好友请求。", msg.sender)
def newsReport(self) -> None:
receivers = self.config.NEWS
if not receivers:
return
news = News().get_important_news()
for r in receivers:
self.sendTextMsg(news, r)