diff --git a/yunke-server/app/api/routers/messages.py b/yunke-server/app/api/routers/messages.py index 58a6297..27cfdf5 100644 --- a/yunke-server/app/api/routers/messages.py +++ b/yunke-server/app/api/routers/messages.py @@ -8,6 +8,10 @@ Read-only query over the mirrored ``message`` table. Filters: may be supplied (AND), but typically the frontend sends exactly one. * ``start`` / ``end`` — ms-epoch window compared against ``timestamp`` (inclusive on both ends). +* ``msg_type`` — normalized type label (``text`` / ``image`` / ..., + see ``_MSG_TYPE_FILTER``). Unknown values are rejected with 422. +* ``q`` — substring match on ``text`` (``%``/``_`` escaped + as literals). Ordered by ``timestamp asc`` so the timeline renders oldest-first. The index ``ix_message_channel_owner_talker_time`` covers the hot path @@ -16,7 +20,7 @@ Ordered by ``timestamp asc`` so the timeline renders oldest-first. The index from __future__ import annotations -from typing import Annotated +from typing import Annotated, Literal import sqlalchemy as sa from fastapi import APIRouter, Depends, Query @@ -32,6 +36,36 @@ SessionDep = Annotated[AsyncSession, Depends(get_session)] PageParam = Annotated[int, Query(ge=1)] PageSizeParam = Annotated[int, Query(ge=1, le=100)] +# msg_type 参数值 → 覆盖的 Message.type 标签集合(对齐云客 10 类下拉; +# mini_program 暂无样本行也保留, 查空是数据现实) +_MSG_TYPE_FILTER: dict[str, tuple[str, ...]] = { + "text": ("text",), + "image": ("image", "image_legacy"), + "voice": ("voice",), + "video": ("video",), + "link": ("link",), + "file": ("file",), + "red_packet": ("red_packet",), + "transfer": ("transfer",), + "voice_call": ("voice_call",), + "video_call": ("video_call",), + "mini_program": ("mini_program",), +} + +MsgTypeParam = Literal[ + "text", + "image", + "voice", + "video", + "link", + "file", + "red_packet", + "transfer", + "voice_call", + "video_call", + "mini_program", +] + @router.get("/messages") async def list_messages( @@ -44,6 +78,8 @@ async def list_messages( room_id: Annotated[str | None, Query()] = None, start: Annotated[int | None, Query(ge=0)] = None, end: Annotated[int | None, Query(ge=0)] = None, + msg_type: Annotated[MsgTypeParam | None, Query()] = None, + q: Annotated[str | None, Query()] = None, ) -> Page[MessageOut]: """List messages for a timeline (ordered by ``timestamp asc``).""" conditions = [] @@ -59,6 +95,12 @@ async def list_messages( conditions.append(Message.timestamp >= start) if end is not None: conditions.append(Message.timestamp <= end) + if msg_type is not None: + conditions.append(Message.type.in_(_MSG_TYPE_FILTER[msg_type])) + if q: + # LIKE 通配符转义: %/_ 按字面匹配 + escaped = q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + conditions.append(Message.text.like(f"%{escaped}%", escape="\\")) total_stmt = sa.select(sa.func.count()).select_from(Message) if conditions: diff --git a/yunke-server/scripts/backfill_msg_types.py b/yunke-server/scripts/backfill_msg_types.py new file mode 100644 index 0000000..9d3bd7a --- /dev/null +++ b/yunke-server/scripts/backfill_msg_types.py @@ -0,0 +1,102 @@ +"""Backfill Message.type via the extended _normalize_msg_type (no re-scrape). + +重扫范围: channel='wechat' AND type_raw IS NOT NULL 的**所有**行 —— 含 +type='unknown'/'file'(主要人群)与历史误标(如 type_raw=3 被标 image_legacy, +实为 messageType=34 语音)。逐行用 raw_json(含 messageType 经典码 + type +云客码)重算, 仅 UPDATE 与现值不同的行: 幂等, 重跑 0 改动; 已正确的类型 +不受影响。WhatsApp 渠道是另一套 type 编码(见 app/scraper/whatsapp.py), +不在本脚本范围。 + +用法: + cd yunke-server && .venv/Scripts/python.exe -m scripts.backfill_msg_types +""" +from __future__ import annotations + +import asyncio +import json +import time +from collections import Counter, defaultdict + +from sqlalchemy import func, select, update +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.db import models # noqa: F401 (ensure tables are registered) +from app.db.models import Message +from scripts.scrape_via_page import _normalize_msg_type + +BATCH = 5000 # 扫描批大小(按 msg_id 主键游标分页) +UPDATE_CHUNK = 400 # 单条 IN (...) 的参数数, 低于 SQLite 999 变量上限 + + +async def main(db_url: str = "sqlite+aiosqlite:///./smoke.db") -> None: + engine = create_async_engine(db_url) + sm = async_sessionmaker(engine, expire_on_commit=False) + scanned = 0 + changed = 0 + by_new_type: Counter[str] = Counter() + by_old_type: Counter[str] = Counter() + async with sm() as s: + last_id = "" + while True: + rows = ( + await s.execute( + select(Message.msg_id, Message.type_raw, Message.type, Message.raw_json) + .where( + Message.channel == "wechat", + Message.type_raw.is_not(None), + Message.msg_id > last_id, + ) + .order_by(Message.msg_id) + .limit(BATCH) + ) + ).all() + if not rows: + break + pending: defaultdict[str, list[str]] = defaultdict(list) + for mid, tr, cur, rj in rows: + last_id = mid + scanned += 1 + if isinstance(rj, dict): + row = rj + elif rj: + try: + row = json.loads(rj) + except (ValueError, TypeError): + continue # raw_json 损坏, 留给下次抓取刷新 + if not isinstance(row, dict): + continue + else: + row = {} + new = _normalize_msg_type(tr, row=row) + if new is None or new == cur: + continue # 幂等: 已正确的行不动 + pending[new].append(mid) + changed += 1 + by_new_type[new] += 1 + by_old_type[cur or "NULL"] += 1 + for new_type, ids in pending.items(): + for i in range(0, len(ids), UPDATE_CHUNK): + await s.execute( + update(Message) + .where(Message.msg_id.in_(ids[i : i + UPDATE_CHUNK])) + .values(type=new_type) + ) + await s.commit() + print(f"[{time.strftime('%H:%M:%S')}] scanned={scanned} changed={changed}...") + # 回填后各 type 分布(wechat 渠道全景) + final = ( + await s.execute( + select(Message.type, func.count()) + .where(Message.channel == "wechat") + .group_by(Message.type) + ) + ).all() + print(f"DONE — scanned={scanned}, updated={changed}") + print("updated -> new type:", dict(by_new_type.most_common())) + print("updated <- old type:", dict(by_old_type.most_common())) + print("wechat type distribution after backfill:", dict(final)) + await engine.dispose() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py index b5b3538..3fa4228 100644 --- a/yunke-server/scripts/scrape_via_page.py +++ b/yunke-server/scripts/scrape_via_page.py @@ -28,6 +28,7 @@ import asyncio import json import logging import os +import re from typing import Any from urllib.parse import urlparse @@ -735,7 +736,7 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100, full_hi bool(row["isSend"]) if row.get("isSend") is not None else (bool(row["mine"]) if row.get("mine") is not None else None) ), - "type": _normalize_msg_type(row.get("type")), + "type": _normalize_msg_type(row.get("type"), row=row), "type_raw": row.get("type"), "text": row.get("content") or row.get("text") or "", "file_url": row.get("file") or row.get("fileUrl"), @@ -827,7 +828,7 @@ def _normalize_group_message(row: dict, owner_id: str, room_id: str) -> dict | N bool(row["isSend"]) if row.get("isSend") is not None else (bool(row["mine"]) if row.get("mine") is not None else None) ), - "type": _normalize_msg_type(row.get("type")), + "type": _normalize_msg_type(row.get("type"), row=row), "type_raw": row.get("type"), "text": row.get("content") or row.get("text") or "", "file_url": row.get("file") or row.get("fileUrl"), @@ -972,25 +973,140 @@ async def scrape_group_messages(cookie_str: str, max_groups: int = 500, full_his return total_fetched, total_upserted -def _normalize_msg_type(raw_type) -> str | None: +def _normalize_msg_type(raw_type, row: dict | None = None) -> str | None: """Normalize a raw message type to our canonical type string. - 微信 type=2 = 图片(新版,fileTh 带完整 OSS URL;真机 A**杰对话确认 - 2026-08-20——旧映射漏掉它导致图片被存成 unknown 前端不渲染)。 + 判别依据 smoke.db 2026-08-25 spike(每类 ≥3 条样本): 云客 API 行里 + ``messageType`` 是经典微信协议码、``type`` 是云客自编码, 两者同在 + raw_json。优先级: messageType 经典码直判 → 49/appmsg 家族行级拆分 + (云客子码/paymsgid/text XML) → 云客码兜底(合成 flag 码时 type 才是 + 唯一可靠来源)。 + + spike 修正: 云客 type=3 的行 8754/8955 是 messageType='34'(经典语音 + 码, file=语音十六进制键) —— 旧 "3"→image_legacy 是误标, 现指向 + voice; messageType='3' 的行(旧 orchestrator 路径写入)才是真图片。 """ + row = row or {} if raw_type is None: - return None + raw_type = row.get("type") + if raw_type is None: + return None t = str(raw_type) - mapping = { - "1": "text", - "2": "image", # 新版图片(fileTh URL) - "3": "image_legacy", # 旧版图片(md5/键) - "34": "voice", - "43": "video", - "47": "emoji", - "49": "file", - } - return mapping.get(t, "unknown") + + mt_raw = row.get("messageType") + mt = str(mt_raw) if mt_raw is not None else None + + # --- 1) appmsg 家族(经典码 49): 红包/转账/聊天记录/链接/文件 --- + if mt == "49" or t == "49": + return _normalize_appmsg_type(t, row) + + # --- 2) 经典微信协议码直判 --- + if mt is not None: + if mt == "50": + # VOIP 通话: 云客 6=语音通话(2077 行)/7=视频通话(77 行, 频率推断) + return {"6": "voice_call", "7": "video_call"}.get(t, "voice_call") + classic = _CLASSIC_MSG_TYPE.get(mt) + if classic is not None: + return classic + + # --- 3) 云客自编码兜底(messageType 是合成 flag 码或缺失); 无行信息时 + # 也查经典码表(row-less 调用, 如单测里的 "34"/"42"/"10000") --- + for candidate in (row.get("type"), t): + if candidate is None: + continue + s = str(candidate) + if s in _YUNKE_MSG_TYPE: + return _YUNKE_MSG_TYPE[s] + if s in _CLASSIC_MSG_TYPE: + return _CLASSIC_MSG_TYPE[s] + return "unknown" + + +# 经典微信协议码(raw_json.messageType)——spike 样本证实 +_CLASSIC_MSG_TYPE = { + "1": "text", # 473922 行 + "3": "image", # 50355 行(md5+OSS fileTh URL) + "34": "voice", # 云客 type=3 的主体人群(8754 行, 语音十六进制键) + "43": "video", # 4835 行 + "47": "emoji", # 4242 行(tc.qq.com 表情 CDN) + "42": "namecard", # 953 行(text=昵称, file=头像) + "48": "location", # 651 行(text=地址, file="lat,lng") + "10000": "system", # 13820 行(打招呼/已添加提示) +} + +# 云客自编码(raw_json.type)——spike 样本证实; 22(1880 行无 text 无字段)/ +# 19(仅 5 行)证据不足, 宁留 unknown +_YUNKE_MSG_TYPE = { + "1": "text", + "2": "image", # messageType=3 + "3": "voice", # messageType=34(旧 image_legacy 为误标) + "4": "video", # messageType=43 + "5": "red_packet", # messageType=436207665, file=wx.gtimg.com/hongbao 封面 + "6": "voice_call", # messageType=50 VOIP + "7": "video_call", # messageType=50 VOIP + "8": "emoji", # messageType=47 + "9": "file", # messageType=49, text=文件名(xlsx/pdf) + "10": "link", # messageType=49, 92% 带封面图 + "11": "transfer", # messageType=419430449, text="发送转账:¥…" + "13": "namecard", # messageType=42 + "14": "location", # messageType=48 + "15": "system", # messageType=10000/268445456/285222674 + "18": "link", # messageType=49, 99.8% 带封面图(公众号分享) + "20": "link", # messageType=754974769, 视频号分享(封面+话题文本) + "21": "text", # messageType=822083633 引用回复(referMsg) +} + +# appmsg 家族内云客子码(row.type)——spike 样本: 9/10/18/21(+5/11 同语义码); +# 不复用 _YUNKE_MSG_TYPE 全表, 避免假想组合(如 mt=49+type=6)错判成通话 +_APPMSG_SUB_TYPE = { + "5": "red_packet", + "9": "file", + "10": "link", + "11": "transfer", + "18": "link", +} + +# appmsg text XML 里的 子码(仅 49 家族无云客子码时的老数据兜底) +_APPMSG_XML_TYPE = { + "5": "link", + "6": "file", + "19": "merged_forward", # 聊天记录(合并转发), spike 1694+7 行 + "2000": "transfer", + "2001": "red_packet", +} + + +def _normalize_appmsg_type(t: str, row: dict) -> str: + """49/appmsg 家族的行级拆分(红包/转账/聊天记录/链接/文件)。 + + 云客子码 row.type 优先(spike: 9=file 10/18=link 21+XML=聊天记录); + 老数据缺子码时按 paymsgid(实测全空, 保留兼容)与 text XML 子码拆; + 全缺 → file(现状回退)。未知子码不猜 → unknown。 + """ + text = str(row.get("text") or "") + sub = row.get("type") + if sub is not None and str(sub) != "49": + s = str(sub) + if s == "21": + # 云客 21 在 appmsg 家族里=聊天记录 XML(messageType=822083633 + # 的引用回复不走本分支) + return "merged_forward" + return _APPMSG_SUB_TYPE.get(s, "unknown") + if row.get("paymsgid"): + if ">2001<" in text: + return "red_packet" + if ">2000<" in text: + return "transfer" + # paymsgid 有值但 XML 无 type → transfer(转账才有 paymsgid 概率高) + return "transfer" + if ">2001<" in text: + return "red_packet" + if ">2000<" in text: + return "transfer" + m = re.search(r"(\d+)", text) + if m: + return _APPMSG_XML_TYPE.get(m.group(1), "unknown") + return "file" # --------------------------------------------------------------------------- diff --git a/yunke-server/tests/test_api_messages.py b/yunke-server/tests/test_api_messages.py index 852487e..f9ab581 100644 --- a/yunke-server/tests/test_api_messages.py +++ b/yunke-server/tests/test_api_messages.py @@ -360,3 +360,119 @@ async def test_media_fields_default_none_when_unset(make_client): ): assert key in it assert it[key] is None + + +# --------------------------------------------------------------------------- +# msg_type / q 过滤(Task 4: 类型归一化扩展 + API 过滤, 对齐云客 10 类下拉) +# --------------------------------------------------------------------------- + + +async def _seed_types(session: AsyncSession) -> None: + """text / image / image_legacy / red_packet / transfer 各一条 + 关键词行。""" + session.add_all( + [ + Message( + msg_id="ty_text", channel="wechat", owner_account_id="wx_owner_1", + talker_id="t_a", text="报价 135800 元", timestamp=1000, is_send=True, + type="text", type_raw=1, + ), + Message( + msg_id="ty_img", channel="wechat", owner_account_id="wx_owner_1", + talker_id="t_a", text="", timestamp=2000, is_send=False, + type="image", type_raw=2, + ), + Message( + msg_id="ty_img_legacy", channel="wechat", owner_account_id="wx_owner_1", + talker_id="t_a", text="", timestamp=2500, is_send=False, + type="image_legacy", type_raw=3, + ), + Message( + msg_id="ty_red", channel="wechat", owner_account_id="wx_owner_1", + talker_id="t_a", text="", timestamp=3000, is_send=True, + type="red_packet", type_raw=5, + ), + Message( + msg_id="ty_transfer", channel="wechat", owner_account_id="wx_owner_1", + talker_id="t_a", text="发送转账:¥100.00", timestamp=4000, is_send=True, + type="transfer", type_raw=11, + ), + Message( + msg_id="ty_text_2", channel="wechat", owner_account_id="wx_owner_1", + talker_id="t_a", text="含 100% 通配符字面量", timestamp=5000, is_send=True, + type="text", type_raw=1, + ), + ] + ) + await session.commit() + + +async def test_msg_type_filter_text(make_client): + client = await make_client(_seed_types) + r = await client.get("/api/messages", params={"msg_type": "text"}) + assert r.status_code == 200 + body = r.json() + assert {it["msg_id"] for it in body["items"]} == {"ty_text", "ty_text_2"} + + +async def test_msg_type_filter_image_includes_legacy(make_client): + """msg_type=image 覆盖 image + image_legacy(过渡期兼容)。""" + client = await make_client(_seed_types) + r = await client.get("/api/messages", params={"msg_type": "image"}) + assert r.status_code == 200 + body = r.json() + assert {it["msg_id"] for it in body["items"]} == {"ty_img", "ty_img_legacy"} + + +async def test_msg_type_filter_red_packet_only(make_client): + client = await make_client(_seed_types) + r = await client.get("/api/messages", params={"msg_type": "red_packet"}) + assert r.status_code == 200 + body = r.json() + assert [it["msg_id"] for it in body["items"]] == ["ty_red"] + assert body["total"] == 1 + + +async def test_msg_type_filter_no_rows_returns_empty(make_client): + """voice_call 等暂无样本类型: 过滤语法有效, 查空是数据现实。""" + client = await make_client(_seed_types) + r = await client.get("/api/messages", params={"msg_type": "voice_call"}) + assert r.status_code == 200 + body = r.json() + assert body["items"] == [] + assert body["total"] == 0 + + +async def test_msg_type_unknown_rejected_422(make_client): + client = await make_client(_seed_types) + r = await client.get("/api/messages", params={"msg_type": "bogus"}) + assert r.status_code == 422 + + +async def test_q_substring_filter(make_client): + client = await make_client(_seed_types) + r = await client.get("/api/messages", params={"q": "报价"}) + assert r.status_code == 200 + body = r.json() + assert [it["msg_id"] for it in body["items"]] == ["ty_text"] + assert body["total"] == 1 + + +async def test_q_escapes_like_wildcards(make_client): + """q 里的 %/_ 按字面匹配, 不当通配符。""" + client = await make_client(_seed_types) + r = await client.get("/api/messages", params={"q": "100%"}) + assert r.status_code == 200 + body = r.json() + # 只命中字面含 "100%" 的行; 不因 % 通配漏扩到 "¥100.00" 之外的行 + assert [it["msg_id"] for it in body["items"]] == ["ty_text_2"] + r2 = await client.get("/api/messages", params={"q": "100"}) + ids2 = {it["msg_id"] for it in r2.json()["items"]} + assert ids2 == {"ty_transfer", "ty_text_2"} + + +async def test_msg_type_and_q_combine(make_client): + client = await make_client(_seed_types) + r = await client.get("/api/messages", params={"msg_type": "transfer", "q": "转账"}) + assert r.status_code == 200 + body = r.json() + assert [it["msg_id"] for it in body["items"]] == ["ty_transfer"] diff --git a/yunke-server/tests/test_scrape_msg_type.py b/yunke-server/tests/test_scrape_msg_type.py new file mode 100644 index 0000000..f79838a --- /dev/null +++ b/yunke-server/tests/test_scrape_msg_type.py @@ -0,0 +1,144 @@ +"""Tests for :func:`scripts.scrape_via_page._normalize_msg_type` (extended). + +判别依据 smoke.db 2026-08-25 spike 实测(wechat 渠道): 云客 API 行里 +``messageType`` 是经典微信协议码(1文本/3图片/34语音/43视频/47表情/42名片/ +48位置/49appmsg/50VOIP通话/10000系统), ``type`` 是云客自编码, 两者同在 +raw_json 里。历史 DB 的 type_raw 列是二者混合(新抓取路径存 type, 旧 +orchestrator 路径存 messageType), 因此归一化以 messageType 优先、云客码 +兜底, 并对 49/appmsg 家族做行级拆分(红包/转账/聊天记录/链接/文件)。 +""" + +from __future__ import annotations + +from scripts.scrape_via_page import _normalize_msg_type + + +# --------------------------------------------------------------------------- +# 既有映射(1/2/34/43/47/49)保持不变 + None 透传 +# --------------------------------------------------------------------------- + + +def test_basic_mapping_unchanged(): + assert _normalize_msg_type("1") == "text" + assert _normalize_msg_type("2") == "image" + assert _normalize_msg_type("34") == "voice" + assert _normalize_msg_type("43") == "video" + assert _normalize_msg_type("47") == "emoji" + # 缺行级信息时 49 回退 file(现状) + assert _normalize_msg_type("49") == "file" + assert _normalize_msg_type(None) is None + + +def test_type_3_is_voice_not_image_legacy(): + """spike 修正: 云客 type=3 行 8754/8955 的 messageType='34'(经典语音码, + file=语音十六进制键) —— 原 "3"→image_legacy 是误标(作者把语音键当成了 + 旧版图片键)。messageType='3' 的 201 行(旧路径)才是真图片。""" + assert _normalize_msg_type("3", row={"messageType": "34", "type": 3}) == "voice" + assert _normalize_msg_type("3", row={"messageType": "3", "type": 3}) == "image" + # 无行信息: 3 的主体人群(98%)是语音 + assert _normalize_msg_type("3") == "voice" + + +# --------------------------------------------------------------------------- +# spike 判别出的新映射(每个 code 至少 3 条 smoke.db 样本支撑) +# --------------------------------------------------------------------------- + + +def test_new_mappings_row_based(): + # 云客码兜底表(raw_json.messageType 为合成 flag 码时靠 type 判) + assert _normalize_msg_type("4", row={"messageType": "43", "type": 4}) == "video" + assert _normalize_msg_type("5", row={"messageType": "436207665", "type": 5}) == "red_packet" + assert _normalize_msg_type("6", row={"messageType": "50", "type": 6}) == "voice_call" + assert _normalize_msg_type("7", row={"messageType": "50", "type": 7}) == "video_call" + assert _normalize_msg_type("8", row={"messageType": "47", "type": 8}) == "emoji" + assert _normalize_msg_type("9", row={"messageType": "49", "type": 9}) == "file" + assert _normalize_msg_type("10", row={"messageType": "49", "type": 10}) == "link" + assert _normalize_msg_type("11", row={"messageType": "419430449", "type": 11}) == "transfer" + assert _normalize_msg_type("13", row={"messageType": "42", "type": 13}) == "namecard" + assert _normalize_msg_type("14", row={"messageType": "48", "type": 14}) == "location" + assert _normalize_msg_type("15", row={"messageType": "10000", "type": 15}) == "system" + assert _normalize_msg_type("18", row={"messageType": "49", "type": 18}) == "link" + assert _normalize_msg_type("20", row={"messageType": "754974769", "type": 20}) == "link" + # 21 引用回复(messageType=822083633 合成码, referMsg 带 quoted 内容) + assert _normalize_msg_type("21", row={"messageType": "822083633", "type": 21}) == "text" + + +def test_classic_message_type_wins(): + """messageType(经典码)优先于云客码/传入 raw_type。""" + assert _normalize_msg_type(3, row={"messageType": "34"}) == "voice" + assert _normalize_msg_type(2, row={"messageType": "3", "type": 2}) == "image" + assert _normalize_msg_type(1, row={"messageType": "1", "type": 1}) == "text" + assert _normalize_msg_type(8, row={"messageType": "47", "type": 8}) == "emoji" + assert _normalize_msg_type(13, row={"messageType": "42", "type": 13}) == "namecard" + assert _normalize_msg_type(14, row={"messageType": "48", "type": 14}) == "location" + assert _normalize_msg_type(15, row={"messageType": "10000", "type": 15}) == "system" + + +def test_voip_call_types(): + """messageType=50(经典 VOIP): 云客 6=语音通话(2077 行) / 7=视频通话(77 行)。""" + assert _normalize_msg_type("6", row={"messageType": "50", "type": 6, "duration": 68}) == "voice_call" + assert _normalize_msg_type("7", row={"messageType": "50", "type": 7, "duration": 107}) == "video_call" + + +def test_quote_reply_maps_to_text(): + """引用回复(822083633)带 referMsg, 正文是纯文本 → text。""" + row = {"messageType": "822083633", "type": 21, "text": "你去咋没交上?", + "referMsg": {"text": "老子自己去也没交上"}} + assert _normalize_msg_type("21", row=row) == "text" + + +def test_merged_forward_chat_record(): + """appmsg 里的 19 聊天记录(合并转发, spike 1694+7 行)。""" + row = { + "messageType": "49", "type": 21, + "text": "聊天记录19", + } + assert _normalize_msg_type("21", row=row) == "merged_forward" + + +def test_evidence_insufficient_stays_unknown(): + """无判别证据的类宁留 unknown: 云客 22(拍一拍类系统 flag, 1880 行全无 text + 全无字段)与 19(仅 5 行空 text)不猜。""" + assert _normalize_msg_type("22", row={"messageType": "922746929", "type": 22}) == "unknown" + assert _normalize_msg_type("19", row={"messageType": "-1879048186", "type": 19}) == "unknown" + assert _normalize_msg_type("9999") == "unknown" + + +# --------------------------------------------------------------------------- +# 49 / appmsg 家族行级拆分(红包/转账) +# --------------------------------------------------------------------------- + + +def test_type_49_redpacket_vs_transfer(): + # 拆分带行级信息: paymsgid 非空 → 按 text XML appmsg type 判, XML 无 type → transfer + assert _normalize_msg_type("49", row={"paymsgid": "abc"}) in ("red_packet", "transfer") + assert _normalize_msg_type("49", row={"paymsgid": "abc"}) == "transfer" + # text 含 >2001< → red_packet; >2000< → transfer + assert _normalize_msg_type("49", row={"text": "2001"}) == "red_packet" + assert _normalize_msg_type("49", row={"text": "2000"}) == "transfer" + assert _normalize_msg_type("49", row={"paymsgid": "abc", "text": "2001"}) == "red_packet" + # 子码直判(云客 type 字段) + assert _normalize_msg_type("49", row={"type": "10"}) == "link" + assert _normalize_msg_type("49", row={"type": "9"}) == "file" + assert _normalize_msg_type("49", row={"type": "18"}) == "link" + # 缺信息回退现有行为 + assert _normalize_msg_type("49", row={}) == "file" + # 未知子码(无样本)不猜成 file 之外的语义 + assert _normalize_msg_type("49", row={"type": "777"}) == "unknown" + + +def test_legacy_classic_49_rows_use_yunke_subcode(): + """DB 里 202 条 type_raw=49 旧行(orchestrator 路径写入) raw_json.type + 已是云客子码(10/21/18/9) —— 子码优先拆分。""" + assert _normalize_msg_type("49", row={"messageType": "49", "type": 10}) == "link" + assert _normalize_msg_type("49", row={"messageType": "49", "type": 18}) == "link" + assert _normalize_msg_type("49", row={"messageType": "49", "type": 9}) == "file" + assert _normalize_msg_type("49", row={"messageType": "49", "type": 21, + "text": "19"}) == "merged_forward" + + +def test_row_only_without_raw_type(): + """raw_type 缺失时从 row.type 取(spike 后调用点都传整行)。""" + assert _normalize_msg_type(None, row={"type": 1, "messageType": "1"}) == "text" + assert _normalize_msg_type(None, row={"type": 5, "messageType": "436207665"}) == "red_packet" + assert _normalize_msg_type(None, row={}) is None