chuanqi/protocol_bot/scripts/scan_sourid.py

#!/usr/bin/env python3
"""scan_sourid.py — 从存量抓包 jsonl 提取每张图的 SC51/SC634 sourid。

SC51/SC634 body 是 encode6 编码 JSON(含 MapName/MapId/sourid/Minimap)。
body_hex 字段是编码后字节 → decode6 → JSON → 提取。

用法: uv run python scripts/scan_sourid.py  (protocol_bot/ 下执行)
"""
from __future__ import annotations

import glob
import json
import sys

sys.path.insert(0, "src")
from pbot.codec.bitbuf import decode6  # noqa: E402


def try_decode_body(hexstr: str) -> dict | None:
    try:
        raw = decode6(bytes.fromhex(hexstr))
        text = raw.decode("utf-8", errors="replace")
        start = text.find("{")
        if start < 0:
            return None
        return json.loads(text[start:text.rfind("}") + 1])
    except Exception:
        return None


def main() -> None:
    results: dict[str, dict] = {}
    for path in sorted(glob.glob("/tmp/cap_*.jsonl")):
        try:
            with open(path, encoding="utf-8", errors="replace") as f:
                for line in f:
                    if '"msg": 51,' not in line and '"msg": 634,' not in line:
                        continue
                    try:
                        rec = json.loads(line)
                    except ValueError:
                        continue
                    if rec.get("msg") not in (51, 634):
                        continue
                    body = rec.get("body_hex")
                    if not body:
                        continue
                    j = try_decode_body(body)
                    if not j:
                        continue
                    name = j.get("MapName") or j.get("MapId")
                    if not name:
                        continue
                    key = f"{name}"
                    if key in results and results[key].get("_src") == "SC51":
                        continue
                    j["_src"] = f"SC{rec['msg']}"
                    j["_file"] = path.split("/")[-1]
                    results[key] = j
        except OSError:
            continue

    print(f"{'显示名':<12} {'MapId':<12} {'sourid':<8} {'Minimap':<8} 来源")
    print("-" * 70)
    for name in sorted(results):
        j = results[name]
        print(f"{name:<12} {str(j.get('MapId','')):<12} "
              f"{str(j.get('sourid','')):<8} {str(j.get('Minimap','')):<8} "
              f"{j['_src']} @{j['_file']}")


if __name__ == "__main__":
    main()