scripts/prove_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 import audit_goal_readiness, check_deployment_readiness, check_research_sources, research_watchlist
from scripts.next_deployment_step import deployment_url_warnings
from scripts.verify_site import verify_site
from scripts.verify_worker import verify_worker


def prove_live_deployment(
    worker_url: str,
    origin: str,
    code: str = "1234",
    out: Path = ROOT_DIR / "outputs" / "worker-verification.json",
    site_out: Path = ROOT_DIR / "outputs" / "site-verification.json",
    proof_out: Path = ROOT_DIR / "outputs" / "live-deployment-proof.json",
    smoke_voice_id: str = "espeak-ar-clear",
    smoke_ocr_engine: str = "arabic",
    smoke_timeout: float = 180,
    check_hf_metadata: bool = False,
    hf_metadata_report: Path | None = None,
) -> dict[str, object]:
    url_warnings = deployment_url_warnings(worker_url, origin)
    if url_warnings:
        raise ValueError("Live proof requires real deployed URLs: " + "; ".join(url_warnings))
    out.parent.mkdir(parents=True, exist_ok=True)
    site_out.parent.mkdir(parents=True, exist_ok=True)
    proof_out.parent.mkdir(parents=True, exist_ok=True)
    site_checks = verify_site(origin, code, worker_url)
    worker_checks = verify_worker(
        worker_url,
        code,
        origin,
        require_cors=True,
        smoke_upload=True,
        smoke_scanned=True,
        smoke_voice_id=smoke_voice_id,
        smoke_ocr_engine=smoke_ocr_engine,
        smoke_timeout=smoke_timeout,
    )
    site_out.write_text(json.dumps([check.__dict__ for check in site_checks], ensure_ascii=False, indent=2), encoding="utf-8")
    out.write_text(json.dumps([check.__dict__ for check in worker_checks], ensure_ascii=False, indent=2), encoding="utf-8")
    audit_checks = audit_goal_readiness.collect_checks(worker_report=out, site_report=site_out)
    summary = audit_goal_readiness.summarize(audit_checks)
    if check_hf_metadata and hf_metadata_report is None:
        hf_metadata_report = proof_out.parent / "hf-model-metadata.md"
    research_checks = check_research_sources.collect_command_checks(
        check_hf_metadata=check_hf_metadata,
        write_hf_metadata_report=hf_metadata_report if check_hf_metadata else None,
    )
    research = check_research_sources.summarize(research_checks)
    license_policy_violations = research_watchlist.license_policy_violations(research_watchlist.CANDIDATES)
    license_policy = {"ready": not license_policy_violations, "violations": license_policy_violations}
    deployment = check_deployment_readiness.summarize(check_deployment_readiness.collect_checks())
    complete = bool(
        summary["complete"]
        and research["ready"]
        and license_policy["ready"]
        and deployment["ready"]
        and all(check.ok for check in site_checks)
        and all(check.ok for check in worker_checks)
    )
    result = {
        "workerUrl": worker_url.rstrip("/"),
        "origin": origin.rstrip("/"),
        "workerReport": str(out),
        "siteReport": str(site_out),
        "proofReport": str(proof_out),
        "hfMetadataReport": str(hf_metadata_report) if hf_metadata_report else None,
        "siteOk": all(check.ok for check in site_checks),
        "workerOk": all(check.ok for check in worker_checks),
        "researchOk": bool(research["ready"]),
        "licensePolicyOk": bool(license_policy["ready"]),
        "deploymentOk": bool(deployment["ready"]),
        "complete": complete,
        "audit": summary,
        "research": research,
        "licensePolicy": license_policy,
        "deployment": deployment,
        "siteChecks": [check.__dict__ for check in site_checks],
        "workerChecks": [check.__dict__ for check in worker_checks],
    }
    proof_out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    return result


def print_result(result: dict[str, object]) -> None:
    print("Vercel site verification")
    for item in result["siteChecks"]:
        label = "PASS" if item.get("ok") else "FAIL"
        print(f"{label:<4}  {item.get('name', ''):<28}  {item.get('detail', '')}")
    print()
    print("Worker verification")
    for item in result["workerChecks"]:
        label = "PASS" if item.get("ok") else "FAIL"
        print(f"{label:<4}  {item.get('name', ''):<28}  {item.get('detail', '')}")
    print()
    print(f"Saved site report: {result['siteReport']}")
    print(f"Saved worker report: {result['workerReport']}")
    print(f"Saved live proof: {result['proofReport']}")
    if result.get("hfMetadataReport"):
        print(f"Saved Hugging Face metadata report: {result['hfMetadataReport']}")
    print()
    audit_goal_readiness.print_table(
        [
            audit_goal_readiness.RequirementCheck(
                requirement=str(item["requirement"]),
                status=item["status"],
                evidence=str(item["evidence"]),
            )
            for item in result["audit"]["checks"]
        ]
    )


def main() -> None:
    parser = argparse.ArgumentParser(description="Verify the live worker and complete the Arabic audio goal audit with evidence.")
    parser.add_argument("worker_url", help="Worker URL, for example https://your-space.hf.space")
    parser.add_argument("--origin", required=True, help="Vercel site origin, for example https://your-app.vercel.app")
    parser.add_argument("--code", default="1234", help="Access code used by the worker.")
    parser.add_argument("--out", type=Path, default=ROOT_DIR / "outputs" / "worker-verification.json")
    parser.add_argument("--site-out", type=Path, default=ROOT_DIR / "outputs" / "site-verification.json")
    parser.add_argument("--proof-out", type=Path, default=ROOT_DIR / "outputs" / "live-deployment-proof.json")
    parser.add_argument("--smoke-voice-id", default="espeak-ar-clear", help="Voice id for the smoke uploads.")
    parser.add_argument("--smoke-ocr-engine", default="arabic", help="OCR mode for the smoke uploads.")
    parser.add_argument("--smoke-timeout", type=float, default=180, help="Seconds to wait for each smoke job.")
    parser.add_argument("--check-hf-metadata", action="store_true", help="Fetch Hugging Face model metadata during live proof.")
    parser.add_argument("--hf-metadata-report", type=Path, help="Markdown output path for --check-hf-metadata.")
    parser.add_argument("--json", action="store_true", help="Print machine-readable JSON.")
    args = parser.parse_args()

    try:
        result = prove_live_deployment(
            args.worker_url,
            args.origin,
            code=args.code,
            out=args.out,
            site_out=args.site_out,
            proof_out=args.proof_out,
            smoke_voice_id=args.smoke_voice_id,
            smoke_ocr_engine=args.smoke_ocr_engine,
            smoke_timeout=args.smoke_timeout,
            check_hf_metadata=args.check_hf_metadata,
            hf_metadata_report=args.hf_metadata_report,
        )
    except ValueError as exc:
        parser.error(str(exc))
    if args.json:
        print(json.dumps(result, ensure_ascii=False, indent=2))
    else:
        print_result(result)
    if not result["complete"]:
        raise SystemExit(1)


if __name__ == "__main__":
    main()