1

wechat_monitor/main.py

@@ -0,0 +1,109 @@
+"""微信消息监听 — 主入口"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import signal
+import sys
+import time
+
+from .config import Config
+from .monitor.event import MessageEvent
+from .monitor.watcher import WeChatWatcher
+from .storage.sqlite import SQLiteStorage
+
+
+def setup_logging(config: Config) -> None:
+    fmt = "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
+    handlers: list[logging.Handler] = [logging.StreamHandler(sys.stdout)]
+    if config.log_file:
+        handlers.append(logging.FileHandler(config.log_file, encoding="utf-8"))
+    logging.basicConfig(
+        level=getattr(logging, config.log_level, logging.INFO),
+        format=fmt,
+        handlers=handlers,
+    )
+
+
+def create_on_message_callback(storage: SQLiteStorage):
+    """创建消息回调：打印 + 持久化"""
+
+    def _on_message(event: MessageEvent) -> None:
+        msg = event.message
+        # 控制台输出
+        prefix = "[自己]" if msg.is_self else f"[{msg.sender}]"
+        print(f"  {msg.conversation} | {prefix} {msg.content[:120]}")
+        # 存储
+        storage.save_message(msg)
+
+    return _on_message
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="微信聊天消息监听器")
+    parser.add_argument("-c", "--conversation", default="", help="指定监听的会话名称（为空则监听当前活跃会话）")
+    parser.add_argument("--db", default="wechat_messages.db", help="SQLite 数据库路径")
+    parser.add_argument("--interval", type=float, default=1.0, help="轮询间隔（秒）")
+    parser.add_argument("--log-level", default="INFO", help="日志级别")
+    args = parser.parse_args()
+
+    config = Config(
+        poll_interval=args.interval,
+        sqlite_path=args.db,
+        log_level=args.log_level,
+    )
+    setup_logging(config)
+    logger = logging.getLogger("main")
+
+    # 初始化存储
+    storage = SQLiteStorage(config.sqlite_path)
+    storage.connect()
+
+    # 初始化监听器
+    watcher = WeChatWatcher(config)
+    watcher.on_message(create_on_message_callback(storage))
+
+    logger.info("正在连接微信窗口...")
+    if not watcher.connect():
+        logger.error("连接微信失败，请确保微信已登录且窗口可见")
+        storage.close()
+        sys.exit(1)
+
+    # 优雅退出
+    def _signal_handler(sig, frame):
+        logger.info("收到退出信号，正在停止...")
+        watcher.stop()
+        storage.close()
+        sys.exit(0)
+
+    signal.signal(signal.SIGINT, _signal_handler)
+    signal.signal(signal.SIGTERM, _signal_handler)
+
+    # 启动监听
+    logger.info("开始监听会话: %s", args.conversation or "(当前活跃会话)")
+    watcher.start(conversation=args.conversation)
+
+    # 主线程保持运行
+    try:
+        while True:
+            time.sleep(1)
+            if not watcher._wechat.is_alive():
+                logger.warning("微信窗口已关闭，尝试重新连接...")
+                watcher.stop()
+                time.sleep(2)
+                if watcher.connect():
+                    watcher.start(conversation=args.conversation)
+                else:
+                    logger.error("重新连接失败，退出")
+                    break
+    except KeyboardInterrupt:
+        pass
+    finally:
+        watcher.stop()
+        storage.close()
+        logger.info("监听器已退出")
+
+
+if __name__ == "__main__":
+    main()

wechat_monitor/monitor/__init__.py

@@ -0,0 +1,5 @@
+from .watcher import WeChatWatcher
+from .event import MessageEvent
+from .polling import PollingEngine
+
+__all__ = ["WeChatWatcher", "MessageEvent", "PollingEngine"]

wechat_monitor/monitor/event.py

@@ -0,0 +1,23 @@
+"""消息事件定义"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from typing import Any
+
+from ..models.message import Message
+
+
+@dataclass
+class MessageEvent:
+    """一条新消息事件，供外部回调消费"""
+
+    message: Message
+    event_time: float = field(default_factory=time.time)
+    source: str = "ui_automation"
+
+    def __str__(self) -> str:
+        m = self.message
+        prefix = "[自己]" if m.is_self else f"[{m.sender}]"
+        return f"{m.conversation} {prefix}: {m.content[:80]}"

