From eba9a9d4cddb2967f6dddebe3c7a4fa400a5b90c Mon Sep 17 00:00:00 2001 From: Belugary <53219544+Belugary@users.noreply.github.com> Date: Tue, 26 May 2026 16:54:43 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=A7=A3=E6=9E=90=E5=BE=AE=E4=BF=A1?= =?UTF-8?q?=E4=BD=8D=E7=BD=AE=E6=B6=88=E6=81=AF=20(base=5Ftype=3D48)=20+?= =?UTF-8?q?=20decode=5Flocation=20MCP=20=E5=B7=A5=E5=85=B7=20(#124)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 位置消息 (base_type=48) 老逻辑命中 _format_message_text 的 elif base_type != 1 兜底, 以 raw XML 形态进 get_chat_history. 一条 location XML 平均 12-15 个 attr (poiBusinessHour / poiPhone / adcode / buildingId / floorName / infourl / fromusername / maptype / scale / version 等), LLM context 里堆机器字段而看不到 "用户分享了哪里" 的核心语义. 修复 (双层模式, 跟 #83 namecard / #85 transfer / #100 refer 一致): - _extract_location_info: 解析 节点为 dict, 全部 17 个 attr + lat/lng (x/y 数值化) + category_top - _is_location_poiname_placeholder: 检测客户端在用户手扔图钉时的占位符 [位置] / [Location] - _format_location_text: 单行渲染只挑 3 个信号 (poiCategoryTips 主类 → 前缀, poiname, label), fallback chain - decode_location MCP 工具: 给 LLM 拿全部字段 (POI id / 电话 / 营业时间 / 价格 / 城市 / 区划码 / 经纬度), 兼容跨分片场景 字段语义基于 1411 条真实样本统计指导分类决策 (修了 #121 自撤的核心教训 — 没逐字段过语义就套 #83 模板). 关键差异: - poiid (#121 错丢) → 80% 出现率, deeplink 种子, 结构化保留 - infourl (#121 没充分论证就丢) → 1411 样本 0% 非空, render 丢 / structured 防御性留 - poiCategoryTips (#121 未识别) → 渲染前缀 - poiPhone (#121 未识别) → 22% 非空商家电话, 结构化保留 - poiname=[位置] 占位符 (#121 未处理) → 检测并 fallback 到 label 测试: tests/test_location_message.py 18 case, 5 个全合成 fixture (POI 名/地址/城市/坐标/电话/poiid 全部占位符, 不绑定任何真实地理数据). 18/18 全过. baseline 309 → 327. Closes #121 --- mcp_server.py | 224 +++++++++++++++++++++++++++++++++ tests/test_location_message.py | 173 +++++++++++++++++++++++++ 2 files changed, 397 insertions(+) create mode 100644 tests/test_location_message.py diff --git a/mcp_server.py b/mcp_server.py index 8211f5f..2e20938 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -775,6 +775,106 @@ def _format_namecard_text(content): return f"[名片] {head}: {certinfo}" if certinfo else f"[名片] {head}" +# 微信位置消息 的字段名 → 结构化键名。 +# +# 字段语义分类 (#121 自撤回的教训：必须逐字段过语义，不能套 #83 namecard 的 +# "丢敏感字段" 模板)。1411 条真实样本统计 + 用户分享 vs 客户端渲染二分： +# +# user-shared signal (用户在地图上主动选/填) → 进 decode_location 结构化层： +# poiname / label / poiid / poiCategoryTips / poiPhone / poiBusinessHour / +# poiPriceTips / isFromPoiList / cityname / adcode / buildingId / floorName +# 主信号坐标 (精度数字，不进单行渲染避免 LLM context 污染)： +# x (实际是纬度) / y (实际是经度) +# schema slot 但本 corpus 0% 非空 (defensive 保留，别家账号可能填)： +# infourl / version +# 渲染样式 / enum / 冗余 (defensive 保留供 debug，不参与渲染决策)： +# maptype / scale / fromusername +_LOCATION_TEXT_FIELDS = ( + 'label', 'poiname', 'poiid', 'poiCategoryTips', 'poiBusinessHour', + 'poiPhone', 'poiPriceTips', 'isFromPoiList', 'cityname', 'adcode', + 'buildingId', 'floorName', 'infourl', 'maptype', 'scale', + 'fromusername', 'version', +) + + +def _extract_location_info(content): + """Parse type=48 (位置) XML into a structured dict, or return None. + + 返回字段在 _LOCATION_TEXT_FIELDS 之外还有 lat/lng (x/y 数值化)。 + 所有缺失字段返回空串而非 None，跟 _extract_transfer_info 一致。 + """ + root = _parse_xml_root(content) + if root is None: + return None + loc = root.find('.//location') + if loc is None: + return None + + def _attr(name): + return _collapse_text(loc.get(name) or '') + + def _f(name): + v = loc.get(name) + if v is None or v == '': + return None + try: + return float(v) + except (TypeError, ValueError): + return None + + info = {k: _attr(k) for k in _LOCATION_TEXT_FIELDS} + info['lat'] = _f('x') # 微信 x 实际是纬度 + info['lng'] = _f('y') # 微信 y 实际是经度 + info['category_top'] = info['poiCategoryTips'].split(':', 1)[0] if info['poiCategoryTips'] else '' + return info + + +def _is_location_poiname_placeholder(poiname): + """检测客户端在用户手扔图钉 (未选 POI) 时填的占位符。 + + 实测样本：poiname="[位置]" / poiname="[Location]"。形如 [*] 一律视为占位符， + 渲染层需要 fallback 到 label。 + """ + if not poiname: + return True + return len(poiname) >= 2 and poiname.startswith('[') and poiname.endswith(']') + + +def _format_location_text(content): + """Parse type=48 (位置) XML into a compact human-readable line. + + Source XML carries 15+ attrs (poiBusinessHour, poiPhone, adcode, buildingId, + floorName, maptype, scale, fromusername, infourl, ...) but the chat-log line + only needs 3 signals: category (顶层 poiCategoryTips 主类，形如 "主类")， + poiname (POI 名)，label (地址串)。经纬度精度数字进单行只会污染 LLM context； + 电话/营业时间/价格/POI id/adcode/buildingId/floorName/version 留给 + ``decode_location`` 结构化工具。 + + Fallback chain: + 1) 节点缺失 → None (caller 退到 "[位置]") + 2) poiname 是 "[位置]"/"[Location]" 类占位符 → 用 label (用户手扔图钉场景) + 3) poiname 与 label 都缺 → "[位置]" (不堆 lat/lng 数字) + """ + info = _extract_location_info(content) + if not info: + return None + + category = info['category_top'] + head = f"[位置·{category}]" if category else "[位置]" + poiname = info['poiname'] + label = info['label'] + + if _is_location_poiname_placeholder(poiname): + # 用户手扔图钉：poiname 是占位符，label 才是用户看到的描述 + return f"{head} {label}" if label else head + + if not poiname and not label: + return "[位置]" + if poiname and label and poiname != label: + return f"{head} {poiname} @ {label}" + return f"{head} {poiname or label}" + + def _format_app_message_text(content, local_type, is_group, chat_username, chat_display_name, names): if not content or ' str: return "\n".join(lines) +@mcp.tool() +def decode_location(chat_name: str, local_id: int, create_time: int = 0) -> str: + """读取微信位置消息 (base_type=48) 的结构化信息。 + + 返回 POI 名、地址、品类、电话、营业时间、价格档位、城市/区划码、POI id、 + 经纬度等。get_chat_history 渲染的 ``[位置·xxx] poiname @ address`` 只挑了 + 3 个信号；本工具给所有字段。 + + 使用流程：先用 get_chat_history 找到 [位置·xxx] 行 (local_id=N, ts=T)， + 把 N 和 T 一起传进来。create_time(ts) 用于跨分片场景下唯一定位。 + + Args: + chat_name: 聊天对象的名字、备注名或 wxid + local_id: 位置消息的 local_id (从 get_chat_history 获取) + create_time: 消息的 unix 时间戳，从 get_chat_history 输出 ts=N 部分获取。 + 用于在 local_id 跨分片冲突时唯一定位；传 0 时若多个分片含同 local_id 会报歧义错误 + """ + try: + local_id = int(local_id) + create_time = int(create_time) + except (TypeError, ValueError): + return "错误: local_id 和 create_time 必须是整数" + + username = resolve_username(chat_name) + if not username: + return f"找不到聊天对象: {chat_name}" + + shards = _find_msg_tables_for_user(username) + if not shards: + return f"找不到 {chat_name} 的消息表" + + matches = [] + for shard in shards: + if not _is_safe_msg_table_name(shard['table_name']): + continue + with closing(sqlite3.connect(shard['db_path'])) as conn: + if create_time: + candidate_row = conn.execute( + f"SELECT local_type, create_time, message_content, WCDB_CT_message_content " + f"FROM [{shard['table_name']}] WHERE local_id=? AND create_time=?", + (local_id, create_time) + ).fetchone() + else: + candidate_row = conn.execute( + f"SELECT local_type, create_time, message_content, WCDB_CT_message_content " + f"FROM [{shard['table_name']}] WHERE local_id=?", + (local_id,) + ).fetchone() + if candidate_row: + matches.append((shard['db_path'], candidate_row)) + + if not matches: + if create_time: + return f"找不到 (local_id={local_id}, create_time={create_time}) 的消息 (已扫描 {len(shards)} 个分片)" + return f"找不到 local_id={local_id} 的消息 (已扫描 {len(shards)} 个分片)" + if len(matches) > 1: + details = [] + for db_p, r in matches: + ct = r[1] + ts_str = datetime.fromtimestamp(ct).isoformat() if ct else '?' + details.append(f"{os.path.basename(db_p)} create_time={ct} ({ts_str})") + return ( + f"local_id={local_id} 在 {len(matches)} 个分片中都存在，无法唯一定位:\n " + + '\n '.join(details) + + f"\n请加 create_time 参数：decode_location(chat_name, local_id={local_id}, create_time=N)" + ) + + _, row = matches[0] + local_type, _msg_create_time, content, ct_compress = row + base_type, _ = _split_msg_type(local_type) + if base_type != 48: + return ( + f"不是位置消息 (local_type={local_type}, base_type={base_type})，" + f"位置消息应为 base_type=48" + ) + + xml_text = _decompress_content(content, ct_compress) + if not xml_text: + return "消息 content 为空或无法解码" + + is_group = username.endswith('@chatroom') + _, xml_text = _parse_message_content(xml_text, local_type, is_group) + + info = _extract_location_info(xml_text) + if info is None: + return "消息是 type=48 但缺 节点 (schema 异常)" + + lines = ["位置消息:"] + if info['poiname']: + lines.append(f" POI 名: {info['poiname']}") + if info['label']: + lines.append(f" 地址: {info['label']}") + if info['poiCategoryTips']: + lines.append(f" 品类: {info['poiCategoryTips']}") + if info['poiPhone']: + lines.append(f" 电话: {info['poiPhone']}") + if info['poiBusinessHour']: + lines.append(f" 营业时间: {info['poiBusinessHour']}") + if info['poiPriceTips']: + lines.append(f" 价格档位: {info['poiPriceTips']}") + if info['cityname']: + lines.append(f" 城市: {info['cityname']}") + if info['adcode']: + lines.append(f" 行政区划码: {info['adcode']}") + if info['buildingId']: + lines.append(f" buildingId: {info['buildingId']}") + if info['floorName']: + lines.append(f" 楼层: {info['floorName']}") + if info['poiid']: + lines.append(f" POI id: {info['poiid']}") + if info['isFromPoiList']: + lines.append(f" 来源: {info['isFromPoiList']} (true/1=用户从 POI 列表选择，false/0=手扔图钉)") + if info['lat'] is not None and info['lng'] is not None: + lines.append(f" 经纬度: ({info['lat']:.6f}, {info['lng']:.6f}) # 微信 x→纬度，y→经度") + # defensive 字段：本 corpus 实测 0% 非空，但别家账号可能填；仅在非空时展示 + if info['infourl']: + lines.append(f" infourl: {info['infourl']}") + if info['version']: + lines.append(f" version: {info['version']}") + return "\n".join(lines) + + @mcp.tool() def get_chat_images(chat_name: str, limit: int = 20, offset: int = 0, start_time: str = "", end_time: str = "") -> str: """列出某个聊天中的图片消息。 diff --git a/tests/test_location_message.py b/tests/test_location_message.py new file mode 100644 index 0000000..27dcc0e --- /dev/null +++ b/tests/test_location_message.py @@ -0,0 +1,173 @@ +"""测试 type=48 (微信位置消息) 解析与渲染。 + +5 个 fixture **全部合成**，无任何真实地理 / 商户 / 行政区划信息。设计原则：保留 +真实 schema 的字段形态 (attr 顺序、占位符、空字符串、enum 漂移、多段 category)， +但 POI 名 / 地址 / 城市名 / phone / 坐标全部用合成占位符。fixture 的能力是验证 +解析与渲染逻辑，无需绑定任何具体真实数据。 + +覆盖 5 种 schema 形态： + + A: 全字段 new schema (有 poiCategoryTips/poiPhone/adcode/cityname, label 空) + B: 带 infourl 占位的 mid schema (infourl 出现但全 corpus 0% 非空, label 空) + C: qqmap_ poiid 变体 + buildingId/floorName 实际填了的稀有 case (poiname+label 都填) + D: 极简 old schema + poiname="[位置]" 占位符 (用户手扔图钉，必须 fallback 到 label) + E: maptype="0" + 三段 poiCategoryTips + adcode/cityname 都空 +""" +import unittest + +from mcp_server import ( + _extract_location_info, + _format_location_text, + _is_location_poiname_placeholder, +) + + +FIXTURE_A = ''' + +\t +''' + +FIXTURE_B = '''''' + +FIXTURE_C = ''' + +\t +''' + +FIXTURE_D = ''' +\t +''' + +FIXTURE_E = ''' +\t +''' + + +class FormatLocationRenderTests(unittest.TestCase): + """单行 chat-history 渲染：只挑 category / poiname / label 三个信号。""" + + def test_a_full_schema_label_empty(self): + # label 空 → 渲染只用 category + poiname，不带 @ 地址段 + self.assertEqual(_format_location_text(FIXTURE_A), '[位置·示例主类A] 示例POI-A') + + def test_b_mid_schema_with_empty_infourl(self): + # infourl 是 schema 占位符，不影响渲染 + self.assertEqual(_format_location_text(FIXTURE_B), '[位置·示例主类B] 示例POI-B') + + def test_c_full_schema_with_address(self): + # poiname 与 label 都填，label 是地址 → 用 "@ 地址" 拼接 + self.assertEqual( + _format_location_text(FIXTURE_C), + '[位置·示例主类C] 示例POI-C @ 示例市示例区示例路 1 号示例楼 L5' + ) + + def test_d_placeholder_poiname_falls_back_to_label(self): + # poiname="[位置]" 是手扔图钉时客户端填的占位符，必须用 label 渲染 + self.assertEqual(_format_location_text(FIXTURE_D), '[位置] 示例区(近示例公交站)') + + def test_e_maptype_zero_still_renders(self): + # maptype="0" 是另一种 enum，不影响渲染选择；poiCategoryTips 有多层（主:子:孙） + # 取顶层主类 + self.assertEqual( + _format_location_text(FIXTURE_E), + '[位置·示例主类E] 示例POI-E @ 示例市示例区示例路 2 号' + ) + + def test_minimal_no_poiname_no_label(self): + # 极端兜底：poiname 与 label 都缺，不堆 lat/lng 数字 + xml = '' + self.assertEqual(_format_location_text(xml), '[位置]') + + def test_no_category(self): + # poiCategoryTips 缺失 → 渲染前缀退到 [位置] (不带 ·xxx) + xml = '' + self.assertEqual(_format_location_text(xml), '[位置] 示例POI-X') + + def test_missing_location_node(self): + # 内没有 → None, caller 退到 "[位置]" 兜底 + self.assertIsNone(_format_location_text('')) + + def test_invalid_xml(self): + self.assertIsNone(_format_location_text('')) + self.assertIsNone(_format_location_text('not xml at all')) + + +class ExtractLocationInfoTests(unittest.TestCase): + """结构化层：decode_location 拿全字段，字段语义边界检查。""" + + def test_a_all_user_shared_fields_extracted(self): + info = _extract_location_info(FIXTURE_A) + self.assertEqual(info['poiname'], '示例POI-A') + self.assertEqual(info['poiid'], 'nearby_000000000000000000') + self.assertEqual(info['poiCategoryTips'], '示例主类A:示例子类') + self.assertEqual(info['category_top'], '示例主类A') + self.assertEqual(info['poiPhone'], '0000-00000000') + self.assertEqual(info['cityname'], '城市A') + self.assertEqual(info['adcode'], '000000') + self.assertEqual(info['isFromPoiList'], 'false') + # 经纬度：x→lat, y→lng + self.assertAlmostEqual(info['lat'], 0.0, places=4) + self.assertAlmostEqual(info['lng'], 0.0, places=4) + # defensive 字段：empty in fixture A but present in dict + self.assertEqual(info['infourl'], '') + self.assertEqual(info['floorName'], '') + + def test_b_empty_infourl_preserved_in_structured_layer(self): + # infourl 在本 corpus 全部空字符串，但 decode_location 仍要暴露 schema slot + info = _extract_location_info(FIXTURE_B) + self.assertEqual(info['infourl'], '') + self.assertEqual(info['poiCategoryTips'], '示例主类B:示例子类') + self.assertEqual(info['category_top'], '示例主类B') + + def test_c_rare_building_floor_filled(self): + # buildingId 和 floorName 在 corpus 里只有 ~1% 非空，但 C 是稀有真实案例 + info = _extract_location_info(FIXTURE_C) + self.assertEqual(info['buildingId'], '000000000000') + self.assertEqual(info['floorName'], 'L5') + self.assertEqual(info['poiBusinessHour'], '00:00-24:00') + self.assertEqual(info['poiPriceTips'], '100') + self.assertEqual(info['poiid'], 'qqmap_000000000000000000') + + def test_d_dropped_pin_minimal(self): + info = _extract_location_info(FIXTURE_D) + self.assertEqual(info['poiname'], '[位置]') + self.assertEqual(info['label'], '示例区(近示例公交站)') + # 字段缺失 → 空串（跟 _extract_transfer_info 风格一致） + self.assertEqual(info['poiCategoryTips'], '') + self.assertEqual(info['poiPhone'], '') + self.assertEqual(info['poiid'], '') + + def test_e_multi_segment_category(self): + # 三段 "主:子:孙" → category_top 仍只取主类 + info = _extract_location_info(FIXTURE_E) + self.assertEqual(info['poiCategoryTips'], '示例主类E:示例子类:示例孙类') + self.assertEqual(info['category_top'], '示例主类E') + + def test_missing_location_node_returns_none(self): + self.assertIsNone(_extract_location_info('')) + self.assertIsNone(_extract_location_info('')) + + def test_invalid_coordinates_become_none(self): + xml = '' + info = _extract_location_info(xml) + self.assertIsNone(info['lat']) + self.assertIsNone(info['lng']) + + +class PlaceholderDetectionTests(unittest.TestCase): + + def test_recognized_placeholders(self): + self.assertTrue(_is_location_poiname_placeholder('[位置]')) + self.assertTrue(_is_location_poiname_placeholder('[Location]')) + self.assertTrue(_is_location_poiname_placeholder('[]')) + self.assertTrue(_is_location_poiname_placeholder('')) + + def test_real_poi_names_not_placeholder(self): + self.assertFalse(_is_location_poiname_placeholder('示例POI-A')) + self.assertFalse(_is_location_poiname_placeholder('示例POI-B')) + # 中括号在 POI 名中间不算占位符 + self.assertFalse(_is_location_poiname_placeholder('示例POI [子]总店')) + + +if __name__ == '__main__': + unittest.main()