#!/usr/bin/env python3
"""虎牙 live-launch doLaunch 复放器 —— GUID(sGuid) 铸币机原型 (PC 侧, 离线自检先行).

背景 (docs/HUYA_HDID_ALGORITHM_GEN.md §11):
  32hex 登录帧 hdid = 服务端 doLaunch 下发的 sGuid (LiveLaunchRsp.tag0),
  App 经 HalImpl.sGuidProperty → WupHelper.getGuid → HyDeviceProxy.setAppInfoId(pnc.a)
  → NativeBridge.b(100) → native getGUID 缓存回写, 全程无本地公式.
  铸币 = 构造虚拟设备指纹的 doLaunch 请求 → 服务端签发 sGuid → 登录帧 t1.t0 用它.

本工具:
  - build_live_launch_wup(profile) : UniPacket 信封 + LiveLaunchReq JCE (规格来自
    classes9/com/duowan/HUYA/{LiveLaunchReq,LiveUserbase,LiveAppUAEx,LiveLaunchRsp}
    + classes11 UserId, servant=@WupServant("launch") func=doLaunch)
  - parse_launch_rsp(bytes)        : 解 UniPacket → sBuffer → LiveLaunchRsp → sGuid
  - --self-test                    : 编码→解码回环自检 (不联网)
  - --dump                         : 打印请求体结构/hex (不联网)
  - --live [url]                   : 真实发送 (默认 https://wup.huya.com)

实测状态 (2026-08-29):
  - 信封/路径/键均被服务端接受: servant/func 回声, UniAttribute 找到 tReq 键;
    但 tReq 值被拒: "read 'struct' type mismatch, tag: 0, get type: 12" (恒定).
  - 诊断: 值=0x0c 复现同错 (E:空值->require field; F:缺键->not found key) => 服务端在
    解析值, 但首字段始终判为 ZERO. 疑点: a09 传输类型由服务端动态配置
    (IFunctionTranspotModule.getTransportType("launch#doLaunch")), HTTP POST 路径
    未必是 doLaunch 真渠道; wire 已逐字节对齐 App (见 docs §11.6), 待真机帧定案.

用法:
    python tools/huya_launch_mint.py --self-test
    python tools/huya_launch_mint.py --dump --mid a1b2c3d4e5f60718 --imei 860000000000000
    python tools/huya_launch_mint.py --live [url]
"""
from __future__ import annotations

import argparse
import re
import struct
import sys
from pathlib import Path

HERE = Path(__file__).resolve().parent
if str(HERE) not in sys.path:
    sys.path.insert(0, str(HERE))

from huya_wup_encoder import _Writer  # noqa: E402

try:
    from core.huya.taf_protocol import TafInputStream, TafType  # noqa: E402
    HAVE_TAF = True
except Exception:  # pragma: no cover
    HAVE_TAF = False

WUP_URL = "https://wup.huya.com"
SERVANT = "launch"
FUNC = "doLaunch"

# ---------------------------------------------------------------------------
# JCE 编码 (写端) —— 规格来自 dex writeTo()
# ---------------------------------------------------------------------------


def _w_string_or_skip(w: _Writer, tag: int, v: str | None) -> None:
    if v:
        w.string(tag, v)


def encode_live_launch_req(profile: dict) -> bytes:
    """LiveLaunchReq: t0=tId(UserId) t1=tLiveUB(LiveUserbase) t2=bSupportDomain(int16).

    profile 键: mid, imei, device_id, guid(当前sGuid, 新设备=空), huya_ua,
                model, qimei, luid, apn, net_type
    """
    w = _Writer()
    # 外层 LiveLaunchReq struct (tag0) —— 缺它整体结构上移一层
    w.struct_begin(0)
    w.struct_begin(0)  # UserId tId
    w.int64(0, int(profile.get("luid", 0)))
    _w_string_or_skip(w, 1, profile.get("guid"))
    _w_string_or_skip(w, 2, profile.get("token"))
    _w_string_or_skip(w, 3, profile.get("huya_ua"))
    _w_string_or_skip(w, 4, profile.get("cookie"))
    w.int16(5, int(profile.get("i_token_type", 0)))
    _w_string_or_skip(w, 6, profile.get("model"))
    _w_string_or_skip(w, 7, profile.get("qimei"))
    w.struct_end()

    w.struct_begin(1)  # LiveUserbase: t0=eSource t1=eType t2=tUAEx
    w.int16(0, int(profile.get("e_source", 2)))
    w.int16(1, int(profile.get("e_type", 1)))
    w.struct_begin(2)  # LiveAppUAEx: t1=sIMEI t2=sAPN t3=sNetType t4=sDeviceId t5=sMId
    _w_string_or_skip(w, 1, profile.get("imei"))
    _w_string_or_skip(w, 2, profile.get("apn"))
    _w_string_or_skip(w, 3, profile.get("net_type"))
    _w_string_or_skip(w, 4, profile.get("device_id"))
    _w_string_or_skip(w, 5, profile.get("mid"))
    w.struct_end()
    w.struct_end()

    w.int16(2, int(profile.get("b_support_domain", 1)))  # bSupportDomain
    w.struct_end()  # 外层 LiveLaunchReq
    return w.get()


