1

wechat_monitor/config.py

@@ -0,0 +1,33 @@
+"""全局配置"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+@dataclass
+class Config:
+    """监听运行配置"""
+
+    # ---- 微信窗口 ----
+    wechat_exe: str = "WeChat.exe"           # 微信进程名
+    wechat_title: str = "微信"                # 微信主窗口标题关键字
+
+    # ---- 轮询 ----
+    poll_interval: float = 1.0                # 消息轮询间隔（秒）
+    idle_poll_interval: float = 3.0           # 空闲时的轮询间隔
+
+    # ---- 消息 ----
+    max_message_length: int = 10_000          # 单条消息最大字符数
+    dedup_window: int = 300                   # 消息去重窗口（秒）
+
+    # ---- 存储 ----
+    sqlite_path: str = "wechat_messages.db"
+    redis_url: str = ""                       # 为空则不启用 Redis
+
+    # ---- 日志 ----
+    log_level: str = "INFO"
+    log_file: str = "wechat_monitor.log"
+
+    # ---- 监控目标 ----
+    watch_conversations: list[str] = field(default_factory=list)  # 为空则监控所有

wechat_monitor/models/message.py

@@ -0,0 +1,50 @@
+"""消息数据模型"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any
+
+
+class MessageType(Enum):
+    TEXT = "text"
+    IMAGE = "image"
+    FILE = "file"
+    VOICE = "voice"
+    VIDEO = "video"
+    LINK = "link"
+    SYSTEM = "system"
+    UNKNOWN = "unknown"
+
+
+@dataclass
+class Message:
+    """单条聊天消息"""
+
+    msg_id: str
+    conversation: str          # 会话名称（联系人/群名）
+    sender: str                # 发送者
+    content: str               # 消息内容（文本 / 文件路径 / 图片路径等）
+    msg_type: MessageType = MessageType.TEXT
+    timestamp: float = field(default_factory=time.time)
+    is_self: bool = False      # 是否自己发送
+    raw_data: dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "msg_id": self.msg_id,
+            "conversation": self.conversation,
+            "sender": self.sender,
+            "content": self.content,
+            "msg_type": self.msg_type.value,
+            "timestamp": self.timestamp,
+            "is_self": self.is_self,
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> Message:
+        data = data.copy()
+        data["msg_type"] = MessageType(data["msg_type"])
+        return cls(**data)

wechat_monitor/ui/__init__.py

@@ -0,0 +1,5 @@
+from .wechat import WeChatWindow
+from .conversation import ConversationList
+from .message import MessageList
+
+__all__ = ["WeChatWindow", "ConversationList", "MessageList"]

wechat_monitor/ui/conversation.py

@@ -0,0 +1,103 @@
+"""会话列表操作"""
+
+from __future__ import annotations
+
+import logging
+import time
+
+import uiautomation as uia
+
+from ..config import Config
+
+logger = logging.getLogger(__name__)
+
+
+class ConversationList:
+    """操作微信左侧会话列表"""
+
+    def __init__(self, window: uia.WindowControl, config: Config) -> None:
+        self._window = window
+        self.config = config
+        self._list: uia.ListControl | None = None
+
+    # ------------------------------------------------------------------
+    # 初始化面板
+    # ------------------------------------------------------------------
+
+    def _ensure_list(self) -> uia.ListControl | None:
+        if self._list is not None and self._list.Exists(maxSearchSeconds=0.5):
+            return self._list
+        try:
+            # 微信会话列表通常是一个 ListControl
+            # 通过搜索树定位第一个 ListControl
+            lst = self._window.ListControl(searchDepth=8)
+            if lst.Exists(maxSearchSeconds=2):
+                self._list = lst
+                return lst
+        except Exception:
+            logger.debug("定位会话列表失败")
+        return None
+
+    # ------------------------------------------------------------------
+    # 公开方法
+    # ------------------------------------------------------------------
+
+    def get_items(self) -> list[uia.Control]:
+        """返回会话列表中的所有会话项"""
+        lst = self._ensure_list()
+        if lst is None:
+            return []
+        try:
+            return lst.GetChildren()
+        except Exception:
+            logger.exception("获取会话列表项失败")
+            return []
+
+    def get_current_conversation_name(self) -> str:
+        """获取当前激活的会话名称（聊天窗口顶部标题）"""
+        try:
+            # 聊聊窗口标题区域
+            edit = self._window.EditControl(searchDepth=8)
+            if edit.Exists(maxSearchSeconds=1):
+                return edit.Name or ""
+        except Exception:
+            pass
+        return ""
+
+    def click_conversation(self, name: str) -> bool:
+        """点击指定名称的会话"""
+        lst = self._ensure_list()
+        if lst is None:
+            return False
+        try:
+            for item in lst.GetChildren():
+                if name in (item.Name or ""):
+                    item.Click()
+                    time.sleep(0.5)
+                    logger.info("已切换到会话: %s", name)
+                    return True
+        except Exception:
+            logger.exception("点击会话失败: %s", name)
+        logger.warning("未找到会话: %s", name)
+        return False
+
+    def search_conversation(self, keyword: str) -> bool:
+        """通过搜索框定位会话"""
+        try:
+            search_edit = self._window.EditControl(Name="搜索", searchDepth=8)
+            if not search_edit.Exists(maxSearchSeconds=2):
+                search_edit = self._window.EditControl(searchDepth=3)
+            if search_edit.Exists(maxSearchSeconds=1):
+                search_edit.Click()
+                time.sleep(0.3)
+                search_edit.SendKeys(keyword, interval=0.05)
+                time.sleep(1.0)
+                # 点击第一个搜索结果
+                result = self._window.ListItemControl(searchDepth=8)
+                if result.Exists(maxSearchSeconds=2):
+                    result.Click()
+                    time.sleep(0.5)
+                    return True
+        except Exception:
+            logger.exception("搜索会话失败: %s", keyword)
+        return False

