feat(mcp): get_chat_history 加 msg_types 按类型过滤

LLM 用 \`get_chat_history\` 查"和 X 的所有图片消息"时, 只能拉 50 条
混合消息再客户端过滤 —— 大部分 token 浪费在不需要的文本上。同样
"只看转账记录" / "只看语音" 的场景, 没有原生过滤手段。

\`get_chat_history\` 加一个可选 kwarg \`msg_types: list[str] | None = None\`:

- 接受 \`['text', 'image', 'voice', 'video', 'file', 'emoji', 'location',
  'namecard', 'voip', 'system']\` 子集
- \`'file'\` 是 alias → 'app' (WeChat 把文件归到 \`local_type=49\`,
  俗称 file)
- 输入大小写不敏感, 自动 strip
- 未知类型立即报错并列出可选值 (不偷偷过滤合法部分)
- None 或 \`[]\` 表示不过滤, 完全等价于旧行为 (向后兼容)

实现上拆 3 件:

1. \`_MSG_TYPE_MAP\` 常量 (字符串 → \`local_type\` 整数列表)
2. \`_resolve_msg_types()\` helper 做输入校验 + 翻译
3. \`_build_message_filters\` / \`_query_messages\` /
   \`_collect_chat_history_lines\` 链路加 \`type_filter=None\` 透传, SQL
   注入 \`local_type IN (?,?,...)\` clause

\`tests/test_msg_types_filter.py\` 12 个 case:

- None / 空 → 不过滤
- 单类型 / 多类型解析
- \`file\` alias → app
- 大小写 + strip 不敏感
- 未知类型报错且不放过合法的
- SQL 生成: 无过滤时 clauses 不含 \`local_type\`, 单类型生成 \`IN (?)\`,
  多类型生成 \`IN (?,?,?)\`
- 与 time / keyword 组合时 param 顺序正确

全量 \`pytest tests/\` 212/212 通过。

新参数默认 None, **既有调用方零修改**。

类型映射表 (\`_MSG_TYPE_MAP\`) 命名是有立场的判断 (比如 \`'app'\` 这一
桶实际混了文件 / 分享卡 / 小程序 / 转账 / 引用回复), 如果维护者
不同意具体 label 或想拆细, 改 dict 就行, 不影响接口。

与 #103 (\`_pagination_hint\`) 触碰同一文件 \`mcp_server.py\`, 后合的
rebase 即可, 无逻辑冲突。
This commit is contained in:
Belugary
2026-05-14 12:38:42 +08:00
committed by ylytdeng
parent 9450e46ca5
commit 6606122c86
2 changed files with 139 additions and 5 deletions

View File

@@ -0,0 +1,85 @@
"""测试 _resolve_msg_types 和 _build_message_filters 的 type_filter 路径。"""
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import mcp_server
def test_resolve_none_returns_no_filter():
assert mcp_server._resolve_msg_types(None) == (None, None)
def test_resolve_empty_returns_no_filter():
assert mcp_server._resolve_msg_types([]) == (None, None)
def test_resolve_single_text_type():
type_filter, err = mcp_server._resolve_msg_types(['text'])
assert err is None
assert type_filter == [1]
def test_resolve_multiple_types():
type_filter, err = mcp_server._resolve_msg_types(['image', 'voice', 'video'])
assert err is None
assert sorted(type_filter) == [3, 34, 43]
def test_file_alias_maps_to_app():
"""'file' 是常见叫法, 实际是 type=49 (app message)。"""
type_filter, err = mcp_server._resolve_msg_types(['file'])
assert err is None
assert type_filter == [49]
def test_case_insensitive_and_strip():
type_filter, err = mcp_server._resolve_msg_types([' Text ', 'IMAGE'])
assert err is None
assert sorted(type_filter) == [1, 3]
def test_unknown_type_returns_error():
type_filter, err = mcp_server._resolve_msg_types(['unknown'])
assert type_filter is None
assert err is not None
assert 'unknown' in err
assert 'text' in err # 错误提示列出可选值
def test_partial_unknown_aborts_whole():
"""混入一个未知类型时整体失败, 不偷偷过滤合法的。"""
type_filter, err = mcp_server._resolve_msg_types(['text', 'invalid_type'])
assert type_filter is None
assert 'invalid_type' in err
def test_build_filters_without_type_filter():
"""type_filter=None 时 SQL 不包含 local_type 子句。"""
clauses, params = mcp_server._build_message_filters()
assert not any('local_type' in c for c in clauses)
def test_build_filters_with_single_type():
clauses, params = mcp_server._build_message_filters(type_filter=[1])
assert any('local_type IN (?)' == c for c in clauses)
assert 1 in params
def test_build_filters_with_multiple_types():
clauses, params = mcp_server._build_message_filters(type_filter=[1, 3, 34])
type_clause = [c for c in clauses if 'local_type' in c][0]
assert type_clause == 'local_type IN (?,?,?)'
assert params == [1, 3, 34]
def test_build_filters_combines_with_time_and_keyword():
clauses, params = mcp_server._build_message_filters(
start_ts=1000, end_ts=2000, keyword='hello', type_filter=[1]
)
assert 'create_time >= ?' in clauses
assert 'create_time <= ?' in clauses
assert 'message_content LIKE ?' in clauses
assert any('local_type' in c for c in clauses)
assert params == [1000, 2000, '%hello%', 1]