From 7e7f7a251606c77b2892548d8256014330011736 Mon Sep 17 00:00:00 2001 From: dsjzazs Date: Sat, 14 Mar 2026 10:21:21 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A2=9E=E5=BC=BA=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E6=9F=A5=E8=AF=A2=E5=8A=9F=E8=83=BD=EF=BC=8C=E6=94=AF=E6=8C=81?= =?UTF-8?q?=E6=97=B6=E9=97=B4=E8=8C=83=E5=9B=B4=E5=92=8C=E5=88=86=E9=A1=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 10 +- USAGE.md | 78 ++++- mcp_server.py | 858 ++++++++++++++++++++++++++++++++++---------------- 3 files changed, 676 insertions(+), 270 deletions(-) diff --git a/README.md b/README.md index 969cd83..b7003c9 100644 --- a/README.md +++ b/README.md @@ -132,13 +132,19 @@ claude mcp add wechat -- python C:\Users\你的用户名\wechat-decrypt\mcp_serv | Tool | 功能 | |------|------| | `get_recent_sessions(limit)` | 最近会话列表（含消息摘要、未读数） | -| `get_chat_history(chat_name, limit)` | 指定聊天的消息记录（支持模糊匹配名字） | -| `search_messages(keyword, limit)` | 全库搜索消息内容 | +| `get_chat_history(chat_name, limit, offset, start_time, end_time)` | 指定聊天的消息记录，支持时间范围和分页 | +| `search_messages(keyword, chat_name, start_time, end_time, limit, offset)` | 统一搜索消息；支持全库、单个聊天对象、多个聊天对象、时间范围和分页 | | `get_contacts(query, limit)` | 搜索/列出联系人 | | `get_new_messages()` | 获取自上次调用以来的新消息 | 前置条件：需要先运行 `python main.py` 或 `python find_all_keys.py` 完成密钥提取。 +新增能力： +- `get_chat_history` 支持 `offset` 分页，以及 `start_time` / `end_time` 时间范围过滤 +- `search_messages` 支持“全库 / 单个联系人或群聊 / 多个联系人或群聊”的统一搜索入口 +- `search_messages` 在定向搜索时会汇报无法解析或无消息表的对象 +- 时间格式支持 `YYYY-MM-DD`、`YYYY-MM-DD HH:MM`、`YYYY-MM-DD HH:MM:SS` + **[查看使用案例 →](USAGE.md)** ### 图片解密 (V2 格式) diff --git a/USAGE.md b/USAGE.md index b37cfcb..c193191 100644 --- a/USAGE.md +++ b/USAGE.md @@ -66,7 +66,7 @@ Claude 调用 `get_chat_history`，然后自动分析总结： > 搜一下谁提过"claude" ``` -Claude 调用 `search_messages(keyword="claude")`： +Claude 调用 `search_messages(keyword="claude")`： ``` 搜索 "claude" 找到 20 条结果: @@ -78,7 +78,77 @@ Claude 调用 `search_messages(keyword="claude")`： ... ``` -## 4. 搜索联系人 +## 4. 时间范围 + 分页查看聊天记录 + +``` +> 帮我看一下██群 3 月 1 日到 3 月 7 日的聊天，先给我前 20 条 +``` + +Claude 可以调用： + +```python +get_chat_history( + chat_name="██群", + start_time="2026-03-01", + end_time="2026-03-07", + limit=20, + offset=0, +) +``` + +下一页： + +```python +get_chat_history( + chat_name="██群", + start_time="2026-03-01", + end_time="2026-03-07", + limit=20, + offset=20, +) +``` + +## 5. 搜索指定联系人/群聊在某个时间段内的消息 + +``` +> 帮我搜一下██群这周谁提到过 Claude +``` + +Claude 可以调用统一接口： + +```python +search_messages( + keyword="Claude", + chat_name="██群", + start_time="2026-03-01", + end_time="2026-03-07", + limit=20, + offset=0, +) +``` + +## 6. 多个联系人/群聊联合搜索 + +``` +> 帮我看看联系人A、联系人B 和 ██项目群 这周谁提到过“项目” +``` + +Claude 可以调用统一接口： + +```python +search_messages( + keyword="项目", + chat_name=["联系人A", "联系人B", "██项目群"], + start_time="2026-03-01", + end_time="2026-03-07", + limit=20, + offset=0, +) +``` + +如果某些名字没匹配到联系人，或没有对应消息表，结果里会单独说明。 + +## 7. 搜索联系人 ``` > 帮我找一下姓张的联系人 @@ -95,7 +165,7 @@ wxid_████ 备注: 张██ 昵称: 小██ ... ``` -## 5. 获取新消息 +## 8. 获取新消息 ``` > 有没有新消息 @@ -113,7 +183,7 @@ Claude 调用 `get_new_messages()`： [16:22] ██群 [群] (19条未读): (图片) ``` -## 6. 高级用法：群聊分析 +## 9. 高级用法：群聊分析 Claude 可以获取大量消息后自动分析活跃度、话题分布、关键人物： diff --git a/mcp_server.py b/mcp_server.py index d01b06c..2f141e1 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -217,11 +217,11 @@ atexit.register(_cache.cleanup) # ============ 联系人缓存 ============ -_contact_names = None # {username: display_name} -_contact_full = None # [{username, nick_name, remark}] -_self_username = None -_XML_UNSAFE_RE = re.compile(r' 0xFFFFFFFF: - return t & 0xFFFFFFFF, t >> 32 - return t, 0 +def _split_msg_type(t): + try: + t = int(t) + except (TypeError, ValueError): + return 0, 0 + # WeChat packs the base type into the low 32 bits and app subtype into the high 32 bits. + if t > 0xFFFFFFFF: + return t & 0xFFFFFFFF, t >> 32 + return t, 0 def resolve_username(chat_name): @@ -353,42 +353,42 @@ def _collapse_text(text): return re.sub(r'\s+', ' ', text).strip() -def _get_self_username(): - global _self_username - if _self_username: - return _self_username - - if not DB_DIR: - return '' - - names = get_contact_names() - account_dir = os.path.basename(os.path.dirname(DB_DIR)) - candidates = [account_dir] +def _get_self_username(): + global _self_username + if _self_username: + return _self_username + + if not DB_DIR: + return '' + + names = get_contact_names() + account_dir = os.path.basename(os.path.dirname(DB_DIR)) + candidates = [account_dir] m = re.fullmatch(r'(.+)_([0-9a-fA-F]{4,})', account_dir) if m: candidates.insert(0, m.group(1)) - for candidate in candidates: - if candidate and candidate in names: - _self_username = candidate - return _self_username - - return '' - - -def _load_name2id_maps(conn): - id_to_username = {} - try: - rows = conn.execute("SELECT rowid, user_name FROM Name2Id").fetchall() - except sqlite3.Error: - return id_to_username - - for rowid, user_name in rows: - if not user_name: - continue - id_to_username[rowid] = user_name - return id_to_username + for candidate in candidates: + if candidate and candidate in names: + _self_username = candidate + return _self_username + + return '' + + +def _load_name2id_maps(conn): + id_to_username = {} + try: + rows = conn.execute("SELECT rowid, user_name FROM Name2Id").fetchall() + except sqlite3.Error: + return id_to_username + + for rowid, user_name in rows: + if not user_name: + continue + id_to_username[rowid] = user_name + return id_to_username def _display_name_for_username(username, names): @@ -416,62 +416,62 @@ def _resolve_sender_label(real_sender_id, sender_from_content, is_group, chat_us return '' -def _resolve_quote_sender_label(ref_user, ref_display_name, is_group, chat_username, chat_display_name, names): - if is_group: - if ref_user: - return _display_name_for_username(ref_user, names) - return ref_display_name or '' - - self_username = _get_self_username() - if ref_user: - if ref_user == chat_username: - return chat_display_name - if self_username and ref_user == self_username: - return 'me' - return names.get(ref_user, ref_display_name or ref_user) - if ref_display_name: - if ref_display_name == chat_display_name: - return chat_display_name - self_display_name = names.get(self_username, self_username) if self_username else '' - if self_display_name and ref_display_name == self_display_name: - return 'me' - return ref_display_name - return '' - - -def _parse_xml_root(content): - if not content or len(content) > _XML_PARSE_MAX_LEN or _XML_UNSAFE_RE.search(content): - return None - - try: - return ET.fromstring(content) - except ET.ParseError: - return None - - -def _parse_int(value, fallback=0): - try: - return int(value) - except (TypeError, ValueError): - return fallback - - -def _format_app_message_text(content, local_type, is_group, chat_username, chat_display_name, names): - if not content or ' _XML_PARSE_MAX_LEN or _XML_UNSAFE_RE.search(content): + return None + + try: + return ET.fromstring(content) + except ET.ParseError: + return None + + +def _parse_int(value, fallback=0): + try: + return int(value) + except (TypeError, ValueError): + return fallback + + +def _format_app_message_text(content, local_type, is_group, chat_username, chat_display_name, names): + if not content or ' end_ts: + raise ValueError('start_time 不能晚于 end_time') + return start_ts, end_ts + + +def _build_message_filters(start_ts=None, end_ts=None, keyword=''): + clauses = [] + params = [] + if start_ts is not None: + clauses.append('create_time >= ?') + params.append(start_ts) + if end_ts is not None: + clauses.append('create_time <= ?') + params.append(end_ts) + if keyword: + clauses.append('message_content LIKE ?') + params.append(f'%{keyword}%') + return clauses, params + + +def _query_messages(conn, table_name, start_ts=None, end_ts=None, keyword='', limit=20, offset=0): + if not _is_safe_msg_table_name(table_name): + raise ValueError(f'非法消息表名: {table_name}') + + clauses, params = _build_message_filters(start_ts, end_ts, keyword) + where_sql = f"WHERE {' AND '.join(clauses)}" if clauses else '' + sql = f""" + SELECT local_id, local_type, create_time, real_sender_id, message_content, + WCDB_CT_message_content + FROM [{table_name}] + {where_sql} + ORDER BY create_time DESC + LIMIT ? OFFSET ? + """ + return conn.execute(sql, (*params, limit, offset)).fetchall() + + +def _resolve_chat_context(chat_name): + username = resolve_username(chat_name) + if not username: + return None + + names = get_contact_names() + display_name = names.get(username, username) + db_path, table_name = _find_msg_table_for_user(username) + if not db_path: + return { + 'query': chat_name, + 'username': username, + 'display_name': display_name, + 'db_path': None, + 'table_name': None, + 'is_group': '@chatroom' in username, + } + + return { + 'query': chat_name, + 'username': username, + 'display_name': display_name, + 'db_path': db_path, + 'table_name': table_name, + 'is_group': '@chatroom' in username, + } + + +def _resolve_chat_contexts(chat_names): + if not chat_names: + raise ValueError('chat_names 不能为空') + + resolved = [] + unresolved = [] + missing_tables = [] + seen = set() + + for chat_name in chat_names: + name = (chat_name or '').strip() + if not name: + unresolved.append('(空)') + continue + ctx = _resolve_chat_context(name) + if not ctx: + unresolved.append(name) + continue + if not ctx['db_path']: + missing_tables.append(ctx['display_name']) + continue + if ctx['username'] in seen: + continue + seen.add(ctx['username']) + resolved.append(ctx) + + return resolved, unresolved, missing_tables + + +def _normalize_chat_names(chat_name): + if chat_name is None: + return [] + if isinstance(chat_name, str): + value = chat_name.strip() + return [value] if value else [] + if isinstance(chat_name, (list, tuple, set)): + normalized = [] + for item in chat_name: + if item is None: + continue + value = str(item).strip() + if value: + normalized.append(value) + return normalized + value = str(chat_name).strip() + return [value] if value else [] + + +def _format_history_lines(rows, username, display_name, is_group, names, id_to_username): + lines = [] + for local_id, local_type, create_time, real_sender_id, content, ct in reversed(rows): + time_str = datetime.fromtimestamp(create_time).strftime('%Y-%m-%d %H:%M') + content = _decompress_content(content, ct) + if content is None: + content = '(无法解压)' + + sender, text = _format_message_text( + local_id, local_type, content, is_group, username, display_name, names + ) + if text and len(text) > 500: + text = text[:500] + '...' + + sender_label = _resolve_sender_label( + real_sender_id, sender, is_group, username, display_name, names, id_to_username + ) + if sender_label: + lines.append(f'[{time_str}] {sender_label}: {text}') + else: + lines.append(f'[{time_str}] {text}') + return lines + + +def _build_search_entry(row, ctx, names, id_to_username): + local_id, local_type, create_time, real_sender_id, content, ct = row + content = _decompress_content(content, ct) + if content is None: + return None + + sender, text = _format_message_text( + local_id, local_type, content, ctx['is_group'], ctx['username'], ctx['display_name'], names + ) + if text and len(text) > 300: + text = text[:300] + '...' + + sender_label = _resolve_sender_label( + real_sender_id, + sender, + ctx['is_group'], + ctx['username'], + ctx['display_name'], + names, + id_to_username, + ) + time_str = datetime.fromtimestamp(create_time).strftime('%Y-%m-%d %H:%M') + entry = f"[{time_str}] [{ctx['display_name']}]" + if sender_label: + entry += f" {sender_label}:" + entry += f" {text}" + return create_time, entry + + # ============ MCP Server ============ mcp = FastMCP("wechat", instructions="查询微信消息、联系人等数据") @@ -664,92 +864,218 @@ def get_recent_sessions(limit: int = 20) -> str: @mcp.tool() -def get_chat_history(chat_name: str, limit: int = 50) -> str: +def get_chat_history(chat_name: str, limit: int = 50, offset: int = 0, start_time: str = "", end_time: str = "") -> str: """获取指定聊天的消息记录。 Args: chat_name: 聊天对象的名字、备注名或wxid，自动模糊匹配 limit: 返回的消息数量，默认50 + offset: 分页偏移量，默认0 + start_time: 起始时间，支持 YYYY-MM-DD / YYYY-MM-DD HH:MM / YYYY-MM-DD HH:MM:SS + end_time: 结束时间，支持 YYYY-MM-DD / YYYY-MM-DD HH:MM / YYYY-MM-DD HH:MM:SS """ - username = resolve_username(chat_name) - if not username: + try: + _validate_pagination(limit, offset) + start_ts, end_ts = _parse_time_range(start_time, end_time) + except ValueError as e: + return f"错误: {e}" + + ctx = _resolve_chat_context(chat_name) + if not ctx: return f"找不到聊天对象: {chat_name}\n提示: 可以用 get_contacts(query='{chat_name}') 搜索联系人" + if not ctx['db_path']: + return f"找不到 {ctx['display_name']} 的消息记录（可能在未解密的DB中或无消息）" names = get_contact_names() - display_name = names.get(username, username) - is_group = '@chatroom' in username - - db_path, table_name = _find_msg_table_for_user(username) - if not db_path: - return f"找不到 {display_name} 的消息记录（可能在未解密的DB中或无消息）" - - conn = sqlite3.connect(db_path) - try: - id_to_username = _load_name2id_maps(conn) - rows = conn.execute(f""" - SELECT local_id, local_type, create_time, real_sender_id, message_content, - WCDB_CT_message_content - FROM [{table_name}] - ORDER BY create_time DESC - LIMIT ? - """, (limit,)).fetchall() + conn = sqlite3.connect(ctx['db_path']) + try: + id_to_username = _load_name2id_maps(conn) + rows = _query_messages( + conn, + ctx['table_name'], + start_ts=start_ts, + end_ts=end_ts, + limit=limit, + offset=offset, + ) except Exception as e: conn.close() return f"查询失败: {e}" conn.close() if not rows: - return f"{display_name} 无消息记录" + return f"{ctx['display_name']} 无消息记录" - lines = [] - for local_id, local_type, create_time, real_sender_id, content, ct in reversed(rows): - time_str = datetime.fromtimestamp(create_time).strftime('%m-%d %H:%M') + lines = _format_history_lines( + rows, + ctx['username'], + ctx['display_name'], + ctx['is_group'], + names, + id_to_username, + ) - # zstd 解压 - content = _decompress_content(content, ct) - if content is None: - content = '(无法解压)' - - sender, text = _format_message_text( - local_id, local_type, content, is_group, username, display_name, names - ) - - if text and len(text) > 500: - text = text[:500] + "..." - - sender_label = _resolve_sender_label( - real_sender_id, sender, is_group, username, display_name, names, id_to_username - ) - if sender_label: - lines.append(f"[{time_str}] {sender_label}: {text}") - else: - lines.append(f"[{time_str}] {text}") - - header = f"{display_name} 的最近 {len(lines)} 条消息" - if is_group: + header = f"{ctx['display_name']} 的消息记录（返回 {len(lines)} 条，offset={offset}, limit={limit}）" + if ctx['is_group']: header += " [群聊]" + if start_time or end_time: + header += f"\n时间范围: {start_time or '最早'} ~ {end_time or '最新'}" return header + ":\n\n" + "\n".join(lines) -@mcp.tool() -def search_messages(keyword: str, limit: int = 20) -> str: - """在所有聊天记录中搜索包含关键词的消息。 - - Args: - keyword: 搜索关键词 - limit: 返回的结果数量，默认20 - """ - if not keyword or len(keyword) < 1: - return "请提供搜索关键词" - - names = get_contact_names() - results = [] - - for rel_key in MSG_DB_KEYS: - if len(results) >= limit: - break - - path = _cache.get(rel_key) +@mcp.tool() +def search_messages( + keyword: str, + chat_name: str | list[str] | None = None, + start_time: str = "", + end_time: str = "", + limit: int = 20, + offset: int = 0, +) -> str: + """搜索消息内容，支持全库、单个聊天对象、多个聊天对象，以及时间范围和分页。 + + Args: + keyword: 搜索关键词 + chat_name: 聊天对象名称，可为空、单个字符串或字符串列表 + start_time: 起始时间，可为空 + end_time: 结束时间，可为空 + limit: 返回的结果数量，默认20 + offset: 分页偏移量，默认0 + """ + if not keyword or len(keyword) < 1: + return "请提供搜索关键词" + + chat_names = _normalize_chat_names(chat_name) + + try: + _validate_pagination(limit, offset) + start_ts, end_ts = _parse_time_range(start_time, end_time) + except ValueError as e: + return f"错误: {e}" + + if len(chat_names) == 1: + ctx = _resolve_chat_context(chat_names[0]) + if not ctx: + return f"找不到聊天对象: {chat_names[0]}\n提示: 可以用 get_contacts(query='{chat_names[0]}') 搜索联系人" + if not ctx['db_path']: + return f"找不到 {ctx['display_name']} 的消息记录（可能在未解密的DB中或无消息）" + + names = get_contact_names() + conn = sqlite3.connect(ctx['db_path']) + try: + id_to_username = _load_name2id_maps(conn) + rows = _query_messages( + conn, + ctx['table_name'], + start_ts=start_ts, + end_ts=end_ts, + keyword=keyword, + limit=limit, + offset=offset, + ) + except Exception as e: + conn.close() + return f"查询失败: {e}" + conn.close() + + if not rows: + return f"未在 {ctx['display_name']} 中找到包含 \"{keyword}\" 的消息" + + entries = [] + for row in rows: + formatted = _build_search_entry(row, ctx, names, id_to_username) + if formatted: + entries.append(formatted) + + if not entries: + return f"未在 {ctx['display_name']} 中找到包含 \"{keyword}\" 的可读消息" + + entries.sort(key=lambda x: x[0]) + header = f"在 {ctx['display_name']} 中搜索 \"{keyword}\" 找到 {len(entries)} 条结果（offset={offset}, limit={limit}）" + if start_time or end_time: + header += f"\n时间范围: {start_time or '最早'} ~ {end_time or '最新'}" + return header + ":\n\n" + "\n\n".join(item[1] for item in entries) + + if len(chat_names) > 1: + try: + resolved_contexts, unresolved, missing_tables = _resolve_chat_contexts(chat_names) + except ValueError as e: + return f"错误: {e}" + + if not resolved_contexts: + details = [] + if unresolved: + details.append("未找到联系人: " + "、".join(unresolved)) + if missing_tables: + details.append("无消息表: " + "、".join(missing_tables)) + suffix = f"\n{chr(10).join(details)}" if details else "" + return f"错误: 没有可查询的聊天对象{suffix}" + + names = get_contact_names() + collected = [] + failures = [] + per_chat_limit = limit + offset + + for ctx in resolved_contexts: + conn = sqlite3.connect(ctx['db_path']) + try: + id_to_username = _load_name2id_maps(conn) + rows = _query_messages( + conn, + ctx['table_name'], + start_ts=start_ts, + end_ts=end_ts, + keyword=keyword, + limit=per_chat_limit, + offset=0, + ) + for row in rows: + formatted = _build_search_entry(row, ctx, names, id_to_username) + if formatted: + collected.append(formatted) + except Exception as e: + failures.append(f"{ctx['display_name']}: {e}") + finally: + conn.close() + + collected.sort(key=lambda x: x[0], reverse=True) + paged = collected[offset:offset + limit] + + notes = [] + if unresolved: + notes.append("未找到联系人: " + "、".join(unresolved)) + if missing_tables: + notes.append("无消息表: " + "、".join(missing_tables)) + if failures: + notes.append("查询失败: " + "；".join(failures)) + + if not paged: + header = f"在 {len(resolved_contexts)} 个聊天对象中未找到包含 \"{keyword}\" 的消息" + if start_time or end_time: + header += f"\n时间范围: {start_time or '最早'} ~ {end_time or '最新'}" + if notes: + header += "\n" + "\n".join(notes) + return header + + header = ( + f"在 {len(resolved_contexts)} 个聊天对象中搜索 \"{keyword}\" 找到 {len(paged)} 条结果" + f"（offset={offset}, limit={limit}）" + ) + if start_time or end_time: + header += f"\n时间范围: {start_time or '最早'} ~ {end_time or '最新'}" + if notes: + header += "\n" + "\n".join(notes) + return header + ":\n\n" + "\n\n".join(item[1] for item in paged) + + names = get_contact_names() + results = [] + max_results = limit + offset + + for rel_key in MSG_DB_KEYS: + if len(results) >= max_results: + break + + path = _cache.get(rel_key) if not path: continue @@ -768,25 +1094,27 @@ def search_messages(keyword: str, limit: int = 20) -> str: name2id[f"Msg_{h}"] = r[0] except Exception: pass - - for (tname,) in tables: - if len(results) >= limit: - break - username = name2id.get(tname, '') - is_group = '@chatroom' in username - display = names.get(username, username) if username else tname - - try: - rows = conn.execute(f""" - SELECT local_type, create_time, message_content, - WCDB_CT_message_content - FROM [{tname}] - WHERE message_content LIKE ? - ORDER BY create_time DESC - LIMIT ? - """, (f'%{keyword}%', limit - len(results))).fetchall() - except Exception: - continue + + for (tname,) in tables: + if len(results) >= max_results: + break + username = name2id.get(tname, '') + is_group = '@chatroom' in username + display = names.get(username, username) if username else tname + + try: + clauses, params = _build_message_filters(start_ts, end_ts, keyword) + where_sql = f"WHERE {' AND '.join(clauses)}" if clauses else '' + rows = conn.execute(f""" + SELECT local_type, create_time, message_content, + WCDB_CT_message_content + FROM [{tname}] + {where_sql} + ORDER BY create_time DESC + LIMIT ? OFFSET ? + """, (*params, max_results - len(results), 0)).fetchall() + except Exception: + continue for local_type, ts, content, ct in rows: content = _decompress_content(content, ct) @@ -808,17 +1136,19 @@ def search_messages(keyword: str, limit: int = 20) -> str: finally: conn.close() - results.sort(key=lambda x: x[0], reverse=True) - entries = [r[1] for r in results[:limit]] + results.sort(key=lambda x: x[0], reverse=True) + entries = [r[1] for r in results[offset:offset + limit]] + + if not entries: + return f"未找到包含 \"{keyword}\" 的消息" + + header = f"搜索 \"{keyword}\" 找到 {len(entries)} 条结果（offset={offset}, limit={limit}）" + if start_time or end_time: + header += f"\n时间范围: {start_time or '最早'} ~ {end_time or '最新'}" + return header + ":\n\n" + "\n\n".join(entries) - if not entries: - return f"未找到包含 \"{keyword}\" 的消息" - - return f"搜索 \"{keyword}\" 找到 {len(entries)} 条结果:\n\n" + "\n\n".join(entries) - - -@mcp.tool() -def get_contacts(query: str = "", limit: int = 50) -> str: +@mcp.tool() +def get_contacts(query: str = "", limit: int = 50) -> str: """搜索或列出微信联系人。 Args: