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：选择模板

步骤 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
            except Exception as e:
                logger.error(f"运行出错: {e}", exc_info=True)
                time.sleep(5)

if __name__ == "__main__":
    AllMessageBot().start()

代码质量检查清单

生成代码后按顺序检查：

• python3 -m py_compile wxbot.py 通过（必须第一步）
• 所有 wxautox4 调用有 try-except + 3 次重试
• 发送消息使用 safe_send_msg（先切换再验证再发送）
• init_wechat 处理了"未找到主窗口"错误
• GetNextNewMessage 的 callback 内没有 SendMsg/ChatWith 调用
• 如果用了 AI API：有 load_dotenv()，requirements.txt 含 python-dotenv
• 只生成了一份 README.md 作为文档

常见问题

记住：

1. 环境告知即可：开始时简要说明运行环境要求（Windows/微信/Python/激活），不要帮用户配置环境，直接进入代码生成
2. 需求不清晰时，必须用询问工具（按当前环境实际工具名调用）引导用户
3. 生成代码后，必须用 py_compile 检查语法
4. 只生成一份 README.md
5. 接入 AI API 时，必须用 load_dotenv()，默认模型名： claude-sonnet-4-6，默认BaseURL：https://api.dusapi.com