hitsword
/
wx-cli
mirror of https://github.com/jackwener/wx-cli.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat: 增强消息查询功能，支持时间范围和分页

feat/daemon-cli
dsjzazs 2026-03-14 10:21:21 +08:00
parent 7020409543
commit 7e7f7a2516
3 changed files with 676 additions and 270 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
README.md
View File

@ -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 格式)

78
USAGE.md
View File

@ -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 可以获取大量消息后自动分析活跃度、话题分布、关键人物：

858
mcp_server.py
View File

@ -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'<!DOCTYPE|<!ENTITY', re.IGNORECASE)
_XML_PARSE_MAX_LEN = 20000
_contact_names = None # {username: display_name}
_contact_full = None # [{username, nick_name, remark}]
_self_username = None
_XML_UNSAFE_RE = re.compile(r'<!DOCTYPE|<!ENTITY', re.IGNORECASE)
_XML_PARSE_MAX_LEN = 20000
def _load_contacts_from(db_path):
@ -283,15 +283,15 @@ def format_msg_type(t):
}.get(base_type, f'type={t}')
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 _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 '<appmsg' not in content:
return None
_, sub_type = _split_msg_type(local_type)
root = _parse_xml_root(content)
if root is None:
return None
appmsg = root.find('.//appmsg')
if appmsg is None:
return None
title = _collapse_text(appmsg.findtext('title') or '')
app_type_text = (appmsg.findtext('type') or '').strip()
app_type = _parse_int(app_type_text, _parse_int(sub_type, 0))
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 '<appmsg' not in content:
return None
_, sub_type = _split_msg_type(local_type)
root = _parse_xml_root(content)
if root is None:
return None
appmsg = root.find('.//appmsg')
if appmsg is None:
return None
title = _collapse_text(appmsg.findtext('title') or '')
app_type_text = (appmsg.findtext('type') or '').strip()
app_type = _parse_int(app_type_text, _parse_int(sub_type, 0))
if app_type == 57:
ref = appmsg.find('.//refermsg')
@ -500,85 +500,85 @@ def _format_app_message_text(content, local_type, is_group, chat_username, chat_
return f"[链接] {title}" if title else "[链接]"
if app_type in (33, 36, 44):
return f"[小程序] {title}" if title else "[小程序]"
if title:
return f"[链接/文件] {title}"
return "[链接/文件]"
def _format_voip_message_text(content):
if not content or '<voip' not in content:
return None
root = _parse_xml_root(content)
if root is None:
return "[通话]"
raw_text = _collapse_text(root.findtext('.//msg') or '')
if not raw_text:
return "[通话]"
status_map = {
'Canceled': '已取消',
'Line busy': '对方忙线',
'Already answered elsewhere': '已在其他设备接听',
'Declined on other device': '已在其他设备拒接',
'Call canceled by caller': '主叫已取消',
'Call not answered': '未接听',
"Call wasn't answered": '未接听',
}
if raw_text.startswith('Duration:'):
duration = raw_text.split(':', 1)[1].strip()
return f"[通话] 通话时长 {duration}" if duration else "[通话]"
return f"[通话] {status_map.get(raw_text, raw_text)}"
def _format_message_text(local_id, local_type, content, is_group, chat_username, chat_display_name, names):
sender_from_content, text = _parse_message_content(content, local_type, is_group)
base_type, _ = _split_msg_type(local_type)
if title:
return f"[链接/文件] {title}"
return "[链接/文件]"
if base_type == 3:
text = f"[图片] (local_id={local_id})"
elif base_type == 47:
text = "[表情]"
elif base_type == 50:
text = _format_voip_message_text(text) or "[通话]"
elif base_type == 49:
text = _format_app_message_text(
text, local_type, is_group, chat_username, chat_display_name, names
) or "[链接/文件]"
def _format_voip_message_text(content):
if not content or '<voip' not in content:
return None
root = _parse_xml_root(content)
if root is None:
return "[通话]"
raw_text = _collapse_text(root.findtext('.//msg') or '')
if not raw_text:
return "[通话]"
status_map = {
'Canceled': '已取消',
'Line busy': '对方忙线',
'Already answered elsewhere': '已在其他设备接听',
'Declined on other device': '已在其他设备拒接',
'Call canceled by caller': '主叫已取消',
'Call not answered': '未接听',
"Call wasn't answered": '未接听',
}
if raw_text.startswith('Duration:'):
duration = raw_text.split(':', 1)[1].strip()
return f"[通话] 通话时长 {duration}" if duration else "[通话]"
return f"[通话] {status_map.get(raw_text, raw_text)}"
def _format_message_text(local_id, local_type, content, is_group, chat_username, chat_display_name, names):
sender_from_content, text = _parse_message_content(content, local_type, is_group)
base_type, _ = _split_msg_type(local_type)
if base_type == 3:
text = f"[图片] (local_id={local_id})"
elif base_type == 47:
text = "[表情]"
elif base_type == 50:
text = _format_voip_message_text(text) or "[通话]"
elif base_type == 49:
text = _format_app_message_text(
text, local_type, is_group, chat_username, chat_display_name, names
) or "[链接/文件]"
elif base_type != 1:
type_label = format_msg_type(local_type)
text = f"[{type_label}] {text}" if text else f"[{type_label}]"
return sender_from_content, text
def _is_safe_msg_table_name(table_name):
return bool(re.fullmatch(r'Msg_[0-9a-f]{32}', table_name))
# 消息 DB 的 rel_keys
# 用 message_\d+\.db$ 匹配，自然排除 message_resource.db / message_fts_*.db
MSG_DB_KEYS = sorted([
return sender_from_content, text
def _is_safe_msg_table_name(table_name):
return bool(re.fullmatch(r'Msg_[0-9a-f]{32}', table_name))
# 消息 DB 的 rel_keys
# 用 message_\d+\.db$ 匹配，自然排除 message_resource.db / message_fts_*.db
MSG_DB_KEYS = sorted([
k for k in ALL_KEYS
if any(v.startswith("message/") for v in key_path_variants(k))
and any(re.search(r"message_\d+\.db$", v) for v in key_path_variants(k))
])
def _find_msg_table_for_user(username):
"""在所有 message_N.db 中查找用户的消息表，返回 (db_path, table_name)"""
table_hash = hashlib.md5(username.encode()).hexdigest()
table_name = f"Msg_{table_hash}"
if not _is_safe_msg_table_name(table_name):
return None, None
for rel_key in MSG_DB_KEYS:
path = _cache.get(rel_key)
if not path:
continue
def _find_msg_table_for_user(username):
"""在所有 message_N.db 中查找用户的消息表，返回 (db_path, table_name)"""
table_hash = hashlib.md5(username.encode()).hexdigest()
table_name = f"Msg_{table_hash}"
if not _is_safe_msg_table_name(table_name):
return None, None
for rel_key in MSG_DB_KEYS:
path = _cache.get(rel_key)
if not path:
continue
conn = sqlite3.connect(path)
try:
exists = conn.execute(
@ -596,6 +596,206 @@ def _find_msg_table_for_user(username):
return None, None
def _validate_pagination(limit, offset=0):
if limit <= 0:
raise ValueError("limit 必须大于 0")
if offset < 0:
raise ValueError("offset 不能小于 0")
def _parse_time_value(value, field_name, is_end=False):
value = (value or '').strip()
if not value:
return None
formats = [
('%Y-%m-%d %H:%M:%S', False),
('%Y-%m-%d %H:%M', False),
('%Y-%m-%d', True),
]
for fmt, date_only in formats:
try:
dt = datetime.strptime(value, fmt)
if date_only and is_end:
dt = dt.replace(hour=23, minute=59, second=59)
return int(dt.timestamp())
except ValueError:
continue
raise ValueError(
f"{field_name} 格式无效: {value}。支持 YYYY-MM-DD / YYYY-MM-DD HH:MM / YYYY-MM-DD HH:MM:SS"
)
def _parse_time_range(start_time='', end_time=''):
start_ts = _parse_time_value(start_time, 'start_time', is_end=False)
end_ts = _parse_time_value(end_time, 'end_time', is_end=True)
if start_ts is not None and end_ts is not None and start_ts > 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: