climperor/web/deploy_relations.py

"""Deploy the Climperor web site static bundle to Cloudflare Pages.

Bundles export + integrity check + wrangler direct upload + custom domain
binding into one local command. Designed to run with credentials injected
by keyzoo asset_exec, but also works with manually exported env vars.

Usage:
    python deploy_relations.py [--no-export] [--project-name NAME] \\
        [--domain DOMAIN] [--dist PATH]

Env vars (credentials, never printed):
    CLOUDFLARE_EMAIL / CLOUDFLARE_API_KEY    -- preferred
    KEYZOO_ASSET_META_USERNAME              -- fallback email (keyzoo inject)
    KEYZOO_ASSET_SECRET_GLOBAL_API_KEY      -- fallback key  (keyzoo inject)

Flow:
    1. resolve credentials + account_id (GET /accounts)
    2. ensure Pages project exists (create if missing, production_branch=main)
    3. run export_relations_site.py unless --no-export
    4. asset-integrity check: abort if key dirs are empty
    5. wrangler pages deploy dist/relations (CI=true, non-interactive)
    6. bind custom domain if not already
    7. print deployment + custom domain URLs

Security: the API key is only ever placed into HTTP headers and the
wrangler subprocess env. It is never printed, logged, or written to disk.
"""

from __future__ import annotations

import argparse
import json
import os
import shutil
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parent

DEFAULT_PROJECT = "climperor-relations"
DEFAULT_DOMAIN = "dota2.refining.dev"
DEFAULT_DIST = ROOT / "dist" / "relations"
DEFAULT_OSS_BASE = "https://climperor.oss-cn-shanghai.aliyuncs.com"

# Directories export_relations_site.py writes under dist/relations. Keys are
# the sub-directory names; values flag whether an empty dir is a hard failure
# (True) or just a warning (False). Portrait/ui-icon must be non-empty or the
# site is visibly broken; item/ability degrade with onerror but should exist.
INTEGRITY_DIRS: dict[str, bool] = {
    "portrait": True,
    "ui-icon": True,
    "item": True,
    "ability": False,
    "attr": True,
    "item-cat": True,
}


def resolve_credentials() -> tuple[str, str]:
    """Return (email, api_key) from env, preferring CLOUDFLARE_* vars."""
    email = os.environ.get("CLOUDFLARE_EMAIL") or os.environ.get(
        "KEYZOO_ASSET_META_USERNAME"
    )
    api_key = os.environ.get("CLOUDFLARE_API_KEY") or os.environ.get(
        "KEYZOO_ASSET_SECRET_GLOBAL_API_KEY"
    )
    if not email or not api_key:
        raise SystemExit(
            "missing credentials: set CLOUDFLARE_EMAIL + CLOUDFLARE_API_KEY "
            "(or run via keyzoo asset_exec on the refining/cloudflare asset)"
        )
    return email, api_key


def cf_api(
    method: str,
    path: str,
    *,
    email: str,
    api_key: str,
    body: dict | None = None,
) -> dict:
    """Call the Cloudflare REST API with Global API Key auth.

    Uses X-Auth-Email + X-Auth-Key headers. Returns the parsed JSON envelope.
    Never raises on HTTP error; returns the error envelope so callers can
    inspect `success` / `errors` without a traceback.
    """
    url = "https://api.cloudflare.com/client/v4" + path
    headers = {
        "X-Auth-Email": email,
        "X-Auth-Key": api_key,
        "Content-Type": "application/json",
    }
    data = json.dumps(body).encode("utf-8") if body is not None else None
    req = urllib.request.Request(url, data=data, method=method, headers=headers)
    try:
        with urllib.request.urlopen(req) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        # HTTPError still carries the CF JSON envelope; surface it verbatim
        # (it contains no secrets, only error codes/messages).
        try:
            return json.loads(exc.read().decode("utf-8"))
        except Exception:
            return {"success": False, "errors": [{"code": exc.code, "message": str(exc)}]}
    except urllib.error.URLError as exc:
        return {"success": False, "errors": [{"code": -1, "message": str(exc)}]}


