chuanqi/protocol_bot/scripts/watch_sourid_live.py

#!/usr/bin/env python3
"""watch_sourid_live.py — 实时监视 pcap/jsonl 抓包, 解出全部地图 sourid。

用法(等待用户进图时后台跑):
  sudo protocol_bot/.venv/bin/python protocol_bot/scripts/watch_sourid_live.py \
      --out /tmp/cap_wugong.jsonl

边抓边解: 每 2s 扫描新增行, SC-1 批量帧 zlib 解压 → SC51/SC634 decode6
→ 打印新发现的 MapName→sourid。Ctrl-C 结束。
"""
from __future__ import annotations

import json
import sys
import time
import zlib

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


def scan_file(path: str, seen: dict[str, dict]) -> None:
    try:
        with open(path, encoding="utf-8", errors="replace") as f:
            for line in f:
                if '"msg": -1' not in line and '"msg": 51' not in line \
                        and '"msg": 634' not in line:
                    continue
                try:
                    rec = json.loads(line)
                except ValueError:
                    continue
                msg, bh = rec.get("msg"), rec.get("body_hex", "")
                if msg not in (-1, 51, 634) or not bh:
                    continue

                def handle(m: int, b: bytes) -> None:
                    if m == -1:
                        try:
                            raw = zlib.decompress(b)
                        except zlib.error:
                            return
                        for f2 in FrameSplitter().feed(raw):
                            handle(f2.msg_id, f2.body)
                        return
                    if m not in (51, 634):
                        return
                    try:
                        text = decode6(b).decode("utf-8", errors="replace")
                        s = text.find("{")
                        if s < 0:
                            return
                        j = json.loads(text[s:text.rfind("}") + 1])
                        name = str(j.get("MapName") or j.get("MapId") or "?")
                        if name in seen and seen[name].get("_m") == "SC51":
                            return
                        j["_m"] = f"SC{m}"
                        seen[name] = j
                    except Exception:
                        pass

                handle(msg, bytes.fromhex(bh))
    except OSError:
        pass


def main() -> int:
    out = "/tmp/cap_wugong.jsonl"
    if "--out" in sys.argv:
        out = sys.argv[sys.argv.index("--out") + 1]
    print(f"监视 {out} — 等待 SC51/SC634 地图属性帧(Ctrl-C 结束)")
    print(f"{'时间':<9} {'显示名':<14} {'MapId':<14} {'sourid':<8} Minimap")
    print("-" * 68)
    seen: dict[str, dict] = {}
    try:
        while True:
            scan_file(out, seen)
            time.sleep(2)
    except KeyboardInterrupt:
        print("\n==== 最终结果 ====")
        for name in sorted(seen):
            j = seen[name]
            print(f"{name:<14} MapId={j.get('MapId','')} "
                  f"sourid={j.get('sourid','')} Minimap={j.get('Minimap','')} {j['_m']}")
    return 0


if __name__ == "__main__":
    sys.exit(main())