From ae016a6af79b47078818a778edd28eaa3fc0f6ff Mon Sep 17 00:00:00 2001 From: zhanglei Date: Tue, 8 Sep 2026 15:53:11 +0800 Subject: [PATCH] 1 --- wechat_monitor/config.py | 33 ++++++ wechat_monitor/models/__init__.py | 0 wechat_monitor/models/message.py | 50 +++++++++ wechat_monitor/ui/__init__.py | 5 + wechat_monitor/ui/conversation.py | 103 ++++++++++++++++++ wechat_monitor/ui/message.py | 169 ++++++++++++++++++++++++++++++ wechat_monitor/ui/wechat.py | 117 +++++++++++++++++++++ 7 files changed, 477 insertions(+) create mode 100644 wechat_monitor/config.py create mode 100644 wechat_monitor/models/__init__.py create mode 100644 wechat_monitor/models/message.py create mode 100644 wechat_monitor/ui/__init__.py create mode 100644 wechat_monitor/ui/conversation.py create mode 100644 wechat_monitor/ui/message.py create mode 100644 wechat_monitor/ui/wechat.py diff --git a/wechat_monitor/config.py b/wechat_monitor/config.py new file mode 100644 index 0000000..e594922 --- /dev/null +++ b/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) # 为空则监控所有 diff --git a/wechat_monitor/models/__init__.py b/wechat_monitor/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/wechat_monitor/models/message.py b/wechat_monitor/models/message.py new file mode 100644 index 0000000..25e5c91 --- /dev/null +++ b/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) diff --git a/wechat_monitor/ui/__init__.py b/wechat_monitor/ui/__init__.py new file mode 100644 index 0000000..71d0a22 --- /dev/null +++ b/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"] diff --git a/wechat_monitor/ui/conversation.py b/wechat_monitor/ui/conversation.py new file mode 100644 index 0000000..cc970fd --- /dev/null +++ b/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 diff --git a/wechat_monitor/ui/message.py b/wechat_monitor/ui/message.py new file mode 100644 index 0000000..6af1907 --- /dev/null +++ b/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 diff --git a/wechat_monitor/ui/wechat.py b/wechat_monitor/ui/wechat.py new file mode 100644 index 0000000..61c9e32 --- /dev/null +++ b/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