def get_account_id(email: str, api_key: str) -> str:
    """Resolve the Cloudflare account id via GET /accounts."""
    d = cf_api("GET", "/accounts", email=email, api_key=api_key)
    if not d.get("success"):
        raise SystemExit(f"GET /accounts failed: {d.get('errors')}")
    accounts = d.get("result") or []
    if not accounts:
        raise SystemExit("no Cloudflare accounts found for these credentials")
    if len(accounts) > 1:
        names = [a.get("name", a.get("id")) for a in accounts]
        print(f"multiple accounts found, using the first: {names}", file=sys.stderr)
    return str(accounts[0]["id"])


def ensure_project(email: str, api_key: str, account_id: str, project: str) -> None:
    """Create the Pages project if it does not already exist."""
    d = cf_api(
        "GET",
        f"/accounts/{account_id}/pages/projects/{project}",
        email=email,
        api_key=api_key,
    )
    if d.get("success"):
        print(f"project exists: {project}")
        return
    # 8000xxx = project not found; anything else is a real error.
    print(f"project not found, creating: {project}")
    d = cf_api(
        "POST",
        f"/accounts/{account_id}/pages/projects",
        email=email,
        api_key=api_key,
        body={"name": project, "production_branch": "main"},
    )
    if not d.get("success"):
        raise SystemExit(f"create project failed: {d.get('errors')}")


def run_export(
    ability_video_base: str | None = None,
    static_asset_base: str | None = None,
) -> None:
    """Re-run export_relations_site.py to refresh dist/relations."""
    print("exporting static site ...")
    cmd = [sys.executable, str(ROOT / "export_relations_site.py")]
    oss = ability_video_base
    if oss is None:
        oss = os.environ.get("ABILITY_VIDEO_BASE", DEFAULT_OSS_BASE)
    static = static_asset_base
    if static is None:
        static = os.environ.get("STATIC_ASSET_BASE", oss or DEFAULT_OSS_BASE)
    if oss:
        cmd.extend(["--ability-video-base", oss])
        print(f"ABILITY_VIDEO_BASE={oss}")
    if static:
        cmd.extend(["--static-asset-base", static])
        print(f"STATIC_ASSET_BASE={static}")
    subprocess.run(cmd, cwd=str(ROOT), check=True)


def read_config_js_var(dist: Path, name: str) -> str:
    """Parse a string literal from generated config.js."""
    cfg = dist / "config.js"
    if not cfg.is_file():
        return ""
    prefix = f"var {name} = "
    for line in cfg.read_text(encoding="utf-8").splitlines():
        if line.startswith(prefix):
            return json.loads(line[len(prefix) :].rstrip(";"))
    return ""


def check_integrity(dist: Path) -> None:
    """Abort if a hard-required asset directory is empty; warn otherwise."""
    static_base = read_config_js_var(dist, "STATIC_ASSET_BASE")
    if static_base:
        print(
            f"asset integrity check: skipped local dirs "
            f"(STATIC_ASSET_BASE={static_base})"
        )
        return
    problems: list[str] = []
    for sub, required in INTEGRITY_DIRS.items():
        d = dist / sub
        n = sum(1 for _ in d.glob("*") if _.is_file()) if d.is_dir() else 0
        if n == 0:
            tag = "MISSING (abort)" if required else "empty (warn)"
            problems.append(f"  {sub}/: {tag}")
            hint = {
                "portrait": "run: python fetch_cdn_templates.py && python fetch_hero_portraits.py",
                "ability": "run: python fetch_hero_abilities.py --icons-only",
                "item": "run: python fetch_hero_items.py",
                "attr": "assets/attr_icons is committed; check git checkout",
                "item-cat": "assets/item_cat_icons is committed; check git checkout",
                "ui-icon": "assets/ui_icons is committed; check git checkout",
            }.get(sub, "see AGENTS.md fetch commands")
            problems.append(f"      -> {hint}")
    if not problems:
        return
    hard = [s for s, r in INTEGRITY_DIRS.items() if r and not any((dist / s).glob("*"))]
    print("asset integrity check:")
    print("\n".join(problems))
    if hard:
        raise SystemExit(
            f"aborting deploy: required asset dirs empty: {hard}. "
            "Run the fetch scripts first, or set STATIC_ASSET_BASE for OSS hosting."
        )


def resolve_npx() -> str:
    """Locate the npx executable (Windows ships npx.cmd)."""
    for name in ("npx", "npx.cmd"):
        path = shutil.which(name)
        if path:
            return path
    raise SystemExit("npx not found on PATH; install Node.js first")


