chuanqi/cq2_sdk/scripts/business/b1_find_server.py

"""B1: 按名称查区服 — 调 list-box 拉全部 serverlist, 匹配后存进 MySQL servers 表。

数据流：
    .env / --name "玖扬复古176"
        ↓
    fetch_srvlist(206923)  →  GET list-box.dhsf.xqhuyu.com/box_serverlist/206923.txt
        ↓
    遍历 serverlist, 找 list_name 包含 --name 的 group
        ↓
    取 group 的 main server (serverid + engin_id)
        ↓
    ServerStore.upsert(server_name, server_id, game_id)
        ↓
    DebugLogger 留痕

跑法：
    uv run python scripts/business/b1_find_server.py
    uv run python scripts/business/b1_find_server.py --name "玖扬复古176"
    uv run python scripts/business/b1_find_server.py --game-id 999999
    uv run python scripts/business/b1_find_server.py --check
"""
from __future__ import annotations
import argparse
import asyncio
import os
import sys
from pathlib import Path
from typing import Any

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

from cq2_sdk.env_loader import load_env  # noqa: E402
load_env(_PROJECT / ".env")

from cq2_sdk.debug_log import DebugLogger  # noqa: E402
from cq2_sdk.launcher.navigate import ServerNotFound, fetch_srvlist  # noqa: E402
from cq2_sdk.store import ServerStore, connection  # noqa: E402

DEFAULT_GAME_ID = 206923
DEFAULT_SERVER_NAME = "玖扬复古176"  # 当前的"主"区服，未来可改


def pick_server_from_srvlist(srvlist: dict, name: str) -> dict[str, Any]:
    """Find the first server group whose list_name contains ``name``.

    Returns a dict with: server_id, group_id, group_name, engin_id.
    Raises ServerNotFound if no match.
    """
    for gid_str, group in srvlist.items():
        group_name = group.get("list_name", "") or ""
        if name in group_name or group_name == name:
            tree = group.get("tree") or []
            if not tree:
                continue
            main_id = group.get("main_serverid")
            target = next(
                (s for s in tree if s.get("serverid") == main_id),
                tree[0],
            )
            return {
                "server_id": int(target.get("serverid", 0)),
                "group_id": int(gid_str),
                "group_name": group_name,
                "engin_id": target.get("engin_id"),
            }
    raise ServerNotFound(f"server not found: {name!r}")


async def find_and_save(
    name: str,
    game_id: int,
    log: DebugLogger,
) -> dict[str, Any]:
    """Run B1: fetch srvlist, match by name, persist to MySQL.

    Returns the picked server info dict.
    """
    log.step("b1_find_server_start", target=name, game_id=game_id)

    # 1. fetch (no auth needed for browsing the public server list)
    log.info("fetching_srvlist", game_id=game_id)
    srvlist = await fetch_srvlist(str(game_id))
    log.info("srvlist_fetched", group_count=len(srvlist))

    # 2. match
    picked = pick_server_from_srvlist(srvlist, name)
    log.info("server_matched", **picked)

    # 3. persist
    log.step("b1_persist_to_mysql", server_name=name, server_id=picked["server_id"])
    with connection() as conn:
        servers = ServerStore(conn)
        server = servers.upsert(
            server_name=name,
            server_id=picked["server_id"],
            game_id=game_id,
        )
    log.info("server_saved_to_mysql",
             db_id=server.id, server_name=server.server_name,
             server_id=server.server_id)

    return {
        **picked,
        "db_id": server.id,
    }


def check_config(name: str | None, game_id: int) -> bool:
    print("[check] 配置检查:")
    print(f"  target server name = {name!r}  [OK]" if name else
          "  target server name = (未设置)  [FAIL]")
    print(f"  game_id = {game_id}  [OK]")
    return bool(name)


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(
        description="B1: 按名称查 996box 区服, 存进 MySQL servers 表",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "示例：\n"
            "  uv run python scripts/business/b1_find_server.py --check\n"
            "  uv run python scripts/business/b1_find_server.py\n"
            "  uv run python scripts/business/b1_find_server.py --name 复古180 --game-id 999\n"
        ),
    )
    p.add_argument(
        "--name",
        default=os.environ.get("CQ2_DEFAULT_SERVER", DEFAULT_SERVER_NAME),
        help=f"区服名 (默认 {DEFAULT_SERVER_NAME}，可设 .env CQ2_DEFAULT_SERVER)",
    )
    p.add_argument(
        "--game-id",
        type=int,
        default=int(os.environ.get("CQ2_GAME_ID", DEFAULT_GAME_ID)),
        help=f"game id (默认 {DEFAULT_GAME_ID})",
    )
    p.add_argument(
        "--check",
        action="store_true",
        help="只检查 .env / CLI 配置和 import 依赖, 不实际调 list-box",
    )
    return p.parse_args()


def main() -> int:
    args = parse_args()

    if args.check:
        ok = check_config(args.name, args.game_id)
        print(f"\n  result: {'READY' if ok else 'NOT READY'}")
        return 0 if ok else 1

    if not args.name:
        print("[FAIL] 缺少 server name")
        return 1

    with DebugLogger(
        extra={"task": "b1", "server_name": args.name, "game_id": args.game_id}
    ) as log:
        log.info("b1_start", name=args.name, game_id=args.game_id)
        try:
            picked = asyncio.run(find_and_save(args.name, args.game_id, log))
        except ServerNotFound as e:
            log.error(e, step="b1_match_failed")
            print(f"\n[FAIL] {e}")
            print(f"  提示: --name 不在 list-box 返回的区服列表里")
            print(f"  log file: {log.path}")
            return 2
        except Exception as e:
            log.error(e, step="b1_unexpected")
            print(f"\n[FAIL] {type(e).__name__}: {e}")
            print(f"  log file: {log.path}")
            return 1

    print("\n[OK] B1 查区服 + 落库成功:")
    print(f"  server_name = {picked['group_name']!r}  (996box 列表里的全名)")
    print(f"  server_id   = {picked['server_id']}  (996box 数字 ID)")
    print(f"  group_id    = {picked['group_id']}")
    print(f"  engin_id    = {picked.get('engin_id')}")
    print(f"  db_id       = {picked['db_id']}  (MySQL servers.id)")
    print(f"  log file    = {log.path}")
    return 0


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