wxauto-dev

name wxauto-dev description 快速构建微信自动化机器人应用。当用户想要创建微信机器人、监听微信消息、自动回复、处理微信群消息、发送微信消息、发布朋友圈、获取朋友圈内容、或任何与微信自动化相关的需求时使用此 skill。支持消息监听、自动回复、群管理、好友管理、朋友圈操作、全量消息监听等功能。 license MIT metadata {"author":"wxauto","version":"2.5.0","category":"automation","wxauto_version":4} WxAuto-Dev Skill 基于 wxautox4 库，帮助用户快速构建微信自动化机器人。 执行流程 步骤 -1：环境说明（最优先） 用户可能在开发机上编写代码，在另一台 Windows 设备上运行。 不要尝试检测或帮助用户配置环境 ，只需在开始时简要告知运行环境要求： wxautox4 机器人需要在以下环境中运行（开发机无需满足）： Windows 10/11，微信 PC 版 4.1 Python 3.9-3.12，安装 wxautox4： pip install wxautox4 激活： wxautox4 -a 激活码 （激活码在 https://dusapi.com/register 注册后于 https://wxauto.org/purchase 购买） 检查激活状态： wxautox4 -k 告知后直接进入需求澄清和代码生成，不要等待用户确认环境，不要帮用户执行任何环境配置命令。 步骤 0：需求澄清 需求不清晰时， 必须 使用当前环境可用的询问工具（AskUserQuestion / Ask User / question 等，按实际工具名调用）引导用户，不要猜测。 询问示例（根据实际工具名调用，以下为问题内容参考）： # 监听对象不明确时，询问： 问题：您想监听哪些对象的消息？ 选项： - 特定群聊或好友（只监听指定的群或好友） - 所有新消息（无差别获取所有聊天的新消息） # 功能需求模糊时，询问： 问题：您需要机器人实现什么功能？（可多选） 选项： - 消息监听和记录 - 关键词自动回复 - 朋友圈操作（发布、点赞、评论） - 智能对话（接入 AI 大模型） 步骤 1：选择模板 需求 模板 监听指定群/好友 模板 1 关键词自动回复 模板 2 朋友圈发布 模板 3 朋友圈监控（点赞/评论） 模板 4 无差别监听所有消息 模板 5 步骤 2：生成代码 所有生成的代码必须包含： ✅ try-except 错误处理 + 3 次重试 ✅ logging 日志记录 ✅ signal handler 优雅退出 ✅ 安全发送（先 ChatWith → 验证 ChatInfo → 再 SendMsg） ✅ init_wechat 处理"未找到主窗口"错误 接入 AI API 时额外要求 ： 使用 load_dotenv() 加载 .env（ os.getenv 单独无法读取 .env），包含模型配置 requirements.txt 必须含 python-dotenv>=1.0.0 默认模型为 claude-sonnet-4-6 或 claude-opus-4-6 步骤 3：语法检查（必须执行） python3 -m py_compile wxbot.py 有错误则修复后再次检查，直到通过。 步骤 4：输出文件 必须输出 ： wxbot.py + README.md （只生成这一份文档，不要生成 QUICKSTART.md 等） 接入 AI API 时额外输出 ： requirements.txt （含 python-dotenv）+ .env.example 核心 API 速查 from wxautox4 import WeChat from wxautox4.msgs import FriendMessage from wxautox4.msgs.friend import FriendTextMessage, FriendImageMessage wx = WeChat() # 初始化 # 监听指定对象 wx.AddListenChat(nickname= '群名或好友昵称' , callback=on_message) wx.KeepRunning() # 保持运行（阻塞） # 无差别获取所有新消息 result = wx.GetNextNewMessage(filter_mute= True , callback=on_message) # result: {'chat_name': '...', 'chat_type': 'group/friend', 'msg': [...]} # 获取当前聊天所有消息 messages = wx.GetAllMessage() # 切换聊天 + 获取当前聊天信息 wx.ChatWith(who= '张三' ) chat_info = wx.ChatInfo() # {'chat_type': 'friend', 'chat_name': '张三'} # 发送消息（不要直接用，用 safe_send_msg） wx.SendMsg(msg= '你好' ) wx.SendMsg(msg= '请注意' , at= '李四' ) # @某人 wx.AtAll(msg= '重要通知' ) # @所有人 wx.SendFiles(filepath= 'C:/file.txt' ) # 朋友圈 wx.PublishMoment(text= '内容' , media_files=[ '1.png' ], privacy_config={...}) pyq = wx.Moments(timeout= 3 ) moments = pyq.GetMoments() # 获取当前页 moments = pyq.GetMoments(next_page= True ) # 下一页 moment.Like() # 点赞 moment.Like( False ) # 取消赞 moment.Comment( '很棒！' ) # 评论 pyq.Close() # 好友管理 newfriends = wx.GetNewFriends(acceptable= True ) friend.accept(remark= '备注' , tags=[ '客户' ]) wx.AddNewFriend(keywords= '张三' , addmsg= '我是小明' , remark= '老张' , tags=[ '同学' ]) 消息对象属性 ： msg.type （text/image/file/voice/video）、 msg.attr （self/friend/system）、 msg.sender 、 msg.content 、 msg.raw 消息对象方法 ： msg.reply(text) 、 msg.download(savepath) 代码模板 所有模板共用以下样板代码（生成时完整写出）： # 样板：logging + signal + init_wechat + safe_send_msg import logging, signal, sys, time from wxautox4 import WeChat logging.basicConfig( level=logging.INFO, format = '%(asctime)s - %(levelname)s - %(message)s' , handlers=[logging.FileHandler( 'wxbot.log' , encoding= 'utf-8' ), logging.StreamHandler()] ) logger = logging.getLogger(__name__) def signal_handler ( signum, frame ): logger.info( "收到退出信号" ) sys.exit( 0 ) signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) class WxBot : def init_wechat ( self, max_retries= 3 ): for attempt in range (max_retries): try : self .wx = WeChat() logger.info( "微信初始化成功" ) return True except Exception as e: if "未找到已登录的客户端主窗口" in str (e): logger.error( "未找到微信窗口，请确保微信已登录，如问题持续请退出重新登录" ) else : logger.error( f"初始化失败: {e} " ) if attempt < max_retries - 1 : time.sleep( 5 ) return False def safe_send_msg ( self, target: str , msg: str , max_retries: int = 3 ) -> bool : for attempt in range (max_retries): try : self .wx.ChatWith(who=target) time.sleep( 0.5 ) if self .wx.ChatInfo().get( 'chat_name' ) == target: self .wx.SendMsg(msg=msg) return True except Exception as e: logger.error( f"发送失败 ( {attempt+ 1 } / {max_retries} ): {e} " ) if attempt < max_retries - 1 : time.sleep( 1 ) return False 模板 1：监听指定群/好友 核心差异：使用 AddListenChat + KeepRunning ，callback 接收 (msg, chat) 两个参数，可直接用 chat.SendMsg() 回复。 from wxautox4.msgs.friend import FriendTextMessage LISTEN_TARGET = "工作群" # 修改为要监听的群名或好友昵称 KEYWORD = "通知" # 触发回复的关键词 class ListenBot ( WxBot ): def __init__ ( self ): self .wx = None def on_message ( self, msg, chat ): try : logger.info( f" {msg.sender} : {msg.content} " ) if isinstance (msg, FriendTextMessage) and KEYWORD in msg.sender: for attempt in range ( 3 ): try : chat.SendMsg( "收到" ) break except Exception as e: if attempt < 2 : time.sleep( 1 ) except Exception as e: logger.error( f"处理消息出错: {e} " , exc_info= True ) def start ( self ): if not self .init_wechat(): return try : self .wx.AddListenChat(nickname=LISTEN_TARGET, callback= self .on_message) logger.info( f"已添加监听: {LISTEN_TARGET} ，按 Ctrl+C 停止" ) self .wx.KeepRunning() except Exception as e: logger.error( f"启动失败: {e} " ) if __name__ == "__main__" : ListenBot().start() 模板 2：关键词自动回复 核心差异：多目标监听 + 关键词字典匹配。 from wxautox4.msgs.friend import FriendTextMessage LISTEN_TARGETS = [ "客服群" , "咨询群" ] REPLY_RULES = { "价格" : "请查看价格表：..." , "联系方式" : "联系电话：xxx，邮箱：xxx@example.com" , "营业时间" : "周一至周五 9:00-18:00" , } class AutoReplyBot ( WxBot ): def __init__ ( self ): self .wx = None def on_message ( self, msg, chat ): try : if not isinstance (msg, FriendTextMessage): return for keyword, reply in REPLY_RULES.items(): if keyword in msg.content: for attempt in range ( 3 ): try : chat.SendMsg(reply) logger.info( f"已回复关键词[ {keyword} ]给 {msg.sender} " ) break except Exception as e: if attempt < 2 : time.sleep( 1 ) break except Exception as e: logger.error( f"处理消息出错: {e} " , exc_info= True ) def start ( self ): if not self .init_wechat(): return for target in LISTEN_TARGETS: try : self .wx.AddListenChat(nickname=target, callback= self .on_message) logger.info( f"已添加监听: {target} " ) except Exception as e: logger.error( f"添加监听失败 {target} : {e} " ) self .wx.KeepRunning() if __name__ == "__main__" : AutoReplyBot().start() 模板 3：朋友圈定时发布 核心差异：使用 schedule 定时，调用 PublishMoment 。 import schedule class MomentBot ( WxBot ): def __init__ ( self ): self .wx = None def publish_moment ( self, text: str , media_files: list = None , max_retries: int = 3 ): for attempt in range (max_retries): try : self .wx.PublishMoment(text=text, media_files=media_files) logger.info( f"发布朋友圈成功: {text[: 20 ]} " ) return True except Exception as e: logger.error( f"发布失败 ( {attempt+ 1 } / {max_retries} ): {e} " ) if attempt < max_retries - 1 : time.sleep( 2 ) return False def daily_task ( self ): text = f"早安！今天是 {time.strftime( '%Y-%m-%d' )} " self .publish_moment(text=text) # 带图片示例：self.publish_moment(text=text, media_files=[r"D:\1.png"]) def start ( self ): if not self .init_wechat(): return schedule.every().day.at( "09:00" ).do( self .daily_task) logger.info( "朋友圈机器人已启动" ) while True : schedule.run_pending() time.sleep( 60 ) if __name__ == "__main__" : MomentBot().start() 模板 4：朋友圈监控（点赞/评论） 核心差异：定期调用 Moments().GetMoments() ，对新内容点赞/评论。 class MomentMonitorBot ( WxBot ): def __init__ ( self ): self .wx = None self .processed = set () # 已处理的朋友圈 ID def process_moments ( self ): try : pyq = self .wx.Moments(timeout= 3 ) if not pyq: logger.error( "无法打开朋友圈" ) return for moment in pyq.GetMoments(): mid = moment.content if mid in self .processed: continue try : moment.Like() logger.info( f"已点赞: {moment.author} " ) except Exception as e: logger.error( f"点赞失败: {e} " ) if "关键词" in moment.content: try : moment.Comment( "很棒！" ) except Exception as e: logger.error( f"评论失败: {e} " ) self .processed.add(mid) time.sleep( 1 ) pyq.Close() except Exception as e: logger.error( f"处理朋友圈出错: {e} " , exc_info= True ) def start ( self ): if not self .init_wechat(): return logger.info( "朋友圈监控已启动，每 10 分钟检查一次" ) while True : self .process_moments() time.sleep( 600 ) if __name__ == "__main__" : MomentMonitorBot().start() 模板 5：无差别监听所有消息 核心差异：循环调用 GetNextNewMessage 。 ⚠️ callback 严禁调用 SendMsg/ChatWith ，需要回复的消息加入队列，在 GetNextNewMessage 返回后统一发送。 from collections import deque from wxautox4.msgs.friend import FriendTextMessage FILTER_MUTE = True SLEEP_INTERVAL = 1 class AllMessageBot ( WxBot ): def __init__ ( self ): self .wx = None self .running = True self .pending_replies: deque = deque() # 待回复队列 def on_new_message ( self, msg ): """ ⚠️ 严禁在此处调用 SendMsg / ChatWith！ 只做：记录日志、下载文件、将需要回复的消息加入队列。 """ try : logger.info( f" {msg.sender} : {msg.content} " ) if msg. type in [ 'image' , 'file' , 'video' ]: try : msg.download() except Exception as e: logger.error( f"下载失败: {e} " ) # 需要回复时，加入队列 if isinstance (msg, FriendTextMessage) and "帮助" in msg.content: self .pending_replies.append((msg.sender, "您好，我是自动回复机器人" )) except Exception as e: logger.error( f"callback 出错: {e} " , exc_info= True ) def flush_pending_replies ( self ): """GetNextNewMessage 返回后，统一发送回复""" while self .pending_replies: target, reply = self .pending_replies.popleft() self .safe_send_msg(target, reply) def start ( self ): if not self .init_wechat(): return logger.info( "无差别监听已启动，按 Ctrl+C 停止" ) while self .running: try : result = self .wx.GetNextNewMessage(filter_mute=FILTER_MUTE, callback= self .on_new_message) if result: logger.info( f"收到来自 {result.get( 'chat_name' )} 的 { len (result.get( 'msg' , []))} 条消息" ) self .flush_pending_replies() # 必须在 GetNextNewMessage 返回后调用 time.sleep(SLEEP_INTERVAL) except KeyboardInterrupt: self .running = False