wxauto-dev

# 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：语法检查（必须执行） ```bash python3 -m py_compile wxbot.py ``` 有错误则修复后再次检查，直到通过。 ### 步骤 4：输出文件 **必须输出**：`wxbot.py` + `README.md`（只生成这一份文档，不要生成 QUICKSTART.md 等） **接入 AI API 时额外输出**：`requirements.txt`（含 python-dotenv）+ `.env.example` --- ## 核心 API 速查 ```python 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)` --- ## 代码模板 所有模板共用以下样板代码（生成时完整写出）： ```python # 样板：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()` 回复。 ```python 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：关键词自动回复 核心差异：多目标监听 + 关键词字典匹配。 ```python 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`。 ```python 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()`，对新内容点赞/评论。 ```python 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` 返回后统一发送。 ```python 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