def build_live_launch_wup(profile: dict, request_id: int | None = None) -> bytes:
    """UniPacket 信封 (与密码登录同构): t1 version=3 ... t7 sBuffer=map<string,bytes>."""
    if request_id is None:
        request_id = int.from_bytes(__import__("os").urandom(4), "big") & 0x7FFFFFFF
    req_jce = encode_live_launch_req(profile)

    # 真机实测 (WG 全解密捕获): launch servant 的 map 只有 "tReq" 一个键,
    # 不追加 platform/version/channel 等 (此前追加属过度拟合, 服务器仍拒值)
    entries = [("tReq", ("bytes", req_jce))]
    sb = _Writer()
    sb.map_begin(0, len(entries))
    for k, (kind, val) in entries:
        sb.string(0, k)
        if kind == "bytes":
            sb.bytes(1, val)
        else:
            sb.string(1, val)
    s_buffer = sb.get()

    w = _Writer()
    w.int16(1, 3)          # iVersion = 3
    w.int8(2, 0)           # cPacketType
    w.int8(3, 0)           # iMessageType
    w.int32(4, request_id)
    w.string(5, SERVANT)
    w.string(6, FUNC)
    w.bytes(7, s_buffer)
    w.int32(8, 0)          # iTimeout
    w.map_begin(9, 0)      # context
    w.map_begin(10, 0)     # status
    body = w.get()
    return struct.pack(">I", 4 + len(body)) + body


# ---------------------------------------------------------------------------
# JCE 解码 (读端)
# ---------------------------------------------------------------------------

def _payload(i: TafInputStream, dtype: int):
    """读当前 head 之后的 payload (head 已由调用方 read_head 消费)."""
    buf = i.buf
    if dtype == TafType.ZERO:
        return 0
    if dtype == TafType.INT8:
        return struct.unpack("b", buf.read(1))[0]
    if dtype == TafType.INT16:
        return struct.unpack(">h", buf.read(2))[0]
    if dtype == TafType.INT32:
        return struct.unpack(">i", buf.read(4))[0]
    if dtype == TafType.INT64:
        return struct.unpack(">q", buf.read(8))[0]
    if dtype == TafType.STRING1:
        n = buf.read(1)[0]
        return buf.read(n).decode("utf-8", "replace")
    if dtype == TafType.STRING4:
        n = struct.unpack(">I", buf.read(4))[0]
        return buf.read(n).decode("utf-8", "replace")
    if dtype == TafType.MAP or dtype == TafType.LIST:
        raise ValueError(f"map/list 需单独处理 dtype=0x{dtype:02x}")
    if dtype == TafType.SIMPLE_LIST:
        # 元素类型 head(0,INT8) + 长度 head(0,intN)+value + 数据
        i.read_head()           # 元素类型
        ltag, ldtype = i.read_head()
        n = _payload(i, ldtype)
        return buf.read(n)
    raise ValueError(f"payload dtype=0x{dtype:02x} @pos={buf.tell()}")


def _field_count(i: TafInputStream) -> int:
    """map/list 的条目数 (读长度 head + value)."""
    tag, dtype = i.read_head()
    return _payload(i, dtype)


def _read_struct(i: TafInputStream, depth: int = 0) -> dict:
    """读一个匿名 struct 到 STRUCT_END, 返回 tag->value. map/list 内容做粗显."""
    out: dict[str, object] = {}
    buf = i.buf
    while True:
        pos = buf.tell()
        try:
            tag, dtype = i.read_head()
        except Exception:
            break
        if dtype == TafType.STRUCT_END:
            break
        if dtype in (TafType.MAP, TafType.LIST):
            n = _field_count(i)
            out[tag] = f"<{'map' if dtype == TafType.MAP else 'list'} {n} @{pos:#x}>"
        elif dtype == TafType.STRUCT_BEGIN:
            out[tag] = _read_struct(i, depth + 1)
        else:
            out[tag] = _payload(i, dtype)
    return out


