From 8e224aab0e10cd03620436e86598cdefbc71b371 Mon Sep 17 00:00:00 2001 From: zxc <3425933825@qq.com> Date: Wed, 22 Oct 2025 15:12:41 +0800 Subject: [PATCH] Update robot.py --- src/DjangoBlog-master/servermanager/robot.py | 237 +++++++++++-------- 1 file changed, 142 insertions(+), 95 deletions(-) diff --git a/src/DjangoBlog-master/servermanager/robot.py b/src/DjangoBlog-master/servermanager/robot.py index 7b45736..160b9b7 100644 --- a/src/DjangoBlog-master/servermanager/robot.py +++ b/src/DjangoBlog-master/servermanager/robot.py @@ -13,175 +13,222 @@ from servermanager.api.blogapi import BlogApi from servermanager.api.commonapi import ChatGPT, CommandHandler from .MemcacheStorage import MemcacheStorage -robot = WeRoBot(token=os.environ.get('DJANGO_WEROBOT_TOKEN') - or 'lylinux', enable_session=True) +# 初始化微信机器人（WeRoBot） +# 从环境变量获取Token，默认值为'lylinux'；启用会话功能以保存用户状态 +robot = WeRoBot( + token=os.environ.get('DJANGO_WEROBOT_TOKEN') or 'lylinux', + enable_session=True +) + +# 配置会话存储：优先使用Memcache，失败则降级为文件存储 memstorage = MemcacheStorage() -if memstorage.is_available: +if memstorage.is_available: # 检查Memcache是否可用 robot.config['SESSION_STORAGE'] = memstorage else: - if os.path.exists(os.path.join(settings.BASE_DIR, 'werobot_session')): - os.remove(os.path.join(settings.BASE_DIR, 'werobot_session')) + # 清理旧的文件存储，避免残留数据 + session_file = os.path.join(settings.BASE_DIR, 'werobot_session') + if os.path.exists(session_file): + os.remove(session_file) + # 使用文件存储会话数据（适合开发或Memcache不可用场景） robot.config['SESSION_STORAGE'] = FileStorage(filename='werobot_session') -blogapi = BlogApi() -cmd_handler = CommandHandler() -logger = logging.getLogger(__name__) +# 初始化依赖组件 +blogapi = BlogApi() # 博客数据接口（文章搜索、分类查询等） +cmd_handler = CommandHandler() # 系统命令处理（执行预设命令） +logger = logging.getLogger(__name__) # 日志记录器 def convert_to_article_reply(articles, message): + """ + 将文章列表转换为微信公众号的“图文消息”回复格式 + :param articles: 文章对象列表 + :param message: 微信接收的消息对象（用于构建回复） + :return: ArticlesReply 图文回复对象 + """ reply = ArticlesReply(message=message) + # 导入自定义模板标签，用于截取文章内容作为描述 from blog.templatetags.blog_tags import truncatechars_content + for post in articles: + # 正则提取文章正文中的第一张图片（作为图文消息封面） imgs = re.findall(r'(?:http\:|https\:)?\/\/.*\.(?:png|jpg)', post.body) - imgurl = '' - if imgs: - imgurl = imgs[0] + imgurl = imgs[0] if imgs else '' # 无图片则用空字符串 + + # 构建单条图文信息 article = Article( - title=post.title, - description=truncatechars_content(post.body), - img=imgurl, - url=post.get_full_url() + title=post.title, # 文章标题 + description=truncatechars_content(post.body), # 截取内容作为描述 + img=imgurl, # 封面图片URL + url=post.get_full_url() # 文章详情页URL ) - reply.add_article(article) + reply.add_article(article) # 添加到图文回复中 return reply -@robot.filter(re.compile(r"^\?.*")) +# ------------------------------ 微信消息处理过滤器 ------------------------------ +@robot.filter(re.compile(r"^\?.*")) # 匹配以"?"开头的消息（文章搜索） def search(message, session): - s = message.content - searchstr = str(s).replace('?', '') - result = blogapi.search_articles(searchstr) + """处理文章搜索：输入“?关键词”返回匹配的图文消息""" + searchstr = message.content.replace('?', '') # 提取关键词（去除开头的"?"） + result = blogapi.search_articles(searchstr) # 调用博客接口搜索文章 + if result: - articles = list(map(lambda x: x.object, result)) - reply = convert_to_article_reply(articles, message) - return reply + # 将搜索结果（SearchQuerySet）转换为文章对象列表 + articles = [x.object for x in result] + return convert_to_article_reply(articles, message) # 返回图文回复 else: - return '没有找到相关文章。' + return '没有找到相关文章。' # 无结果提示 -@robot.filter(re.compile(r'^category\s*$', re.I)) +@robot.filter(re.compile(r'^category\s*$', re.I)) # 匹配"category"（不区分大小写） def category(message, session): - categorys = blogapi.get_category_lists() - content = ','.join(map(lambda x: x.name, categorys)) - return '所有文章分类目录：' + content + """获取所有文章分类：输入“category”返回分类列表""" + categorys = blogapi.get_category_lists() # 调用接口获取所有分类 + # 拼接分类名称为字符串（如“Python,Java,前端”） + category_names = ','.join([x.name for x in categorys]) + return f'所有文章分类目录：{category_names}' -@robot.filter(re.compile(r'^recent\s*$', re.I)) +@robot.filter(re.compile(r'^recent\s*$', re.I)) # 匹配"recent"（不区分大小写） def recents(message, session): - articles = blogapi.get_recent_articles() + """获取最新文章：输入“recent”返回最新文章的图文消息""" + articles = blogapi.get_recent_articles() # 调用接口获取最新文章 + if articles: - reply = convert_to_article_reply(articles, message) - return reply + return convert_to_article_reply(articles, message) # 返回图文回复 else: - return "暂时还没有文章" + return "暂时还没有文章" # 无文章提示 -@robot.filter(re.compile('^help$', re.I)) +@robot.filter(re.compile('^help$', re.I)) # 匹配"help"（不区分大小写） def help(message, session): + """获取帮助：输入“help”返回功能说明""" return '''欢迎关注! - 默认会与图灵机器人聊天~~ - 你可以通过下面这些命令来获得信息 - ?关键字搜索文章. - 如?python. - category获得文章分类目录及文章数. - category-***获得该分类目录文章 - 如category-python - recent获得最新文章 - help获得帮助. - weather:获得天气 - 如weather:西安 - idcard:获得身份证信息 - 如idcard:61048119xxxxxxxxxx - music:音乐搜索 - 如music:阴天快乐 - PS:以上标点符号都不支持中文标点~~ - ''' - - -@robot.filter(re.compile(r'^weather\:.*$', re.I)) +默认会与图灵机器人聊天~~ +你可以通过下面这些命令来获得信息： +1. ?关键字 → 搜索文章（如?python） +2. category → 获得文章分类目录 +3. category-*** → 获得该分类下的文章（如category-python） +4. recent → 获得最新文章 +5. help → 获得帮助 +6. weather:城市 → 获得天气（如weather:西安） +7. idcard:号码 → 获得身份证信息（如idcard:61048119xxxxxxxxxx） +8. music:歌名 → 音乐搜索（如music:阴天快乐） + +PS: 以上标点符号不支持中文标点~~ +''' + + +@robot.filter(re.compile(r'^weather\:.*$', re.I)) # 匹配"weather:城市"格式 def weather(message, session): + """天气查询（待开发）：返回“建设中”提示""" return "建设中..." -@robot.filter(re.compile(r'^idcard\:.*$', re.I)) +@robot.filter(re.compile(r'^idcard\:.*$', re.I)) # 匹配"idcard:号码"格式 def idcard(message, session): + """身份证信息查询（待开发）：返回“建设中”提示""" return "建设中..." -@robot.handler +# ------------------------------ 默认消息处理器 ------------------------------ +@robot.handler # 未被上述过滤器匹配的消息，进入此默认处理器 def echo(message, session): + """默认消息处理：转发给MessageHandler处理（用户状态管理、管理员命令等）""" handler = MessageHandler(message, session) return handler.handler() +# ------------------------------ 用户状态与命令管理 ------------------------------ class MessageHandler: + """处理用户消息的核心类：管理用户状态（普通用户/管理员）、执行管理员命令""" def __init__(self, message, session): - userid = message.source - self.message = message - self.session = session - self.userid = userid + self.message = message # 微信消息对象 + self.session = session # 会话存储（保存用户状态） + self.userid = message.source # 用户唯一标识（微信OpenID） + + # 从会话中加载用户状态（用jsonpickle反序列化） try: - info = session[userid] - self.userinfo = jsonpickle.decode(info) - except Exception as e: - userinfo = WxUserInfo() - self.userinfo = userinfo + user_info_json = session[self.userid] + self.userinfo = jsonpickle.decode(user_info_json) + except Exception: + # 会话中无用户状态，初始化新的用户信息 + self.userinfo = WxUserInfo() @property def is_admin(self): + """判断当前用户是否处于“管理员模式”""" return self.userinfo.isAdmin @property def is_password_set(self): + """判断管理员是否已通过密码验证""" return self.userinfo.isPasswordSet def save_session(self): - info = jsonpickle.encode(self.userinfo) - self.session[self.userid] = info + """将用户状态序列化后保存到会话""" + user_info_json = jsonpickle.encode(self.userinfo) + self.session[self.userid] = user_info_json def handler(self): - info = self.message.content + """核心处理逻辑：根据用户状态分发消息处理""" + user_input = self.message.content # 用户输入的内容 - if self.userinfo.isAdmin and info.upper() == 'EXIT': - self.userinfo = WxUserInfo() + # 1. 管理员退出：已验证的管理员输入“EXIT”，退出管理员模式 + if self.is_admin and self.is_password_set and user_input.upper() == 'EXIT': + self.userinfo = WxUserInfo() # 重置用户状态为普通用户 self.save_session() return "退出成功" - if info.upper() == 'ADMIN': - self.userinfo.isAdmin = True + + # 2. 进入管理员模式：普通用户输入“ADMIN”，触发管理员验证流程 + if user_input.upper() == 'ADMIN' and not self.is_admin: + self.userinfo.isAdmin = True # 标记为管理员模式 self.save_session() return "输入管理员密码" - if self.userinfo.isAdmin and not self.userinfo.isPasswordSet: - passwd = settings.WXADMIN - if settings.TESTING: - passwd = '123' - if passwd.upper() == get_sha256(get_sha256(info)).upper(): - self.userinfo.isPasswordSet = True + + # 3. 管理员密码验证：处于管理员模式但未验证密码 + if self.is_admin and not self.is_password_set: + # 获取正确密码（测试环境用'123'，正式环境用settings中的WXADMIN） + correct_passwd = '123' if settings.TESTING else settings.WXADMIN + # 密码加密比对（两次SHA256加密，避免明文传输风险） + input_passwd = get_sha256(get_sha256(user_input)).upper() + + if input_passwd == correct_passwd.upper(): + self.userinfo.isPasswordSet = True # 标记为已验证 self.save_session() - return "验证通过,请输入命令或者要执行的命令代码:输入helpme获得帮助" + return "验证通过，请输入命令或执行代码：输入helpme获得帮助" else: + # 密码错误次数限制（3次后重置管理员模式） + self.userinfo.Count += 1 if self.userinfo.Count >= 3: - self.userinfo = WxUserInfo() + self.userinfo = WxUserInfo() # 重置为普通用户 self.save_session() - return "超过验证次数" - self.userinfo.Count += 1 + return "超过验证次数，已退出管理员模式" self.save_session() - return "验证失败，请重新输入管理员密码:" - if self.userinfo.isAdmin and self.userinfo.isPasswordSet: - if self.userinfo.Command != '' and info.upper() == 'Y': + return f"验证失败（剩余{3 - self.userinfo.Count}次），请重新输入管理员密码：" + + # 4. 管理员命令执行：已验证的管理员输入命令 + if self.is_admin and self.is_password_set: + # 确认执行命令：若之前已输入命令且当前输入“Y”，则执行 + if self.userinfo.Command != '' and user_input.upper() == 'Y': return cmd_handler.run(self.userinfo.Command) + # 查看帮助：输入“helpme”返回命令列表 + elif user_input.upper() == 'HELPME': + return cmd_handler.get_help() + # 暂存命令：输入新命令，提示确认 else: - if info.upper() == 'HELPME': - return cmd_handler.get_help() - self.userinfo.Command = info + self.userinfo.Command = user_input self.save_session() - return "确认执行: " + info + " 命令?" + return f"确认执行命令：{user_input}？（输入Y执行）" - return ChatGPT.chat(info) + # 5. 普通用户：默认转发给ChatGPT生成回复 + return ChatGPT.chat(user_input) -class WxUserInfo(): +class WxUserInfo: + """用户状态类：存储用户是否为管理员、密码验证状态、命令暂存等信息""" def __init__(self): - self.isAdmin = False - self.isPasswordSet = False - self.Count = 0 - self.Command = '' + self.isAdmin = False # 是否处于管理员模式（默认否） + self.isPasswordSet = False # 是否已通过密码验证（默认否） + self.Count = 0 # 密码错误次数（默认0） + self.Command = '' # 暂存的管理员命令（默认空）