wechat_monitor/monitor/polling.py

@@ -0,0 +1,87 @@
+"""消息轮询引擎"""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections.abc import Callable
+
+from ..config import Config
+from ..models.message import Message
+from ..ui.message import MessageList
+from .event import MessageEvent
+
+logger = logging.getLogger(__name__)
+
+# 回调签名
+MessageCallback = Callable[[MessageEvent], None]
+
+
+class PollingEngine:
+    """以轮询方式持续从聊天面板提取新消息"""
+
+    def __init__(self, message_list: MessageList, config: Config) -> None:
+        self._message_list = message_list
+        self.config = config
+        self._callbacks: list[MessageCallback] = []
+        self._seen_ids: set[str] = set()
+        self._last_poll: float = 0
+        self._running = False
+
+    # ------------------------------------------------------------------
+    # 回调注册
+    # ------------------------------------------------------------------
+
+    def on_message(self, callback: MessageCallback) -> None:
+        """注册新消息回调"""
+        self._callbacks.append(callback)
+
+    # ------------------------------------------------------------------
+    # 轮询控制
+    # ------------------------------------------------------------------
+
+    def poll_once(self, conversation: str = "") -> list[Message]:
+        """执行一次消息提取，返回去重后的新消息"""
+        try:
+            messages = self._message_list.extract_messages(conversation)
+        except Exception:
+            logger.exception("提取消息失败")
+            return []
+
+        new_messages: list[Message] = []
+        for msg in messages:
+            if msg.msg_id not in self._seen_ids:
+                self._seen_ids.add(msg.msg_id)
+                new_messages.append(msg)
+
+        # 清理过期的去重 ID（防止内存泄漏）
+        if len(self._seen_ids) > 10_000:
+            self._seen_ids = set(list(self._seen_ids)[-5_000:])
+
+        # 触发回调
+        for msg in new_messages:
+            event = MessageEvent(message=msg)
+            for cb in self._callbacks:
+                try:
+                    cb(event)
+                except Exception:
+                    logger.exception("消息回调执行失败")
+
+        self._last_poll = time.time()
+        return new_messages
+
+    def start_loop(self, conversation: str = "", stop_event=None) -> None:
+        """持续轮询，直到 stop_event 被 set 或调用 stop()"""
+        self._running = True
+        logger.info("轮询引擎启动 (间隔 %.1fs)", self.config.poll_interval)
+
+        while self._running:
+            if stop_event is not None and stop_event.is_set():
+                break
+            self.poll_once(conversation)
+            time.sleep(self.config.poll_interval)
+
+        logger.info("轮询引擎已停止")
+
+    def stop(self) -> None:
+        self._running = False

wechat_monitor/monitor/watcher.py