def parse_launch_rsp(resp: bytes) -> dict:
    """解 UniPacket 响应: 顶层字段 + sBuffer map (key->bytes/struct), 捞候选 sGuid.

    返回 {'header': {...}, 'sGuid_candidates': [...], 'structs': [...]}。
    """
    if not HAVE_TAF:
        raise RuntimeError("缺少 core.huya.taf_protocol, 无法解码")
    # HTTP 直连响应可能 gzip 压缩 (1f 8b)
    if resp[:2] == b"\x1f\x8b":
        import gzip as _g
        try:
            resp = _g.decompress(resp)
        except Exception:
            pass
    # 可靠路径: tRsp 结构内 tag0 = sGuid(STRING4 0x06, len 0x20=32hex)
    _sguid_re = re.search(rb"\x06\x20([0-9a-f]{32})", resp)
    # WUP 帧带 4 字节大端长度前缀 (len = 4+body); 剥离后解析
    if len(resp) >= 4:
        declared = struct.unpack(">I", resp[:4])[0]
        if declared == len(resp):
            resp = resp[4:]
    i = TafInputStream(resp)
    header: dict[str, object] = {}
    sguid_candidates: list[str] = ([_sguid_re.group(1).decode()] if _sguid_re else [])
    structs: list[dict] = []
    buf = i.buf
    while True:
        try:
            tag, dtype = i.read_head()
        except Exception:
            break
        if dtype == TafType.STRUCT_END:
            break
        if dtype == TafType.MAP:
            n = _field_count(i)
            header[f"h{tag}"] = f"map<{n}>"
            for _ in range(n):
                ktag, kdtype = i.read_head()
                key = _payload(i, kdtype) if kdtype in (TafType.STRING1, TafType.STRING4) else "?"
                try:
                    vtag, vdtype = i.read_head()
                except Exception:
                    break
                if vdtype == TafType.SIMPLE_LIST:
                    blob = _payload(i, vdtype)
                    if isinstance(blob, bytes):
                        if len(blob) > 2:
                            try:
                                st = _read_struct(TafInputStream(blob))
                                structs.append(st)
                                if 0 in st and isinstance(st[0], str):
                                    sguid_candidates.append(st[0])
                            except Exception:
                                pass
                        else:
                            header[f"h{tag}_map.{key}"] = blob.hex()
                elif vdtype == TafType.STRUCT_BEGIN:
                    st = _read_struct(i)
                    structs.append(st)
                    if 0 in st and isinstance(st[0], str):
                        sguid_candidates.append(st[0])
                else:
                    header[f"h{tag}_map.{key}"] = _payload(i, vdtype)
        elif dtype == TafType.LIST:
            n = _field_count(i)
            header[f"h{tag}"] = f"list<{n}>"
        elif dtype == TafType.STRUCT_BEGIN:
            structs.append(_read_struct(i))
        else:
            payload = _payload(i, dtype)
            header[f"h{tag}"] = payload
            # SIMPLE_LIST/bytes 值 = 嵌套 WUP/JCE 载荷 → 递归捞 sGuid
            if isinstance(payload, bytes) and len(payload) > 2:
                try:
                    nested = parse_launch_rsp(payload)
                except Exception:
                    nested = None
                if nested:
                    structs.extend(nested["structs"])
                    sguid_candidates.extend(nested["sGuid_candidates"])
                    header[f"h{tag}_nested"] = nested["header"]
    return {"header": header, "sGuid_candidates": sguid_candidates, "structs": structs}


# ---------------------------------------------------------------------------
# live 发送
# ---------------------------------------------------------------------------

