chuanqi/cq2_sdk/scripts/business/b3_select_latest_zone.py

"""B3: 选今天能玩的最新区 — 纯 DB 查询, 不调 HTTP。

数据流：
    MySQL servers (B1 存的) + zones (B2 存的)
        ↓
    ZoneStore.get_latest_openable(server_id)
        ↓
    SQL: WHERE status='open' AND open_time<=NOW() ORDER BY open_time DESC LIMIT 1
        ↓
    返单个 Zone (含 host/port 待 C1 填, 这里没有就 None)
        ↓
    DebugLogger 留痕

跑法：
    uv run python scripts/business/b3_select_latest_zone.py
    uv run python scripts/business/b3_select_latest_zone.py --server "玖扬复古176"
    uv run python scripts/business/b3_select_latest_zone.py --check

前置依赖：
    1. B1 跑过 (server 在 MySQL)
    2. B2 跑过 (zones 在 MySQL)
"""
from __future__ import annotations
import argparse
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.db_models import Zone  # noqa: E402
from cq2_sdk.debug_log import DebugLogger  # noqa: E402
from cq2_sdk.store import ServerStore, ZoneStore, connection  # noqa: E402

DEFAULT_SERVER_NAME = "玖扬复古176"


def select_latest_zone(server_name: str, log: DebugLogger) -> Zone | None:
    """查 MySQL 找今天能玩的最新一个区。"""
    log.step("b3_select_latest_zone_start", server_name=server_name)

    # 1. server
    with connection() as conn:
        server = ServerStore(conn).get_by_name(server_name)
        if server is None:
            raise RuntimeError(
                f"server not in MySQL: {server_name!r}。先跑 B1 把它存进去。"
            )
        log.info("server_loaded",
                 server_db_id=server.id, server_name=server.server_name,
                 server_996box_id=server.server_id)

        # 2. latest openable zone
        zone = ZoneStore(conn).get_latest_openable(server.id)

    if zone is None:
        log.warn("no_openable_zone",
                 note="可能 B2 还没跑, 或当前时间早于今天第一个 open_time, "
                      "或所有 zone 都被 mark_missing_closed 了")
        return None

    log.info("latest_zone_selected",
             zone_db_id=zone.id, zone_id=zone.zone_id, zone_name=zone.zone_name,
             open_time=zone.open_time.isoformat() if zone.open_time else None,
             status=zone.status,
             host=zone.host, port=zone.port,
             note="B3 只填 zone 元数据, host/port 待 C1 (launch_game) 拿到")
    return zone


def format_zone(zone: Zone | None) -> dict[str, Any]:
    """输出友好格式。"""
    if zone is None:
        return {"found": False}
    return {
        "found": True,
        "db_id": zone.id,
        "zone_id": zone.zone_id,
        "zone_name": zone.zone_name,
        "open_time": zone.open_time.isoformat() if zone.open_time else None,
        "status": zone.status,
        "host": zone.host,
        "port": zone.port,
    }


def check_config(server_name: str | None) -> bool:
    print("[check] 配置检查:")
    print(f"  server_name = {server_name!r}  [OK]"
          if server_name else "  server_name = (未设置)  [FAIL]")
    print("  (实际跑还需要 B1 + B2 已成功跑过, MySQL 里有 server 和 zones)")
    return bool(server_name)


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(
        description="B3: 从 MySQL 选今天能玩的最新区 (纯 DB, 不调 HTTP)",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    p.add_argument(
        "--server",
        default=os.environ.get("CQ2_DEFAULT_SERVER", DEFAULT_SERVER_NAME),
        help=f"区服名 (默认 {DEFAULT_SERVER_NAME})",
    )
    p.add_argument(
        "--check",
        action="store_true",
        help="只检查 .env / CLI 配置, 不查 DB",
    )
    return p.parse_args()


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

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

    if not args.server:
        check_config(args.server)
        return 1

    with DebugLogger(extra={"task": "b3", "server_name": args.server}) as log:
        log.info("b3_start", server_name=args.server)
        try:
            zone = select_latest_zone(args.server, log)
        except Exception as e:
            log.error(e, step="b3_unexpected")
            print(f"\n[FAIL] {type(e).__name__}: {e}")
            print(f"  log file: {log.path}")
            return 1

    result = format_zone(zone)
    if not result["found"]:
        print("\n[FAIL] 没找到今天能玩的区")
        print(f"  log file: {log.path}")
        return 2

    print("\n[OK] B3 选区成功:")
    for k, v in result.items():
        print(f"  {k:<12} = {v}")
    print(f"  log file    = {log.path}")
    return 0


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