From 637240729aef86676000e0911a09ac16e0f1eee8 Mon Sep 17 00:00:00 2001 From: zhanglei Date: Wed, 9 Sep 2026 14:16:55 +0800 Subject: [PATCH] 1 --- wechat_monitor/__init__.py | 0 wechat_monitor/main.py | 109 ++++++++++++++++++++++++++++ wechat_monitor/monitor/__init__.py | 5 ++ wechat_monitor/monitor/event.py | 23 ++++++ wechat_monitor/monitor/polling.py | 87 +++++++++++++++++++++++ wechat_monitor/monitor/watcher.py | 107 ++++++++++++++++++++++++++++ wechat_monitor/parser/__init__.py | 5 ++ wechat_monitor/parser/file.py | 53 ++++++++++++++ wechat_monitor/parser/image.py | 45 ++++++++++++ wechat_monitor/parser/text.py | 36 ++++++++++ wechat_monitor/storage/__init__.py | 4 ++ wechat_monitor/storage/redis.py | 70 ++++++++++++++++++ wechat_monitor/storage/sqlite.py | 110 +++++++++++++++++++++++++++++ 13 files changed, 654 insertions(+) create mode 100644 wechat_monitor/__init__.py create mode 100644 wechat_monitor/main.py create mode 100644 wechat_monitor/monitor/__init__.py create mode 100644 wechat_monitor/monitor/event.py create mode 100644 wechat_monitor/monitor/polling.py create mode 100644 wechat_monitor/monitor/watcher.py create mode 100644 wechat_monitor/parser/__init__.py create mode 100644 wechat_monitor/parser/file.py create mode 100644 wechat_monitor/parser/image.py create mode 100644 wechat_monitor/parser/text.py create mode 100644 wechat_monitor/storage/__init__.py create mode 100644 wechat_monitor/storage/redis.py create mode 100644 wechat_monitor/storage/sqlite.py diff --git a/wechat_monitor/__init__.py b/wechat_monitor/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/wechat_monitor/main.py b/wechat_monitor/main.py new file mode 100644 index 0000000..f3e996d --- /dev/null +++ b/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() diff --git a/wechat_monitor/monitor/__init__.py b/wechat_monitor/monitor/__init__.py new file mode 100644 index 0000000..c3e5b1a --- /dev/null +++ b/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"] diff --git a/wechat_monitor/monitor/event.py b/wechat_monitor/monitor/event.py new file mode 100644 index 0000000..3f8a406 --- /dev/null +++ b/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]}" diff --git a/wechat_monitor/monitor/polling.py b/wechat_monitor/monitor/polling.py new file mode 100644 index 0000000..103c8d1 --- /dev/null +++ b/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 diff --git a/wechat_monitor/monitor/watcher.py b/wechat_monitor/monitor/watcher.py new file mode 100644 index 0000000..c1fe050 --- /dev/null +++ b/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) diff --git a/wechat_monitor/parser/__init__.py b/wechat_monitor/parser/__init__.py new file mode 100644 index 0000000..bb32e2e --- /dev/null +++ b/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"] diff --git a/wechat_monitor/parser/file.py b/wechat_monitor/parser/file.py new file mode 100644 index 0000000..32ef7be --- /dev/null +++ b/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, + } diff --git a/wechat_monitor/parser/image.py b/wechat_monitor/parser/image.py new file mode 100644 index 0000000..0c646a7 --- /dev/null +++ b/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, + } diff --git a/wechat_monitor/parser/text.py b/wechat_monitor/parser/text.py new file mode 100644 index 0000000..349888e --- /dev/null +++ b/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) diff --git a/wechat_monitor/storage/__init__.py b/wechat_monitor/storage/__init__.py new file mode 100644 index 0000000..a056499 --- /dev/null +++ b/wechat_monitor/storage/__init__.py @@ -0,0 +1,4 @@ +from .sqlite import SQLiteStorage +from .redis import RedisStorage + +__all__ = ["SQLiteStorage", "RedisStorage"] diff --git a/wechat_monitor/storage/redis.py b/wechat_monitor/storage/redis.py new file mode 100644 index 0000000..65ec517 --- /dev/null +++ b/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 diff --git a/wechat_monitor/storage/sqlite.py b/wechat_monitor/storage/sqlite.py new file mode 100644 index 0000000..ff6b022 --- /dev/null +++ b/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()]