Wechat Automation

WeChat Automation Skill

Expert guidance for WeChat monitoring and automation using wxauto (Windows) and Accessibility API (macOS).

Overview

WeReply uses Platform-specific Agents to monitor WeChat conversations and control the input box:

• Windows Agent: Python 3.12 + wxauto v4
• macOS Agent: Swift + Accessibility API
• Communication: JSON protocol via stdin/stdout with Rust Orchestrator

Architecture Pattern

微信窗口
   ↓ (UI Automation)
Platform Agent
   ├→ 监听消息（定时轮询）
   ├→ 提取消息内容
   ├→ 发送到 Orchestrator (JSON via stdout)
   └→ 接收命令 (JSON via stdin)
       ↓
   控制输入框（写入建议）

Windows Agent - wxauto v4

Installation and Setup

# 安装依赖
pip install wxauto==4.0.0

# 确保微信已登录且窗口可见

Message Monitoring Pattern

import json
import time
import sys
from wxauto import WeChat

class WeChatMonitor:
    def __init__(self, interval_ms: int = 500):
        """
        初始化微信监听器

        Args:
            interval_ms: 监听间隔（毫秒），默认 500ms
        """
        self.wechat = WeChat()
        self.interval_ms = interval_ms
        self.last_message_id = None

    def start_monitoring(self):
        """开始监听微信消息"""
        try:
            while True:
                # 获取当前聊天窗口的最新消息
                messages = self.wechat.GetAllMessage()

                if messages and len(messages) > 0:
                    latest_message = messages[-1]

                    # 检查是否是新消息（避免重复处理）
                    message_id = self._generate_message_id(latest_message)
                    if message_id != self.last_message_id:
                        self.last_message_id = message_id
                        self._send_message_to_orchestrator(latest_message)

                # 间隔等待
                time.sleep(self.interval_ms / 1000.0)

        except KeyboardInterrupt:
            self._send_error("监听被用户中断")
        except Exception as e:
            self._send_error(f"监听错误: {str(e)}")

    def _generate_message_id(self, message) -> str:
        """生成消息唯一ID（用于去重）"""
        # 结合时间戳、发送者、内容生成ID
        content = message.get('content', '')
        sender = message.get('sender', '')
        timestamp = message.get('time', '')
        return f"{sender}:{timestamp}:{hash(content)}"

    def _send_message_to_orchestrator(self, message):
        """
        发送消息到 Rust Orchestrator

        格式：
        {
            "type": "MessageNew",
            "content": "消息内容",
            "sender": "发送者",
            "timestamp": "2024-01-23T10:30:00"
        }
        """
        payload = {
            "type": "MessageNew",
            "content": message.get('content', ''),
            "sender": message.get('sender', ''),
            "timestamp": message.get('time', '')
        }

        # 输出到 stdout（Rust 会读取）
        print(json.dumps(payload, ensure_ascii=False), flush=True)

    def _send_error(self, error_message: str):
        """发送错误信息到 Orchestrator"""
        payload = {
            "type": "Error",
            "message": error_message
        }
        print(json.dumps(payload, ensure_ascii=False), flush=True)

# 使用示例
if __name__ == '__main__':
    monitor = WeChatMonitor(interval_ms=500)
    monitor.start_monitoring()

Input Box Control Pattern

class WeChatInputWriter:
    def __init__(self):
        self.wechat = WeChat()

    def write_to_input(self, content: str) -> bool:
        """
        写入内容到微信输入框

        Args:
            content: 要写入的文本

        Returns:
            bool: 写入是否成功
        """
        try:
            # 使用 wxauto 写入输入框
            self.wechat.SendMsg(content)
            return True
        except Exception as e:
            self._send_error(f"写入失败: {str(e)}")
            return False

    def clear_input(self) -> bool:
        """清空输入框"""
        try:
            # wxauto v4 提供的清空方法
            self.wechat.ClearMsg()
            return True
        except Exception as e:
            self._send_error(f"清空失败: {str(e)}")
            return False

    def _send_error(self, error_message: str):
        """发送错误到 Orchestrator"""
        payload = {
            "type": "Error",
            "message": error_message
        }
        print(json.dumps(payload, ensure_ascii=False), flush=True)

Command Handling Pattern

import sys
import json
import threading

class AgentCommandHandler:
    def __init__(self):
        self.input_writer = WeChatInputWriter()
        self.running = True

    def start_command_listener(self):
        """监听来自 Orchestrator 的命令（stdin）"""
        thread = threading.Thread(target=self._listen_commands, daemon=True)
        thread.start()

    def _listen_commands(self):
        """从 stdin 读取命令"""
        try:
            for line in sys.stdin:
                if not self.running:
                    break

                try:
                    command = json.loads(line.strip())
                    self._handle_command(command)
                except json.JSONDecodeError:
                    self._send_error(f"无效的 JSON 命令: {line}")

        except Exception as e:
            self._send_error(f"命令监听错误: {str(e)}")

    def _handle_command(self, command: dict):
        """处理命令"""
        cmd_type = command.get('type')

        if cmd_type == 'WriteInput':
            content = command.get('content', '')
            success = self.input_writer.write_to_input(content)
            self._send_response(success)

        elif cmd_type == 'ClearInput':
            success = self.input_writer.clear_input()
            self._send_response(success)

        elif cmd_type == 'HealthCheck':
            self._send_health_status()

        else:
            self._send_error(f"未知命令类型: {cmd_type}")

    def _send_response(self, success: bool):
        """发送命令执行结果"""
        payload = {
            "type": "CommandResponse",
            "success": success
        }
        print(json.dumps(payload, ensure_ascii=False), flush=True)

    def _send_health_status(self):
        """发送健康状态"""
        payload = {
            "type": "HealthStatus",
            "status": "ok",
            "agent_type": "windows_wxauto"
        }
        print(json.dumps(payload, ensure_ascii=False), flush=True)

    def _send_error(self, error_message: str):
        """发送错误"""
        payload = {
            "type": "Error",
            "message": error_message
        }
        print(json.dumps(payload, ensure_ascii=False), flush=True)

macOS Agent - Accessibility API

Swift Implementation Pattern

import Cocoa
import ApplicationServices

class WeChatMonitor {
    private var monitoringTimer: Timer?
    private var lastMessageId: String?
    private let intervalMs: Int

    init(intervalMs: Int = 500) {
        self.intervalMs = intervalMs
    }

    func startMonitoring() {
        // 请求 Accessibility 权限
        if !AXIsProcessTrusted() {
            let options = [kAXTrustedCheckOptionPrompt.takeUnretainedValue() as String: true]
            AXIsProcessTrustedWithOptions(options as CFDictionary)
            return
        }

        // 启动定时器
        monitoringTimer = Timer.scheduledTimer(
            withTimeInterval: TimeInterval(intervalMs) / 1000.0,
            repeats: true
        ) { [weak self] _ in
            self?.checkForNewMessages()
        }

        RunLoop.main.run()
    }

    private func checkForNewMessages() {
        guard let wechatApp = getWeChatApplication() else {
            return
        }

        // 使用 Accessibility API 获取消息
        if let messages = extractMessages(from: wechatApp) {
            if let latestMessage = messages.last {
                let messageId = generateMessageId(message: latestMessage)

                if messageId != lastMessageId {
                    lastMessageId = messageId
                    sendMessageToOrchestrator(message: latestMessage)
                }
            }
        }
    }

    private func getWeChatApplication() -> AXUIElement? {
        let runningApps = NSWorkspace.shared.runningApplications