@@ -0,0 +1,107 @@
+"""微信消息监听器 — 顶层编排"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from collections.abc import Callable
+
+from ..config import Config
+from ..ui.wechat import WeChatWindow
+from ..ui.conversation import ConversationList
+from ..ui.message import MessageList
+from .event import MessageEvent
+from .polling import PollingEngine
+
+logger = logging.getLogger(__name__)
+
+
+class WeChatWatcher:
+    """微信消息监听器：整合窗口定位、会话切换、消息轮询"""
+
+    def __init__(self, config: Config | None = None) -> None:
+        self.config = config or Config()
+        self._wechat = WeChatWindow(self.config)
+        self._conv_list: ConversationList | None = None
+        self._msg_list: MessageList | None = None
+        self._poller: PollingEngine | None = None
+        self._thread: threading.Thread | None = None
+        self._stop_event = threading.Event()
+        self._callbacks: list[Callable[[MessageEvent], None]] = []
+
+    # ------------------------------------------------------------------
+    # 回调
+    # ------------------------------------------------------------------
+
+    def on_message(self, callback: Callable[[MessageEvent], None]) -> None:
+        """注册消息回调（可在 start 前多次调用）"""
+        self._callbacks.append(callback)
+
+    # ------------------------------------------------------------------
+    # 初始化
+    # ------------------------------------------------------------------
+
+    def connect(self) -> bool:
+        """查找并连接微信窗口"""
+        win = self._wechat.find_window()
+        if win is None:
+            logger.error("无法连接微信窗口，请确保微信已登录并保持窗口可见")
+            return False
+
+        self._conv_list = ConversationList(win, self.config)
+        self._msg_list = MessageList(win)
+        self._poller = PollingEngine(self._msg_list, self.config)
+
+        for cb in self._callbacks:
+            self._poller.on_message(cb)
+
+        logger.info("微信监听器初始化完成")
+        return True
+
+    # ------------------------------------------------------------------
+    # 启动 / 停止
+    # ------------------------------------------------------------------
+
+    def start(self, conversation: str = "", daemon: bool = True) -> bool:
+        """在后台线程中启动消息轮询"""
+        if self._poller is None:
+            if not self.connect():
+                return False
+
+        self._stop_event.clear()
+        self._thread = threading.Thread(
+            target=self._poller.start_loop,
+            args=(conversation, self._stop_event),
+            daemon=daemon,
+            name="wechat-watcher",
+        )
+        self._thread.start()
+        logger.info("监听线程已启动")
+        return True
+
+    def stop(self) -> None:
+        """停止监听"""
+        self._stop_event.set()
+        if self._poller:
+            self._poller.stop()
+        if self._thread and self._thread.is_alive():
+            self._thread.join(timeout=5)
+        logger.info("监听已停止")
+
+    # ------------------------------------------------------------------
+    # 会话切换
+    # ------------------------------------------------------------------
+
+    def switch_conversation(self, name: str) -> bool:
+        """切换到指定会话"""
+        if self._conv_list is None:
+            logger.error("监听器未初始化，请先调用 connect()")
+            return False
+        return self._conv_list.click_conversation(name)
+
+    def search_and_switch(self, keyword: str) -> bool:
+        """搜索并切换到指定会话"""
+        if self._conv_list is None:
+            logger.error("监听器未初始化，请先调用 connect()")
+            return False
+        return self._conv_list.search_conversation(keyword)

wechat_monitor/parser/__init__.py

@@ -0,0 +1,5 @@
+from .text import TextParser
+from .image import ImageParser
+from .file import FileParser
+
+__all__ = ["TextParser", "ImageParser", "FileParser"]

wechat_monitor/parser/file.py

@@ -0,0 +1,53 @@
+"""文件消息解析"""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+_WECHAT_FILE_DIRS = [
+    Path(os.environ.get("USERPROFILE", "")) / "Documents" / "WeChat Files",
+]
+
+
+class FileParser:
+    """解析文件消息"""
+
+    @staticmethod
+    def is_file_message(content: str) -> bool:
+        tag = content.strip("[]")
+        return "文件" in tag or "file" in tag.lower()
+
+    @staticmethod
+    def extract_filename(content: str) -> str:
+        """从消息内容中提取文件名"""
+        # 去除 [文件] 等标签
+        name = content.strip("[]").strip()
+        # 去除 "文件:" 等前缀
+        for prefix in ("文件:", "文件：", "file:", "File:"):
+            if name.startswith(prefix):
+                name = name[len(prefix):].strip()
+        return name
+
+    @staticmethod
+    def find_cached_file(filename: str) -> Path | None:
+        for base_dir in _WECHAT_FILE_DIRS:
+            if not base_dir.exists():
+                continue
+            for p in base_dir.rglob(filename):
+                if p.is_file():
+                    return p
+        return None
+
+    @staticmethod
+    def get_file_info(path: Path) -> dict:
+        stat = path.stat()
+        return {
+            "path": str(path),
+            "size": stat.st_size,
+            "suffix": path.suffix,
+            "name": path.name,
+        }

wechat_monitor/parser/image.py

@@ -0,0 +1,45 @@
+"""图片消息解析"""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+# 微信图片缓存默认路径
+_WECHAT_IMAGE_DIRS = [
+    Path(os.environ.get("APPDATA", "")) / "Tencent" / "WeChat" / "Image",
+    Path(os.environ.get("USERPROFILE", "")) / "Documents" / "WeChat Files",
+]
+
+
+class ImageParser:
+    """解析图片消息，尝试定位本地缓存文件"""
+
+    @staticmethod
+    def is_image_message(content: str) -> bool:
+        tag = content.strip("[]")
+        return "图片" in tag or "image" in tag.lower()
+
+    @staticmethod
+    def find_cached_image(filename: str) -> Path | None:
+        """在常见微信缓存目录中查找图片文件"""
+        for base_dir in _WECHAT_IMAGE_DIRS:
+            if not base_dir.exists():
+                continue
+            for p in base_dir.rglob(filename):
+                if p.is_file():
+                    return p
+        return None
+
+    @staticmethod
+    def get_image_info(path: Path) -> dict:
+        """获取图片基本信息"""
+        stat = path.stat()
+        return {
+            "path": str(path),
+            "size": stat.st_size,
+            "suffix": path.suffix,
+        }

wechat_monitor/parser/text.py

@@ -0,0 +1,36 @@
+"""文本消息解析/清洗"""
+
+from __future__ import annotations
+
+import re
+
+
+class TextParser:
+    """对文本消息内容进行清洗与结构化提取"""
+
+    # 常见的 @ 消息模式
+    _AT_PATTERN = re.compile(r"@(\S+)\s")
+
+    # URL 模式
+    _URL_PATTERN = re.compile(r"https?://\S+")
+
+    @staticmethod
+    def clean(content: str) -> str:
+        """去除多余空白与不可见字符"""
+        content = content.strip()
+        content = re.sub(r"[\u200b\u200c\u200d\ufeff]", "", content)
+        return content
+
+    @staticmethod
+    def extract_urls(content: str) -> list[str]:
+        return TextParser._URL_PATTERN.findall(content)
+
+    @staticmethod
+    def extract_mentions(content: str) -> list[str]:
+        return TextParser._AT_PATTERN.findall(content)
+
+    @staticmethod
+    def is_system_message(content: str) -> bool:
+        """判断是否为系统消息（撤回、入群等）"""
+        keywords = ["撤回了一条消息", "加入了群聊", "你已添加了", "以上是打招呼的内容"]
+        return any(kw in content for kw in keywords)

wechat_monitor/storage/__init__.py

@@ -0,0 +1,4 @@
+from .sqlite import SQLiteStorage
+from .redis import RedisStorage
+
+__all__ = ["SQLiteStorage", "RedisStorage"]

wechat_monitor/storage/redis.py

@@ -0,0 +1,70 @@
+"""Redis 消息缓存（可选）"""
+
+from __future__ import annotations
+
+import json
+import logging
+import time
+
+from ..models.message import Message
+
+logger = logging.getLogger(__name__)
+
+
+class RedisStorage:
+    """基于 Redis 的消息缓存，用于实时消息分发与去重"""
+
+    def __init__(self, redis_url: str = "redis://localhost:6379/0") -> None:
+        self._url = redis_url
+        self._client = None
+
+    def connect(self) -> bool:
+        try:
+            import redis
+            self._client = redis.from_url(self._url, decode_responses=True)
+            self._client.ping()
+            logger.info("Redis 连接成功: %s", self._url)
+            return True
+        except ImportError:
+            logger.warning("redis-py 未安装，Redis 存储不可用")
+            return False
+        except Exception:
+            logger.exception("Redis 连接失败")
+            return False
+
+    def close(self) -> None:
+        if self._client:
+            self._client.close()
+
+    def publish_message(self, msg: Message, channel: str = "wechat:messages") -> bool:
+        """将消息发布到 Redis 频道"""
+        if self._client is None:
+            return False
+        try:
+            data = json.dumps(msg.to_dict(), ensure_ascii=False)
+            self._client.publish(channel, data)
+            return True
+        except Exception:
+            logger.exception("Redis 发布失败")
+            return False
+
+    def cache_message(self, msg: Message, ttl: int = 3600) -> bool:
+        """将消息缓存到 Redis（用于去重）"""
+        if self._client is None:
+            return False
+        try:
+            key = f"wechat:msg:{msg.msg_id}"
+            self._client.setex(key, ttl, json.dumps(msg.to_dict(), ensure_ascii=False))
+            return True
+        except Exception:
+            logger.exception("Redis 缓存失败")
+            return False
+
+    def is_message_seen(self, msg_id: str) -> bool:
+        """检查消息是否已处理"""
+        if self._client is None:
+            return False
+        try:
+            return self._client.exists(f"wechat:msg:{msg_id}") > 0
+        except Exception:
+            return False

wechat_monitor/storage/sqlite.py

@@ -0,0 +1,110 @@
+"""SQLite 消息持久化"""
+
+from __future__ import annotations
+
+import logging
+import sqlite3
+import time
+from pathlib import Path
+
+from ..models.message import Message
+
+logger = logging.getLogger(__name__)
+
+_CREATE_TABLE = """
+CREATE TABLE IF NOT EXISTS messages (
+    msg_id      TEXT PRIMARY KEY,
+    conversation TEXT NOT NULL,
+    sender      TEXT NOT NULL,
+    content     TEXT NOT NULL,
+    msg_type    TEXT NOT NULL,
+    timestamp   REAL NOT NULL,
+    is_self     INTEGER NOT NULL DEFAULT 0,
+    created_at  REAL NOT NULL
+)
+"""
+
+_CREATE_INDEXES = [
+    "CREATE INDEX IF NOT EXISTS idx_conv ON messages(conversation)",
+    "CREATE INDEX IF NOT EXISTS idx_ts ON messages(timestamp)",
+]
+
+
+class SQLiteStorage:
+    """基于 SQLite 的消息存储"""
+
+    def __init__(self, db_path: str = "wechat_messages.db") -> None:
+        self._db_path = db_path
+        self._conn: sqlite3.Connection | None = None
+
+    def connect(self) -> None:
+        self._conn = sqlite3.connect(self._db_path, check_same_thread=False)
+        self._conn.execute(_CREATE_TABLE)
+        for idx_sql in _CREATE_INDEXES:
+            self._conn.execute(idx_sql)
+        self._conn.commit()
+        logger.info("SQLite 存储已初始化: %s", self._db_path)
+
+    def close(self) -> None:
+        if self._conn:
+            self._conn.close()
+            self._conn = None
+
+    def save_message(self, msg: Message) -> bool:
+        """保存单条消息（幂等）"""
+        if self._conn is None:
+            logger.error("数据库未连接")
+            return False
+        try:
+            self._conn.execute(
+                "INSERT OR IGNORE INTO messages "
+                "(msg_id, conversation, sender, content, msg_type, timestamp, is_self, created_at) "
+                "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+                (
+                    msg.msg_id,
+                    msg.conversation,
+                    msg.sender,
+                    msg.content,
+                    msg.msg_type.value,
+                    msg.timestamp,
+                    int(msg.is_self),
+                    time.time(),
+                ),
+            )
+            self._conn.commit()
+            return True
+        except Exception:
+            logger.exception("保存消息失败")
+            return False
+
+    def save_messages(self, messages: list[Message]) -> int:
+        """批量保存，返回成功数量"""
+        count = 0
+        for msg in messages:
+            if self.save_message(msg):
+                count += 1
+        return count
+
+    def query_messages(
+        self,
+        conversation: str = "",
+        limit: int = 100,
+        since: float = 0,
+    ) -> list[dict]:
+        """查询消息"""
+        if self._conn is None:
+            return []
+        sql = "SELECT * FROM messages WHERE 1=1"
+        params: list = []
+        if conversation:
+            sql += " AND conversation = ?"
+            params.append(conversation)
+        if since:
+            sql += " AND timestamp > ?"
+            params.append(since)
+        sql += " ORDER BY timestamp DESC LIMIT ?"
+        params.append(limit)
+
+        cursor = self._conn.execute(sql, params)
+        cols = [d[0] for d in cursor.description]
+        return [dict(zip(cols, row)) for row in cursor.fetchall()]