climperor/deploy_relations.py

"""Deploy the relations preview static site 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 urllib.error
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"

# 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) -> None:
    """Re-run export_relations_site.py to refresh dist/relations."""
    print("exporting static site ...")
    cmd = [sys.executable, str(ROOT / "export_relations_site.py")]
    # Production Pages should load ability demos from OSS, not bundle GB of video.
    base = ability_video_base
    if base is None:
        base = os.environ.get(
            "ABILITY_VIDEO_BASE",
            "https://climperor.oss-cn-shanghai.aliyuncs.com",
        )
    if base:
        cmd.extend(["--ability-video-base", base])
        print(f"ABILITY_VIDEO_BASE={base}")
    subprocess.run(cmd, cwd=str(ROOT), check=True)


def check_integrity(dist: Path) -> None:
    """Abort if a hard-required asset directory is empty; warn otherwise."""
    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 so the exported site is complete."
        )


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)."""
    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",
        str(dist),
        "--project-name",
        project,
        "--commit-dirty",
        "--branch",
        "main",
    ]
    print(f"uploading {dist} via wrangler ...")
    subprocess.run(cmd, cwd=str(ROOT), env=env, check=True)


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 relations 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)",
    )
    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)
    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)

    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()