#!/usr/bin/env python3 """Create or update a Hugging Face Docker Space from this repository and optionally verify it live. Token resolution (first match wins), same pattern as ``deploy_hf_bootstrap_catalog.py`` / Codex parser deploy: 1. ``HF_TOKEN`` or ``HUGGING_FACE_HUB_TOKEN`` in the environment 1b. Repo-root ``.env.local`` or ``.env`` (gitignored; see ``.env.example``) if the key is not already set 2. ``[huggingface].token`` in parser ``secrets.toml`` (default: ``~/PycharmProjects/parser/secrets.toml``, override with ``KINK_PARSER_SECRETS_TOML`` or ``--secrets-toml``) 3. ``huggingface_hub.get_token()`` (e.g. after ``hf auth login``) Space repo resolution: 1. ``HF_SPACE_REPO`` (``owner/slug``) 2. ``[huggingface].kink_space_repo`` or ``[huggingface].space_repo`` in the same TOML if set 3. Else ``{[huggingface].username}/kink-discovery`` when username is present (matches this project's Space id) Optional: export HF_SPACE_HARDWARE=cpu-basic # passed to create_repo when the Space is new python scripts/publish_hf_space.py --verify # wait for RUNNING then run verify_hf_stack (incl. /auth/login contract) Does not print your token. Fails fast if the token is missing or invalid. """ from __future__ import annotations import argparse import os import subprocess import sys import time import tomllib from pathlib import Path ROOT = Path(__file__).resolve().parent.parent if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from backend.repo_dotenv import load_repo_dotenv DEFAULT_PARSER_SECRETS = Path( os.environ.get("KINK_PARSER_SECRETS_TOML", Path.home() / "PycharmProjects/parser/secrets.toml"), ) # Match .dockerignore / sensible defaults — keep the bundled seed DB. _IGNORE = [ ".venv", ".venv/**", ".git", ".env", ".env.local", "**/__pycache__/**", ".pytest_cache/**", "data/**", "audit/**", "ui_audit/**", ".cursor/**", "*.egg-info/**", "kink_cli_prototype.egg-info/**", "terminals/**", "node_modules/**", "**/node_modules/**", ".hypothesis/**", ] def _parser_hf_block(secrets_toml: Path) -> dict: if not secrets_toml.is_file(): return {} with secrets_toml.open("rb") as f: data = tomllib.load(f) raw = data.get("huggingface") return dict(raw) if isinstance(raw, dict) else {} def _resolve_token(secrets_toml: Path) -> str: t = (os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN") or "").strip() if t: return t hf = _parser_hf_block(secrets_toml) t = str(hf.get("token") or "").strip() if t: return t try: from huggingface_hub import get_token return (get_token() or "").strip() except Exception: return "" def _resolve_space_repo(secrets_toml: Path) -> str: r = (os.environ.get("HF_SPACE_REPO") or "").strip() if r: return r hf = _parser_hf_block(secrets_toml) r = str(hf.get("kink_space_repo") or hf.get("space_repo") or "").strip() if r: return r user = str(hf.get("username") or "").strip() if user: return f"{user}/kink-discovery" return "" def _default_hf_space_url(repo_id: str) -> str: """Direct app URL pattern used by Hugging Face (same as parser/scripts/hf_deploy.py).""" owner, _, name = repo_id.partition("/") if not owner or not name: raise ValueError(f"Invalid repo_id: {repo_id}") return f"https://{owner}-{name}.hf.space" def _resolve_app_url(api, repo_id: str, token: str) -> str: from huggingface_hub import HfApi assert isinstance(api, HfApi) # `host` is not a valid expand field on current Hub API; subdomain is optional. try: info = api.space_info(repo_id, token=token, expand=["subdomain"]) if info.subdomain: return f"https://{info.subdomain}.hf.space" except Exception: pass return _default_hf_space_url(repo_id) def _wait_running(api, repo_id: str, token: str, timeout_s: float = 900.0) -> None: """Wait until the Space reports ``RUNNING`` (not ``RUNNING_BUILDING``). ``RUNNING_BUILDING`` means the *previous* revision is still serving while a new build runs; probing the app URL then hits stale code and ``publish_hf_space.py --verify`` can falsely fail. """ bad = {"BUILD_ERROR", "CONFIG_ERROR", "RUNTIME_ERROR", "NO_APP_FILE"} deadline = time.time() + timeout_s last_stage = "" while time.time() < deadline: rt = api.get_space_runtime(repo_id, token=token) stage = rt.stage s = getattr(stage, "value", str(stage)) if s != last_stage: print(f"Space stage: {s}", file=sys.stderr) last_stage = s if s == "RUNNING": return if s in bad: raise RuntimeError(f"Space build/runtime failed (stage={s}). Check logs on the Hub.") time.sleep(15.0) raise TimeoutError(f"Space did not become RUNNING within {timeout_s:.0f}s (last stage: {last_stage})") def main() -> int: load_repo_dotenv() ap = argparse.ArgumentParser(description=__doc__) ap.add_argument( "--secrets-toml", type=Path, default=DEFAULT_PARSER_SECRETS, help="Parser secrets.toml with [huggingface] (token, username); default from KINK_PARSER_SECRETS_TOML", ) ap.add_argument("--verify", action="store_true", help="After upload, wait for RUNNING and run scripts/verify_hf_stack.py") ap.add_argument("--no-wait", action="store_true", help="Do not poll for RUNNING (upload only)") args = ap.parse_args() secrets_toml = args.secrets_toml.expanduser().resolve() token = _resolve_token(secrets_toml) repo_id = _resolve_space_repo(secrets_toml) if not token or not repo_id: print( "Missing Hugging Face token or Space repo id.\n" " Token: repo-root `.env.local` / `.env`, export HF_TOKEN=…, parser secrets.toml [huggingface].token, or `hf auth login`.\n" " Repo: export HF_SPACE_REPO=owner/slug, or set [huggingface].kink_space_repo, or [huggingface].username " f"(defaults to /kink-discovery).\n" f" Parser secrets path tried: {secrets_toml}", file=sys.stderr, ) return 1 try: from huggingface_hub import HfApi except ImportError as e: print("Install huggingface_hub: pip install huggingface_hub", file=sys.stderr) raise SystemExit(1) from e api = HfApi(token=token) from huggingface_hub import SpaceHardware hw_raw = os.environ.get("HF_SPACE_HARDWARE", "cpu-basic").strip() or "cpu-basic" try: hardware = SpaceHardware(hw_raw) except ValueError: hardware = SpaceHardware.CPU_BASIC api.create_repo( repo_id, repo_type="space", space_sdk="docker", exist_ok=True, space_hardware=hardware, ) print(f"Uploading {ROOT} -> spaces/{repo_id} ...", file=sys.stderr) api.upload_folder( folder_path=str(ROOT), repo_id=repo_id, repo_type="space", ignore_patterns=_IGNORE, commit_message="Sync from kink_cli (Docker Space)", ) print(f"Pushed. Hub page: https://huggingface.co/spaces/{repo_id}", file=sys.stderr) if args.no_wait: print( "After the build finishes, verify with:\n" f" HF_VERIFY_BASE_URL= python scripts/verify_hf_stack.py\n" "Resolve from the Space page (direct app URL, often *.hf.space).", file=sys.stderr, ) return 0 print("Waiting for Space to leave BUILDING…", file=sys.stderr) _wait_running(api, repo_id, token) base = _resolve_app_url(api, repo_id, token) print(f"App URL: {base}", file=sys.stderr) if args.verify: env = { **os.environ, "HF_VERIFY_BASE_URL": base, # Default 1: Docker Spaces often ship the bundled seed only (~12 kinks). Set # HF_VERIFY_MIN_KINKS=1000+ when your Space bootstraps a full Hub catalog. "HF_VERIFY_MIN_KINKS": os.environ.get("HF_VERIFY_MIN_KINKS", "1"), # Large kink JSON + cold HF edges can exceed 120s without this. "HF_VERIFY_HTTP_TIMEOUT_S": os.environ.get("HF_VERIFY_HTTP_TIMEOUT_S", "300"), } r = subprocess.run( [sys.executable, str(ROOT / "scripts" / "verify_hf_stack.py"), "--base-url", base], cwd=str(ROOT), env=env, ) return int(r.returncode) print(f"Smoke-test when ready:\n HF_VERIFY_BASE_URL={base} python scripts/verify_hf_stack.py", file=sys.stderr) return 0 if __name__ == "__main__": raise SystemExit(main())