scripts/deploy_hf_space.py

#!/usr/bin/env python3
"""Validate and publish one atomic Gradio Space revision (app plus public results)."""

from __future__ import annotations
import argparse
import json
import os
from pathlib import Path
import shutil
import sys
import tempfile
import time
from types import SimpleNamespace
from dotenv import load_dotenv

ROOT = Path(__file__).resolve().parents[1]
SPACE = ROOT / "space"
sys.path.insert(0, str(ROOT / "scripts"))


def stage_space(destination: Path, results: Path) -> None:
    from push_results_to_hf import stage_public_results

    for name in ("app.py", "requirements.txt", "constraints.txt", "README.md"):
        shutil.copy2(SPACE / name, destination / name)
    shutil.copytree(
        SPACE / "src",
        destination / "src",
        ignore=shutil.ignore_patterns("__pycache__", "*.pyc", ".DS_Store"),
    )
    stage_public_results(results, destination / "results")
    if not list((destination / "results").glob("*/all_results.csv")):
        raise ValueError("No leaderboard CSVs. Run scripts/bootstrap_data.py first.")
    for p in (destination / "results").rglob("*.json"):
        json.loads(p.read_text())


def main() -> int:
    load_dotenv(ROOT / ".env")
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--repo", default=os.getenv("HF_SPACE_REPO", "ThinkcatLab/LiveHouse-TS")
    )
    parser.add_argument("--results", type=Path, default=SPACE / "results")
    parser.add_argument("--results-repo", default=os.getenv("HF_RESULTS_REPO"))
    parser.add_argument(
        "--dry-run", action="store_true", help="Validate deployment without publishing"
    )
    parser.add_argument(
        "--app-only",
        action="store_true",
        help="Update UI code while retaining all remote result tables",
    )
    args = parser.parse_args()
    from push_results_to_hf import (
        ensure_space_repo,
        get_hf_api,
        prune_remote_non_leaderboard_artifacts,
        push_via_api,
    )

    with tempfile.TemporaryDirectory(prefix="livehouse-space-") as temp:
        stage = Path(temp)
        stage_space(stage, args.results)
        # Results have their own Dataset; the Space holds application code only.
        shutil.rmtree(stage / "results")
        count = sum(p.is_file() for p in stage.rglob("*"))
        print(f"Validated Space deployment: {count} public files.")
        if args.dry_run:
            return 0
        token = (os.getenv("HF_TOKEN") or "").strip()
        if not token:
            raise SystemExit("HF_TOKEN required in .env")
        api = get_hf_api(token)
        # Stop an older snapshot from replacing newer online scores.
        from huggingface_hub import hf_hub_download
        from huggingface_hub.utils import EntryNotFoundError, RepositoryNotFoundError

        remote = None
        results_repo = args.results_repo or args.repo + "-results"
        if not args.app_only:
            try:
                remote = Path(
                    hf_hub_download(
                        results_repo,
                        "results/online_status.json",
                        repo_type="dataset",
                        token=token,
                    )
                )
            except (EntryNotFoundError, RepositoryNotFoundError):
                pass
        if remote:
            from datetime import datetime

            online = json.loads(remote.read_text()).get("finished_at")
            local = json.loads((args.results / "online_status.json").read_text()).get(
                "finished_at"
            )
            if online and (
                not local
                or datetime.fromisoformat(online.replace("Z", "+00:00"))
                > datetime.fromisoformat(local.replace("Z", "+00:00"))
            ):
                raise SystemExit(
                    "Remote evaluation is newer. Restore its latest operator backup before deploying; refusing rollback."
                )
        ensure_space_repo(token, args.repo)
        if not args.app_only:
            if push_via_api(SimpleNamespace(repo=args.repo, results_repo=results_repo,
                                            results=args.results, restart_space=False,
                                            commit_message="Publish verified LiveHouse-TS metrics"), token):
                return 1
        if not prune_remote_non_leaderboard_artifacts(token, args.repo):
            return 1
        variables = api.get_space_variables(args.repo)
        current = variables.get("HF_RESULTS_REPO")
        if current is None or current.value != results_repo:
            api.add_space_variable(args.repo, key="HF_RESULTS_REPO", value=results_repo)
        for attempt in range(3):
            try:
                commit = api.upload_folder(
                    folder_path=stage, repo_id=args.repo, repo_type="space",
                    commit_message="Deploy maintained LiveHouse-TS application",
                )
                break
            except Exception as exc:
                status = getattr(getattr(exc, "response", None), "status_code", None)
                if attempt == 2 or status not in (None, 429, 502, 503, 504):
                    raise
                time.sleep(3 * (attempt + 1))
        print(f"Deployed {commit.oid}: https://huggingface.co/spaces/{args.repo}")
        print("Run scripts/check_health.py --online to verify the rebuilt Space.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())