def mint_sguid(profile: dict | None = None, mid: str | None = None,
               url: str = WUP_URL, timeout: float = 20.0) -> str | None:
    """doLaunch 上线铸币: 服务端按指纹(mid)确定性签发 sGuid (32hex).

    返回 tRsp 里的 sGuid; 无则 None (请自行查 HTTP/解析日志)。
    """
    import gzip as _g
    import urllib.request as _ur
    p = dict(profile or DEFAULT_PROFILE)
    if mid:
        p["mid"] = mid
    body = build_live_launch_wup(p)
    req = _ur.Request(
        url, data=body,
        headers={
            "Content-Type": "application/octet-stream",
            "User-Agent": "Dalvik/2.1.0 (Linux; U; Android 11; M2102J2SC Build/RKQ1.200826.002)",
        })
    with _ur.urlopen(req, timeout=timeout) as r:
        resp = r.read()
    if resp[:2] == b"\x1f\x8b":
        try:
            resp = _g.decompress(resp)
        except Exception:
            pass
    parsed = parse_launch_rsp(resp)
    cands = parsed["sGuid_candidates"]
    return cands[0] if cands else None


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

DEFAULT_PROFILE = {
    "mid": "1e8bdf7d4f7a01d3",
    "imei": "860000000000000",
    "device_id": "3b5c1a9f22e7d40c88a6f5b3012e947d",
    "guid": "",
    "huya_ua": "android&13.4.22&xxx&30",
    "model": "M2102J2SC",
    "qimei": "",
    "luid": 0,
    "apn": "",
    "net_type": "",
    "token": "",
    "cookie": "",
    "platform": "android",
    "version": "13.4.22",
    "channel": "xxx",
    "yyuid": "0",
    "uid": "0",
}


def self_test() -> None:
    """编码→解码回环: 生成请求 JCE + 信封, 再解析 (验证层次正确)."""
    p = dict(DEFAULT_PROFILE)
    p["guid"] = "0a7dfaa882938a6ab502511452142c57"
    req_jce = encode_live_launch_req(p)
    print(f"[self-test] LiveLaunchReq JCE: {len(req_jce)}B")
    # 解析回环: 构造一个假响应 = 信封 + LiveLaunchRsp{guid}
    rsp_jce_w = _Writer()
    rsp_jce_w.string(0, "1a2b3c4d5e6f708192a3b4c5d6e7f809")
    rsp_jce_w.int32(1, 1700000000)
    rsp_jce_w.int32(3, 1)
    rsp_jce_w.string(4, "1.2.3.4")
    rsp_jce = rsp_jce_w.get()
    parsed = parse_launch_rsp(rsp_jce)
    print(f"[self-test] roundtrip parse -> {parsed}")
    print("[self-test] OK")


def dump_req(profile: dict) -> None:
    pkt = build_live_launch_wup(profile)
    print(f"[dump] doLaunch UniPacket: {len(pkt)}B")
    print(pkt.hex())
    print("[dump] field profile:", profile)


def live(profile: dict, url: str) -> None:
    import requests

    pkt = build_live_launch_wup(profile)
    print(f"[live] POST {url}  body={len(pkt)}B")
    r = requests.post(
        url,
        data=pkt,
        headers={
            "Content-Type": "application/multipart-formdata; charset=UTF-8",
            "User-Agent": "Dalvik/2.1.0 (Linux; U; Android 11)",
            "Accept-Encoding": "gzip",
        },
        timeout=20,
    )
    print(f"[live] HTTP {r.status_code} len={len(r.content)}")
    if r.status_code != 200:
        print(r.content[:300])
        return
    parsed = parse_launch_rsp(r.content)
    print("[live] parsed:", parsed)
    guid = None
    # 尝试多种可能路径拿 sGuid
    for k, v in parsed.items():
        if isinstance(v, dict) and 0 in v and isinstance(v[0], str):
            guid = v[0]
    print("[live] sGuid =", guid)


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--self-test", action="store_true")
    ap.add_argument("--dump", action="store_true")
    ap.add_argument("--live", nargs="?", const=WUP_URL, default=None)
    ap.add_argument("--mid", default=None)
    ap.add_argument("--imei", default=None)
    ap.add_argument("--device-id", default=None)
    ap.add_argument("--guid", default=None)
    ap.add_argument("--model", default=None)
    args = ap.parse_args()

    profile = dict(DEFAULT_PROFILE)
    for k, v in (("mid", args.mid), ("imei", args.imei),
                 ("device_id", args.device_id), ("guid", args.guid),
                 ("model", args.model)):
        if v:
            profile[k] = v

    if args.self_test:
        self_test()
    elif args.dump:
        dump_req(profile)
    elif args.live:
        live(profile, args.live)
    else:
        ap.print_help()


if __name__ == "__main__":
    main()