scripts/finish_live_deployment.py

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path


ROOT_DIR = Path(__file__).resolve().parent.parent
if str(ROOT_DIR) not in sys.path:
    sys.path.insert(0, str(ROOT_DIR))

from scripts.configure_vercel_worker import DEFAULT_SITE_URL, configure_vercel_worker
from scripts.deploy_hf_space import DEFAULT_BUNDLE_DIR, deploy_hf_space, token_available, worker_url_for_repo
from scripts.prove_live_deployment import prove_live_deployment


def finish_live_deployment(
    repo_id: str,
    site_url: str = DEFAULT_SITE_URL,
    code: str = "1234",
    bundle_dir: Path = DEFAULT_BUNDLE_DIR,
    token: str | None = None,
    private: bool = False,
    force_export: bool = True,
    verify_site_after_vercel: bool = True,
    run_proof: bool = True,
    smoke_ocr_engine: str = "arabic",
    smoke_voice_id: str = "espeak-ar-clear",
    smoke_timeout: float = 180,
) -> dict[str, object]:
    deploy_result = deploy_hf_space(
        repo_id=repo_id,
        bundle_dir=bundle_dir,
        token=token,
        private=private,
        force_export=force_export,
    )
    worker_url = str(deploy_result["workerUrl"])
    vercel_result = configure_vercel_worker(
        worker_url,
        site_url=site_url,
        code=code,
        redeploy=True,
        verify=verify_site_after_vercel,
    )
    proof_result = None
    if run_proof:
        proof_result = prove_live_deployment(
            worker_url,
            site_url,
            code=code,
            smoke_voice_id=smoke_voice_id,
            smoke_ocr_engine=smoke_ocr_engine,
            smoke_timeout=smoke_timeout,
        )
    return {
        "repoId": repo_id,
        "siteUrl": site_url.rstrip("/"),
        "workerUrl": worker_url,
        "workerDeploy": deploy_result,
        "vercel": vercel_result,
        "proof": proof_result,
        "complete": bool(proof_result and proof_result.get("complete")),
        "nextCommand": (
            f"python scripts\\prove_live_deployment.py {worker_url} --origin {site_url.rstrip('/')} "
            f"--code {code} --smoke-ocr-engine {smoke_ocr_engine}"
        ),
    }


def dry_run_finish_live_deployment(
    repo_id: str,
    site_url: str = DEFAULT_SITE_URL,
    code: str = "1234",
    bundle_dir: Path = DEFAULT_BUNDLE_DIR,
    token: str | None = None,
    smoke_ocr_engine: str = "arabic",
) -> dict[str, object]:
    worker_url = worker_url_for_repo(repo_id)
    site_url = site_url.rstrip("/")
    return {
        "dryRun": True,
        "repoId": repo_id,
        "siteUrl": site_url,
        "workerUrl": worker_url,
        "bundleDir": str(bundle_dir),
        "tokenAvailable": token_available(token),
        "willUploadWorker": True,
        "willConfigureVercel": True,
        "willRunProof": True,
        "commands": {
            "setToken": '$env:HF_TOKEN="<your-hugging-face-write-token>"',
            "finish": (
                f"python scripts\\finish_live_deployment.py {repo_id} "
                f"--site-url {site_url} --code {code} --smoke-ocr-engine {smoke_ocr_engine}"
            ),
            "configureVercelOnly": (
                f"python scripts\\configure_vercel_worker.py {worker_url} "
                f"--site-url {site_url} --code {code} --verify"
            ),
            "proofOnly": (
                f"python scripts\\prove_live_deployment.py {worker_url} "
                f"--origin {site_url} --code {code} --smoke-ocr-engine {smoke_ocr_engine}"
            ),
        },
    }


def main_with_args(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Upload the HF worker, configure Vercel WORKER_BASE_URL, redeploy, and run live proof."
    )
    parser.add_argument("repo_id", help="Hugging Face Space repo id, e.g. username/arabic-audio-reader-worker.")
    parser.add_argument("--site-url", default=DEFAULT_SITE_URL)
    parser.add_argument("--code", default="1234")
    parser.add_argument("--bundle-dir", type=Path, default=DEFAULT_BUNDLE_DIR)
    parser.add_argument("--token", help="Hugging Face write token. Prefer HF_TOKEN in the environment.")
    parser.add_argument("--private", action="store_true")
    parser.add_argument("--no-force-export", action="store_true")
    parser.add_argument("--skip-site-verify", action="store_true")
    parser.add_argument("--skip-proof", action="store_true")
    parser.add_argument("--dry-run", action="store_true", help="Print the planned deployment without changing HF or Vercel.")
    parser.add_argument("--smoke-ocr-engine", default="arabic")
    parser.add_argument("--smoke-voice-id", default="espeak-ar-clear")
    parser.add_argument("--smoke-timeout", type=float, default=180)
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args(argv)

    if args.dry_run:
        result = dry_run_finish_live_deployment(
            args.repo_id,
            site_url=args.site_url,
            code=args.code,
            bundle_dir=args.bundle_dir,
            token=args.token,
            smoke_ocr_engine=args.smoke_ocr_engine,
        )
        if args.json:
            print(json.dumps(result, ensure_ascii=False, indent=2))
        else:
            print(f"Predicted worker URL: {result['workerUrl']}")
            print(f"Hugging Face token available: {'yes' if result['tokenAvailable'] else 'no'}")
            print("Command:")
            print(result["commands"]["finish"])
        return 0

    result = finish_live_deployment(
        args.repo_id,
        site_url=args.site_url,
        code=args.code,
        bundle_dir=args.bundle_dir,
        token=args.token,
        private=args.private,
        force_export=not args.no_force_export,
        verify_site_after_vercel=not args.skip_site_verify,
        run_proof=not args.skip_proof,
        smoke_ocr_engine=args.smoke_ocr_engine,
        smoke_voice_id=args.smoke_voice_id,
        smoke_timeout=args.smoke_timeout,
    )
    if args.json:
        print(json.dumps(result, ensure_ascii=False, indent=2))
    else:
        print(f"Worker URL: {result['workerUrl']}")
        print(f"Vercel site: {result['siteUrl']}")
        if result["complete"]:
            print("Live deployment proof is complete.")
        else:
            print("Live deployment still needs proof:")
            print(result["nextCommand"])
    return 0 if result["complete"] or args.skip_proof else 1


def main() -> int:
    return main_with_args()


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