feat(msg): 类型归一化扩展(spike判别+49拆红包/转账)+存量回填+API msg_type/q过滤

- _normalize_msg_type 升级 (raw_type, row): messageType 经典码优先 → 49/appmsg 行级拆分(红包/转账/聊天记录) → 云客码兜底 - spike 修正: 云客 type=3 实为语音(messageType=34, 8754/8955 行), 原 image_legacy 是误标; type_raw=3 的 201 行旧数据才是真图片 - 新映射: 5红包 6/7通话 9文件 10/18/20链接 11转账 13名片 14位置 15/10000系统 21引用回复→text 22/19 证据不足留 unknown - 回填 smoke.db: 620279 行重扫, 94268 行修正, unknown 87198→1885, 幂等 - GET /api/messages 加 msg_type(Literal 422)+q(LIKE 转义)参数 - 新增 test_scrape_msg_type.py(11) + test_api_messages 追加(8), 全量 466 绿 Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-25 15:17:21 +08:00
parent f6ddd0263a
commit 64577a1b02
5 changed files with 537 additions and 17 deletions
@@ -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:
@@ -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())
@@ -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 里的 <type> 子码(仅 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"<type>(\d+)</type>", text)
+    if m:
+        return _APPMSG_XML_TYPE.get(m.group(1), "unknown")
+    return "file"


 # ---------------------------------------------------------------------------
@@ -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"]
@@ -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 里的 <type>19</type> 聊天记录(合并转发, spike 1694+7 行)。"""
+    row = {
+        "messageType": "49", "type": 21,
+        "text": "<msg><appmsg><title>聊天记录</title><type>19</type><recorditem><![CDATA[...]]></recorditem></appmsg></msg>",
+    }
+    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": "<msg><appmsg><type>2001</type>"}) == "red_packet"
+    assert _normalize_msg_type("49", row={"text": "<msg><appmsg><type>2000</type>"}) == "transfer"
+    assert _normalize_msg_type("49", row={"paymsgid": "abc", "text": "<msg><appmsg><type>2001</type>"}) == "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": "<msg><appmsg><type>19</type>"}) == "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