wechat_monitor/ui/message.py

@@ -0,0 +1,169 @@
+"""从微信聊天面板提取消息"""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+import time
+from typing import Any
+
+import uiautomation as uia
+
+from ..models.message import Message, MessageType
+
+logger = logging.getLogger(__name__)
+
+
+class MessageList:
+    """从聊天面板中读取消息列表"""
+
+    def __init__(self, window: uia.WindowControl) -> None:
+        self._window = window
+        self._chat_list: uia.ListControl | None = None
+
+    # ------------------------------------------------------------------
+    # 内部
+    # ------------------------------------------------------------------
+
+    def _ensure_chat_list(self) -> uia.ListControl | None:
+        if self._chat_list is not None and self._chat_list.Exists(maxSearchSeconds=0.5):
+            return self._chat_list
+        try:
+            # 聊天消息列表 — 搜索较深层级
+            lst = self._window.ListControl(searchDepth=10)
+            if lst.Exists(maxSearchSeconds=2):
+                self._chat_list = lst
+                return lst
+        except Exception:
+            logger.debug("定位聊天消息列表失败")
+        return None
+
+    @staticmethod
+    def _make_msg_id(conversation: str, sender: str, content: str, ts: float) -> str:
+        raw = f"{conversation}|{sender}|{content}|{ts}"
+        return hashlib.md5(raw.encode()).hexdigest()
+
+    # ------------------------------------------------------------------
+    # 公开方法
+    # ------------------------------------------------------------------
+
+    def extract_messages(self, conversation: str = "") -> list[Message]:
+        """提取当前聊天面板中的所有可见消息"""
+        lst = self._ensure_chat_list()
+        if lst is None:
+            return []
+
+        messages: list[Message] = []
+        try:
+            items = lst.GetChildren()
+        except Exception:
+            logger.exception("获取消息列表子项失败")
+            return []
+
+        for item in items:
+            msg = self._parse_item(item, conversation)
+            if msg is not None:
+                messages.append(msg)
+
+        return messages
+
+    def get_latest_messages(self, conversation: str = "", since: float = 0) -> list[Message]:
+        """获取指定时间戳之后的新消息"""
+        all_msgs = self.extract_messages(conversation)
+        return [m for m in all_msgs if m.timestamp > since]
+
+    # ------------------------------------------------------------------
+    # 解析单条消息控件
+    # ------------------------------------------------------------------
+
+    def _parse_item(self, item: uia.Control, conversation: str) -> Message | None:
+        """尝试将一个 ListItem 解析为 Message"""
+        try:
+            name = item.Name or ""
+            if not name:
+                return None
+
+            # 判断是否自己发送 — 通过控件的某些属性
+            is_self = self._guess_is_self(item)
+            sender = self._extract_sender(item, is_self)
+            content = name  # Name 通常就是消息内容
+            msg_type = self._guess_type(item, content)
+
+            ts = time.time()
+            msg_id = self._make_msg_id(conversation, sender, content, ts)
+
+            return Message(
+                msg_id=msg_id,
+                conversation=conversation,
+                sender=sender,
+                content=content,
+                msg_type=msg_type,
+                timestamp=ts,
+                is_self=is_self,
+                raw_data={"automation_id": item.AutomationId, "class_name": item.ClassName},
+            )
+        except Exception:
+            logger.debug("解析消息控件失败", exc_info=True)
+            return None
+
+    def _guess_is_self(self, item: uia.Control) -> bool:
+        """启发式判断消息是否为自己发送"""
+        try:
+            # 微信中，自己的消息通常在右侧，ClassName 可能包含 "Self" 或类似标识
+            cls_name = (item.ClassName or "").lower()
+            if "self" in cls_name or "right" in cls_name:
+                return True
+            # 也可以通过位置判断（控件在父容器右侧）
+            rect = item.BoundingRectangle
+            parent_rect = item.GetParentControl().BoundingRectangle if item.GetParentControl() else None
+            if parent_rect and rect:
+                mid = (parent_rect.left + parent_rect.right) / 2
+                return rect.left > mid
+        except Exception:
+            pass
+        return False
+
+    def _extract_sender(self, item: uia.Control, is_self: bool) -> str:
+        """提取消息发送者名称"""
+        try:
+            # 尝试从子控件中获取发送者名称
+            for child in item.GetChildren():
+                name = child.Name or ""
+                # 发送者名称通常较短且不是消息内容本身
+                if name and len(name) < 30 and child.ControlTypeName in (
+                    "TextControl", "ButtonControl",
+                ):
+                    return name
+        except Exception:
+            pass
+        return "self" if is_self else "unknown"
+
+    @staticmethod
+    def _guess_type(item: uia.Control, content: str) -> MessageType:
+        """启发式判断消息类型"""
+        try:
+            cls_name = (item.ClassName or "").lower()
+            if "image" in cls_name or "pic" in cls_name:
+                return MessageType.IMAGE
+            if "file" in cls_name:
+                return MessageType.FILE
+            if "voice" in cls_name or "audio" in cls_name:
+                return MessageType.VOICE
+            if "video" in cls_name:
+                return MessageType.VIDEO
+        except Exception:
+            pass
+
+        # 内容判断
+        if content.startswith("[") and content.endswith("]"):
+            tag = content.strip("[]")
+            if "图片" in tag or "image" in tag.lower():
+                return MessageType.IMAGE
+            if "文件" in tag or "file" in tag.lower():
+                return MessageType.FILE
+            if "语音" in tag or "voice" in tag.lower():
+                return MessageType.VOICE
+            if "视频" in tag or "video" in tag.lower():
+                return MessageType.VIDEO
+
+        return MessageType.TEXT

