"""从微信聊天面板提取消息""" 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