def wrangler_deploy(
    dist: Path,
    project: str,
    *,
    email: str,
    api_key: str,
    account_id: str,
) -> None:
    """Upload dist via `wrangler pages deploy` (direct upload, non-interactive).

    Wrangler looks for ``functions/`` relative to the process cwd (not inside
    an absolute asset-directory argument). Run from ``dist`` and deploy ``.``
    so ``dist/functions/`` (e.g. /api/live-status) is compiled and uploaded.
    """
    functions_dir = dist / "functions"
    if not functions_dir.is_dir():
        print(
            f"warn: {functions_dir} missing — Pages Functions will not deploy "
            "(re-run export_relations_site.py)"
        )
    env = os.environ.copy()
    # wrangler reads Global API Key auth from these env vars. They are passed
    # to the subprocess only; wrangler does not echo them.
    env["CLOUDFLARE_EMAIL"] = email
    env["CLOUDFLARE_API_KEY"] = api_key
    env["CLOUDFLARE_ACCOUNT_ID"] = account_id
    env["CI"] = "true"
    cmd = [
        resolve_npx(),
        "--yes",
        "wrangler@3",
        "pages",
        "deploy",
        ".",
        "--project-name",
        project,
        "--commit-dirty",
        "--branch",
        "main",
    ]
    print(f"uploading {dist} via wrangler (cwd=dist, functions={functions_dir.is_dir()}) ...")
    subprocess.run(cmd, cwd=str(dist), env=env, check=True)


def smoke_test_production(
    base_url: str,
    *,
    expected_run_id: str = "",
    attempts: int = 6,
    delay_s: float = 5.0,
) -> None:
    """Verify the deployed payload and Pages Function after propagation."""
    base = base_url.rstrip("/")
    marker = expected_run_id or str(int(time.time()))
    error = "unknown failure"
    for attempt in range(1, attempts + 1):
        try:
            data_url = f"{base}/data.json?refresh={urllib.parse.quote(marker)}"
            req = urllib.request.Request(data_url, headers={"User-Agent": "climperor-smoke"})
            with urllib.request.urlopen(req, timeout=30) as resp:
                payload = json.loads(resp.read().decode("utf-8"))
            if not isinstance(payload, dict) or not payload.get("heroes"):
                raise RuntimeError("data.json has no heroes")
            actual_run_id = str((payload.get("meta") or {}).get("refresh_run_id") or "")
            if expected_run_id and actual_run_id != expected_run_id:
                raise RuntimeError(
                    f"stale deployment run_id={actual_run_id!r}, expected={expected_run_id!r}"
                )

            live_url = f"{base}/api/live-status?refresh={urllib.parse.quote(marker)}"
            live_req = urllib.request.Request(
                live_url, headers={"User-Agent": "climperor-smoke"}
            )
            with urllib.request.urlopen(live_req, timeout=30) as resp:
                content_type = resp.headers.get_content_type()
                live_cache = (resp.headers.get("X-Live-Cache") or "").lower()
                live_payload = json.loads(resp.read().decode("utf-8"))
            if content_type != "application/json" or not isinstance(live_payload, dict):
                raise RuntimeError(
                    f"/api/live-status is not JSON (content-type={content_type})"
                )
            if live_cache in ("error", "stale-override"):
                print(
                    f"warn: live-status Function is routed but probe state is {live_cache}"
                )
            print(
                f"production smoke passed: {base} "
                f"(heroes={len(payload['heroes'])}, run_id={actual_run_id or '-'}, "
                f"live_cache={live_cache or '?'})"
            )
            return
        except (
            OSError,
            ValueError,
            RuntimeError,
            json.JSONDecodeError,
            urllib.error.URLError,
        ) as exc:
            error = str(exc)
            print(f"smoke attempt {attempt}/{attempts} failed: {error}")
            if attempt < attempts:
                time.sleep(delay_s)
    raise SystemExit(f"deployment smoke failed after {attempts} attempts: {error}")