wechat_monitor/ui/wechat.py

@@ -0,0 +1,117 @@
+"""微信窗口定位与基本交互 — 基于 uiautomation"""
+
+from __future__ import annotations
+
+import logging
+import time
+
+import uiautomation as uia
+
+from ..config import Config
+
+logger = logging.getLogger(__name__)
+
+
+class WeChatWindow:
+    """封装微信主窗口的查找、激活、以及子面板获取"""
+
+    def __init__(self, config: Config) -> None:
+        self.config = config
+        self._window: uia.WindowControl | None = None
+
+    # ------------------------------------------------------------------
+    # 公开方法
+    # ------------------------------------------------------------------
+
+    def find_window(self) -> uia.WindowControl | None:
+        """查找微信主窗口（不激活）"""
+        try:
+            win = uia.WindowControl(
+                searchDepth=1,
+                Name=self.config.wechat_title,
+                ClassName="WeChatMainWndForPC",
+            )
+            if win.Exists(maxSearchSeconds=2):
+                self._window = win
+                logger.info("找到微信窗口: %s", win.Name)
+                return win
+        except Exception:
+            logger.debug("通过 ClassName 查找失败，尝试按标题查找")
+            win = uia.WindowControl(
+                searchDepth=1,
+                Name=self.config.wechat_title,
+            )
+            if win.Exists(maxSearchSeconds=2):
+                self._window = win
+                logger.info("找到微信窗口: %s", win.Name)
+                return win
+
+        logger.warning("未找到微信窗口")
+        self._window = None
+        return None
+
+    def activate(self) -> bool:
+        """激活（前置）微信窗口"""
+        win = self._window or self.find_window()
+        if win is None:
+            return False
+        try:
+            win.SetFocus()
+            time.sleep(0.3)
+            return True
+        except Exception:
+            logger.exception("激活微信窗口失败")
+            return False
+
+    def get_window(self) -> uia.WindowControl | None:
+        return self._window
+
+    def is_alive(self) -> bool:
+        """窗口是否仍然存在"""
+        win = self._window
+        if win is None:
+            return False
+        try:
+            return win.Exists(maxSearchSeconds=1)
+        except Exception:
+            return False
+
+    def get_chat_panel(self) -> uia.Control | None:
+        """获取当前聊天消息列表面板"""
+        win = self._window or self.find_window()
+        if win is None:
+            return None
+        # 聊天消息区域通常在 "ChatPanel" 或包含消息列表的 ListControl
+        try:
+            # 微信 PC 版的消息列表一般是一个 ListControl，ClassName 包含 "ChatMsgList"
+            panel = win.ListControl(searchDepth=8)
+            if panel.Exists(maxSearchSeconds=2):
+                return panel
+        except Exception:
+            logger.debug("通过 ListControl 查找聊天面板失败")
+
+        # 备选方案：搜索名称包含 "消息" 的控件
+        try:
+            panel = win.ListControl(Name="消息", searchDepth=8)
+            if panel.Exists(maxSearchSeconds=1):
+                return panel
+        except Exception:
+            pass
+
+        logger.warning("未找到聊天消息面板")
+        return None
+
+    def get_conversation_list_panel(self) -> uia.Control | None:
+        """获取左侧会话列表面板"""
+        win = self._window or self.find_window()
+        if win is None:
+            return None
+        try:
+            # 会话列表通常在左侧，ClassName 包含 "SessionList"
+            panel = win.ListControl(searchDepth=8)
+            if panel.Exists(maxSearchSeconds=2):
+                return panel
+        except Exception:
+            pass
+        logger.warning("未找到会话列表面板")
+        return None