chuanqi/cq2_sdk/scripts/diag_gateway.py

"""Diagnose the real game-gateway host/port for zone 14426767 (today 16:00).

Findings so far:
  - .env CQ2_GATE_HOST=jycq-206923.dhsf.xqhuyu.com is DEAD (DNS fails) — it was a guess
  - probe_gateway.py (prior RE work): real gateway = gfgame.dhsf.xqhuyu.com,
    per-zone ports live in cached ResServerlist.txt_206923 (compact keys:
    d=serverId p=port i=host s=state t=group)
  - C1's start_info has a signed modlist URL — possibly the FRESH zone list

This script:
  1. Dump 玖扬 group (t=1000514) entries from cached srvlist: d/p/i/s
  2. Fetch the signed modlist URL from run_6_env.json, detect format,
     look for zone 14426767
  3. DNS + TCP probe gfgame.dhsf.xqhuyu.com and known IPs on the port matrix
"""
import asyncio
import json
import socket
import sys
from pathlib import Path

_HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(_HERE.parent / "src"))

from cq2_sdk.env_loader import load_env
load_env(_HERE.parent / ".env")

import httpx

CACHED_SRVLIST = Path(r"D:\ProgramFiles\996box\GameBoxDownload\1\cq\cache_206923\ResServerlist.txt_206923")
ENV_DUMP = _HERE.parent / "env_dumps" / "run_6_env.json"
TARGET_ZONE = 14426767  # today 16:00, B3's pick

print("=== [1] cached srvlist: 玖扬 group (t=1000514) ===")
try:
    data = json.loads(CACHED_SRVLIST.read_text(encoding="utf-8"))
    servers = data["serverData"]
    jy = [s for s in servers if str(s.get("t")) == "1000514"]
    print(f"total serverData: {len(servers)}, 玖扬 group: {len(jy)}")
    for s in jy[-12:]:  # 最新 12 个
        print(f"  d={s.get('d'):<9} p={s.get('p'):<6} i={s.get('i') or '(default gfgame)':<28} s={s.get('s')}")
    hit = [s for s in servers if str(s.get("d")) == str(TARGET_ZONE)]
    print(f"  target {TARGET_ZONE} in cached: {'YES p=' + str(hit[0]['p']) if hit else 'NO (file is stale, need modlist)'}")
    ports_jy = sorted({int(s['p']) for s in jy if s.get('p')})
    print(f"  玖扬 port set: {ports_jy}")
except Exception as e:
    print(f"  FAIL: {type(e).__name__}: {e}")

print("\n=== [2] modlist from run_6_env.json ===")
try:
    env = json.loads(ENV_DUMP.read_text(encoding="utf-8"))
    modlist_url = env.get("modlist", "")
    print(f"  url: {modlist_url[:90]}...")
    r = httpx.get(modlist_url, headers={"User-Agent": "996Box/5.1.1.9007"},
                  timeout=15, verify=False)
    print(f"  http {r.status_code}, {len(r.content)} bytes")
    try:
        mdata = r.json()
        if isinstance(mdata, dict) and "serverData" in mdata:
            mservers = mdata["serverData"]
            mhit = [s for s in mservers if str(s.get("d")) == str(TARGET_ZONE)]
            print(f"  serverData entries: {len(mservers)}")
            if mhit:
                s = mhit[0]
                print(f"  *** TARGET {TARGET_ZONE}: p={s.get('p')} i={s.get('i') or '(default gfgame)'} s={s.get('s')} ***")
            else:
                jym = [s for s in mservers if str(s.get("t")) == "1000514"]
                print(f"  玖扬 in modlist: {len(jym)}")
                for s in jym[-6:]:
                    print(f"    d={s.get('d'):<9} p={s.get('p'):<6} i={s.get('i') or '(default)'} s={s.get('s')}")
        else:
            print(f"  keys: {list(mdata.keys())[:10] if isinstance(mdata, dict) else type(mdata).__name__}")
    except Exception:
        print(f"  not JSON; head: {r.text[:200]!r}")
except Exception as e:
    print(f"  FAIL: {type(e).__name__}: {e}")

print("\n=== [3] gateway probes ===")
async def tcp_ok(host: str, port: int) -> bool:
    try:
        r, w = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=4)
        w.close()
        return True
    except Exception:
        return False

async def probes():
    gh = "gfgame.dhsf.xqhuyu.com"
    try:
        infos = socket.getaddrinfo(gh, None)
        addrs = sorted({i[4][0] for i in infos})
        print(f"  DNS {gh} -> {addrs}")
    except Exception as e:
        addrs = []
        print(f"  DNS {gh} FAIL: {e}")
    targets = [gh] + ["47.118.169.147", "47.118.169.153", "47.118.172.233"]
    ports = [8001, 8021, 8041, 8061, 8081, 12000]
    for t in targets:
        results = await asyncio.gather(*(tcp_ok(t, p) for p in ports))
        marks = " ".join(f"{p}:{'OPEN' if o else '-'}" for p, o in zip(ports, results))
        print(f"  {t:<24} {marks}")

asyncio.run(probes())