def ensure_cname(email: str, api_key: str, domain: str, project: str) -> None:
    """Ensure the zone has a proxied CNAME: <domain> -> <project>.pages.dev.

    Pages binds the custom domain but does not always auto-create the zone
    CNAME; without it SSL validation stalls at 'CNAME record not set'.
    Idempotent: skips if a CNAME already exists for the domain.
    """
    zone = domain.split(".", 1)[1] if "." in domain else domain
    target = f"{project}.pages.dev"
    zd = cf_api("GET", f"/zones?name={zone}", email=email, api_key=api_key)
    zr = zd.get("result") or []
    if not zr:
        print(f"warn: zone {zone} not found on this account; add CNAME manually")
        return
    zid = zr[0]["id"]
    recs = cf_api(
        "GET", f"/zones/{zid}/dns_records?name={domain}", email=email, api_key=api_key
    )
    existing = [r for r in (recs.get("result") or []) if r.get("type") == "CNAME"]
    if existing:
        r = existing[0]
        print(f"CNAME exists: {r.get('name')} -> {r.get('content')} (proxied={r.get('proxied')})")
        return
    print(f"adding CNAME: {domain} -> {target} (proxied=true)")
    d = cf_api(
        "POST",
        f"/zones/{zid}/dns_records",
        email=email,
        api_key=api_key,
        body={"type": "CNAME", "name": domain, "content": target, "proxied": True},
    )
    if d.get("success"):
        print("CNAME added; SSL will activate within a few minutes")
    else:
        print(f"warn: add CNAME failed: {d.get('errors')} (add manually in dashboard)")


def bind_domain(
    email: str,
    api_key: str,
    account_id: str,
    project: str,
    domain: str,
) -> None:
    """Attach the custom domain to the Pages project and ensure its CNAME."""
    d = cf_api(
        "GET",
        f"/accounts/{account_id}/pages/projects/{project}/domains",
        email=email,
        api_key=api_key,
    )
    if not d.get("success"):
        raise SystemExit(f"list domains failed: {d.get('errors')}")
    existing = [x.get("name") for x in (d.get("result") or [])]
    if domain in existing:
        print(f"custom domain already bound: {domain}")
    else:
        print(f"binding custom domain: {domain}")
        d = cf_api(
            "POST",
            f"/accounts/{account_id}/pages/projects/{project}/domains",
            email=email,
            api_key=api_key,
            body={"name": domain},
        )
        if not d.get("success"):
            raise SystemExit(
                f"bind domain failed: {d.get('errors')}. "
                f"Ensure {domain} zone is hosted on this Cloudflare account."
            )
    ensure_cname(email, api_key, domain, project)


def main() -> None:
    ap = argparse.ArgumentParser(description="Deploy Climperor web site to Cloudflare Pages")
    ap.add_argument("--no-export", action="store_true", help="skip re-export, deploy existing dist")
    ap.add_argument("--project-name", default=DEFAULT_PROJECT)
    ap.add_argument("--domain", default=DEFAULT_DOMAIN)
    ap.add_argument("--dist", default=str(DEFAULT_DIST))
    ap.add_argument(
        "--ability-video-base",
        default=None,
        help="passed to export as --ability-video-base (default: OSS climperor endpoint "
        "or ABILITY_VIDEO_BASE env)",
    )
    ap.add_argument(
        "--static-asset-base",
        default=None,
        help="passed to export as --static-asset-base (default: same as ability-video-base "
        "or STATIC_ASSET_BASE env)",
    )
    ap.add_argument(
        "--skip-smoke",
        action="store_true",
        help="skip post-deploy data.json and /api/live-status verification",
    )
    args = ap.parse_args()

    dist = Path(args.dist).resolve()
    email, api_key = resolve_credentials()

    account_id = get_account_id(email, api_key)
    print(f"account_id: {account_id}")

    ensure_project(email, api_key, account_id, args.project_name)

    if not args.no_export:
        run_export(args.ability_video_base, args.static_asset_base)
    else:
        print("--no-export: using existing dist")

    if not dist.is_dir():
        raise SystemExit(f"dist not found: {dist}. Run without --no-export first.")

    check_integrity(dist)
    wrangler_deploy(dist, args.project_name, email=email, api_key=api_key, account_id=account_id)
    bind_domain(email, api_key, account_id, args.project_name, args.domain)
    if not args.skip_smoke:
        smoke_test_production(
            f"https://{args.project_name}.pages.dev",
            expected_run_id=os.environ.get("REFRESH_RUN_ID") or "",
        )

    print()
    print(f"deployment url : https://{args.project_name}.pages.dev")
    print(f"custom domain  : https://{args.domain}")
    print("note: a freshly bound custom domain takes ~1-2 min to issue SSL.")


if __name__ == "__main__":
    main()