#!/usr/bin/env python3 """Validate and publish reproducible model and Gradio Space folders to Hugging Face.""" from __future__ import annotations import argparse import hashlib import json import math import os import random import socket import sys import tempfile import time from collections.abc import Callable, Mapping from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any, TypeVar ROOT = Path(__file__).resolve().parents[1] _HUB_MANAGED_FILES = {".gitattributes"} _TRANSIENT_HTTP_STATUS = {408, 425, 429, 500, 502, 503, 504} _FAILED_SPACE_STAGES = { "BUILD_ERROR", "CONFIG_ERROR", "DELETING", "NO_APP_FILE", "PAUSED", "RUNTIME_ERROR", "STOPPED", } _INTERMEDIATE_SPACE_STAGES = { "APP_STARTING", "BUILDING", "RUNNING_APP_STARTING", "RUNNING_BUILDING", } _T = TypeVar("_T") class PublishFailure(RuntimeError): """A sanitized, actionable publication failure safe to show to a user.""" class SpaceRuntimeFailure(PublishFailure): """A Space build/runtime failure carrying only its non-secret stage.""" def __init__(self, message: str, *, stage: str | None) -> None: super().__init__(message) self.stage = stage @dataclass(frozen=True) class RetryPolicy: max_attempts: int = 4 base_seconds: float = 1.0 max_seconds: float = 30.0 request_timeout_seconds: float = 120.0 sleeper: Callable[[float], None] = time.sleep jitter: Callable[[float, float], float] = random.uniform @dataclass(frozen=True) class RemoteVerification: matches: bool paths: frozenset[str] reason: str @dataclass(frozen=True) class ComponentResult: repo_id: str repo_type: str commit_sha: str commit_url: str action: str verified: bool = True def as_dict(self) -> dict[str, Any]: return { "repo_id": self.repo_id, "repo_type": self.repo_type, "commit_sha": self.commit_sha, "commit_url": self.commit_url, "action": self.action, "verified": self.verified, } @dataclass(frozen=True) class SpaceRuntimeEvidence: stage: str build_transition_observed: bool commit_bound: bool = False def _sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def _validate_inventory(release_dir: Path, manifest: dict[str, Any]) -> None: inventory = manifest.get("files") if not isinstance(inventory, dict) or not inventory: raise SystemExit("release manifest has no file inventory; rebuild the release") manifest_path = release_dir / "release_manifest.json" if manifest_path.is_symlink() or not manifest_path.is_file(): raise SystemExit("release manifest must be a regular file") symlinks = [path for path in release_dir.rglob("*") if path.is_symlink()] if symlinks: raise SystemExit("release contents contain a symlink; rebuild") expected_paths = set(inventory) actual_paths = { path.relative_to(release_dir).as_posix() for path in release_dir.rglob("*") if path.is_file() and path != manifest_path } if actual_paths != expected_paths: raise SystemExit("release contents differ from the hash-bound file inventory; rebuild") for relative, evidence in inventory.items(): path = (release_dir / relative).resolve() try: path.relative_to(release_dir.resolve()) except ValueError as exc: raise SystemExit(f"release inventory escapes its root: {relative}") from exc if path.is_symlink() or not path.is_file() or not isinstance(evidence, dict): raise SystemExit(f"invalid release inventory entry: {relative}") if evidence.get("bytes") != path.stat().st_size or evidence.get("sha256") != _sha256(path): raise SystemExit(f"release file failed integrity validation: {relative}") serialized = json.dumps(inventory, sort_keys=True, separators=(",", ":")).encode() expected_digest = hashlib.sha256(serialized).hexdigest() if manifest.get("release_inventory_sha256") != expected_digest: raise SystemExit("release inventory digest is invalid; rebuild") def _validate_provenance(release_dir: Path) -> None: """Check that the portable inner provenance describes the two upload folders.""" path = release_dir / "model" / "release_provenance.json" try: provenance = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise SystemExit("release_provenance.json is missing or invalid") from exc inventories = provenance.get("files") if not isinstance(inventories, dict): raise SystemExit("release provenance has no component inventories; rebuild") expected_digest = hashlib.sha256( json.dumps(inventories, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() if provenance.get("source_and_artifact_inventory_sha256") != expected_digest: raise SystemExit("release provenance inventory digest is invalid; rebuild") for component in ("model", "space"): expected = inventories.get(component) if not isinstance(expected, dict): raise SystemExit(f"release provenance is missing the {component} inventory") folder = release_dir / component actual: dict[str, dict[str, int | str]] = {} for file_path in folder.rglob("*"): if not file_path.is_file(): continue relative = file_path.relative_to(folder).as_posix() if component == "model" and relative == "release_provenance.json": continue actual[relative] = { "bytes": file_path.stat().st_size, "sha256": _sha256(file_path), } if actual != expected: raise SystemExit(f"release provenance disagrees with the {component} folder; rebuild") def _validated_release_status(release_dir: Path, manifest: dict[str, Any]) -> bool: """Derive policy from inventoried files, not mutable outer-manifest text.""" metadata_path = release_dir / "model" / "model_metadata.json" provenance_path = release_dir / "model" / "release_provenance.json" try: metadata = json.loads(metadata_path.read_text(encoding="utf-8")) provenance = json.loads(provenance_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise SystemExit("inventoried release policy files are invalid") from exc if not isinstance(metadata, dict) or not isinstance(provenance, dict): raise SystemExit("inventoried release policy files must be JSON objects") development_value = metadata.get("development_only") if not isinstance(development_value, bool): raise SystemExit("inventoried metadata development_only must be a boolean") development_only = development_value if not development_only and str(metadata.get("training_status", "")).lower() != "final": raise SystemExit("inventoried metadata has an invalid final-release status") if provenance.get("development_only") is not development_only: raise SystemExit("release provenance disagrees with model metadata") if manifest.get("development_only") is not development_only: raise SystemExit("outer release status disagrees with inventoried metadata") expected_metrics = "development_metrics.json" if development_only else "test_metrics.json" if manifest.get("metrics_file") != expected_metrics: raise SystemExit("outer metrics classification disagrees with inventoried metadata") expected_scope = "development" if development_only else "official_test" if manifest.get("metrics_scope") != expected_scope: raise SystemExit("outer metrics scope disagrees with inventoried metadata") if manifest.get("has_test_metrics") is not (not development_only): raise SystemExit("outer test-metrics flag disagrees with inventoried metadata") if not (release_dir / "model" / expected_metrics).is_file(): raise SystemExit(f"inventoried release is missing {expected_metrics}") model_path = (release_dir / "model" / str(manifest.get("model_file", ""))).resolve() try: model_path.relative_to((release_dir / "model").resolve()) except ValueError as exc: raise SystemExit("outer model_file escapes the model release directory") from exc if model_path.is_symlink() or not model_path.is_file(): raise SystemExit("outer model_file is not an inventoried regular file") export_path = release_dir / "model" / "export_manifest.json" metrics_path = release_dir / "model" / expected_metrics bindings = { "model_sha256": _sha256(model_path), "metadata_sha256": _sha256(metadata_path), "export_manifest_sha256": _sha256(export_path), "metrics_sha256": _sha256(metrics_path), } for name, actual in bindings.items(): if manifest.get(name) != actual: raise SystemExit(f"outer release manifest has invalid {name}") return development_only def _load_env_token(path: Path) -> str | None: if not path.is_file(): return None for raw_line in path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) if key.strip() in {"HF_TOKEN", "hf_token", "HUGGING_FACE_HUB_TOKEN"}: token = value.strip().strip("'\"") return token or None return None def _http_status(exc: BaseException) -> int | None: seen: set[int] = set() current: BaseException | None = exc while current is not None and id(current) not in seen: seen.add(id(current)) response = getattr(current, "response", None) status = getattr(response, "status_code", None) if isinstance(status, int): return status direct_status = getattr(current, "status_code", None) if isinstance(direct_status, int): return direct_status current = current.__cause__ or current.__context__ return None def _exception_chain(exc: BaseException) -> list[BaseException]: chain: list[BaseException] = [] seen: set[int] = set() current: BaseException | None = exc while current is not None and id(current) not in seen: seen.add(id(current)) chain.append(current) current = current.__cause__ or current.__context__ return chain def _is_transient(exc: BaseException) -> bool: status = _http_status(exc) if status is not None: return status in _TRANSIENT_HTTP_STATUS or 500 <= status <= 599 for item in _exception_chain(exc): if isinstance(item, (ConnectionError, TimeoutError, socket.gaierror)): return True module = type(item).__module__.split(".", 1)[0] name = type(item).__name__.lower() if module in {"httpcore", "httpx", "requests", "urllib3"}: hierarchy = {base.__name__.lower() for base in type(item).__mro__} if hierarchy & { "connectionerror", "networkerror", "protocolerror", "timeoutexception", "transporterror", } or any( marker in name for marker in ( "close", "connect", "network", "protocol", "read", "timeout", "write", ) ): return True return False def _failure_kind(exc: BaseException) -> str: status = _http_status(exc) if status in {401, 403}: return f"authentication/authorization rejection (HTTP {status})" if status == 402: return ( "Hugging Face plan/hardware entitlement required (HTTP 402); " "verify free ZeroGPU eligibility and an available slot for " "--space-hardware zero-a10g, otherwise upgrade the account plan" ) if status is not None: return f"HTTP {status}" chain = _exception_chain(exc) if any(isinstance(item, socket.gaierror) for item in chain): return "DNS resolution error" if any( isinstance(item, TimeoutError) or "timeout" in type(item).__name__.lower() for item in chain ): return "network timeout" if any( isinstance(item, ConnectionError) or "connect" in type(item).__name__.lower() for item in chain ): return "connection error" if any( type(item).__module__.split(".", 1)[0] in {"httpcore", "httpx", "requests", "urllib3"} and any( marker in type(item).__name__.lower() for marker in ("close", "network", "protocol", "read", "write") ) for item in chain ): return "network transport error" return type(exc).__name__ def _delay(policy: RetryPolicy, attempt: int) -> float: unjittered = min(policy.max_seconds, policy.base_seconds * (2 ** max(0, attempt - 1))) return max(0.0, policy.jitter(unjittered * 0.8, unjittered * 1.2)) def _call_with_retry(label: str, operation: Callable[[], _T], policy: RetryPolicy) -> _T: for attempt in range(1, policy.max_attempts + 1): try: return operation() except Exception as exc: transient = _is_transient(exc) if not transient or attempt == policy.max_attempts: raise PublishFailure( f"{label} failed after {attempt} attempt(s): {_failure_kind(exc)}" ) from None delay = _delay(policy, attempt) print( f"{label}: transient {_failure_kind(exc)}; " f"retrying {attempt + 1}/{policy.max_attempts} in {delay:.1f}s", file=sys.stderr, ) policy.sleeper(delay) raise AssertionError("unreachable") def _is_not_found(exc: BaseException) -> bool: return _http_status(exc) == 404 def _repo_info_or_none( api: Any, *, repo_id: str, repo_type: str, policy: RetryPolicy, ) -> Any | None: for attempt in range(1, policy.max_attempts + 1): try: return api.repo_info( repo_id=repo_id, repo_type=repo_type, timeout=policy.request_timeout_seconds, ) except Exception as exc: if _is_not_found(exc): return None if not _is_transient(exc) or attempt == policy.max_attempts: raise PublishFailure( f"inspect {repo_type} repository failed after {attempt} attempt(s): " f"{_failure_kind(exc)}" ) from None delay = _delay(policy, attempt) print( f"inspect {repo_type} repository: transient {_failure_kind(exc)}; " f"retrying {attempt + 1}/{policy.max_attempts} in {delay:.1f}s", file=sys.stderr, ) policy.sleeper(delay) raise AssertionError("unreachable") def _component_inventory(manifest: Mapping[str, Any], component: str) -> dict[str, dict[str, Any]]: prefix = f"{component}/" inventory = manifest.get("files", {}) return { relative[len(prefix) :]: evidence for relative, evidence in inventory.items() if relative.startswith(prefix) } def _verify_remote( api: Any, snapshot_download: Callable[..., str], *, repo_id: str, repo_type: str, revision: str, expected: Mapping[str, Mapping[str, Any]], token: str, policy: RetryPolicy, ) -> RemoteVerification: remote_paths = frozenset( _call_with_retry( f"list pinned files for {repo_type} repository", lambda: api.list_repo_files( repo_id=repo_id, repo_type=repo_type, revision=revision, ), policy, ) ) expected_paths = set(expected) missing = expected_paths - remote_paths unexpected = set(remote_paths) - expected_paths - _HUB_MANAGED_FILES if missing or unexpected: details: list[str] = [] if missing: details.append(f"{len(missing)} missing") if unexpected: details.append(f"{len(unexpected)} unexpected") return RemoteVerification(False, remote_paths, ", ".join(details)) with tempfile.TemporaryDirectory(prefix="hf-publish-verify-") as cache_dir: snapshot_path = Path( _call_with_retry( f"download pinned snapshot for {repo_type} repository", lambda: snapshot_download( repo_id=repo_id, repo_type=repo_type, revision=revision, cache_dir=cache_dir, force_download=True, token=token, ), policy, ) ) for relative, evidence in expected.items(): path = snapshot_path / relative if not path.is_file(): return RemoteVerification(False, remote_paths, f"missing snapshot file {relative}") if path.stat().st_size != evidence.get("bytes") or _sha256(path) != evidence.get( "sha256" ): return RemoteVerification(False, remote_paths, f"hash mismatch for {relative}") return RemoteVerification(True, remote_paths, "exact package mirror") def _repo_url(repo_id: str, repo_type: str) -> str: if repo_type == "space": return f"https://huggingface.co/spaces/{repo_id}" return f"https://huggingface.co/{repo_id}" def _commit_url(repo_id: str, repo_type: str, revision: str) -> str: return f"{_repo_url(repo_id, repo_type)}/commit/{revision}" def _validate_repo_settings(info: Any, *, repo_id: str, repo_type: str, private: bool) -> None: actual_id = str(getattr(info, "id", repo_id)) if actual_id.casefold() != repo_id.casefold(): raise PublishFailure(f"Hub returned an unexpected {repo_type} repository identity") actual_private = getattr(info, "private", None) if actual_private is not None and bool(actual_private) is not private: visibility = "private" if actual_private else "public" requested = "private" if private else "public" raise PublishFailure( f"existing {repo_type} repository is {visibility}, but {requested} was requested" ) if repo_type == "space": sdk = getattr(info, "sdk", None) if sdk is not None and str(sdk).casefold() != "gradio": raise PublishFailure(f"existing Space SDK is {sdk!r}, not 'gradio'") def _ensure_repo( api: Any, *, repo_id: str, repo_type: str, private: bool, space_hardware: str | None, policy: RetryPolicy, ) -> Any: info = _repo_info_or_none(api, repo_id=repo_id, repo_type=repo_type, policy=policy) if info is None: create_kwargs: dict[str, Any] = { "repo_id": repo_id, "repo_type": repo_type, "private": private, "exist_ok": True, } if repo_type == "space": create_kwargs["space_sdk"] = "gradio" if space_hardware is not None: create_kwargs["space_hardware"] = space_hardware _call_with_retry( f"create {repo_type} repository", lambda: api.create_repo(**create_kwargs), policy, ) info = _call_with_retry( f"inspect created {repo_type} repository", lambda: api.repo_info( repo_id=repo_id, repo_type=repo_type, timeout=policy.request_timeout_seconds, ), policy, ) _validate_repo_settings(info, repo_id=repo_id, repo_type=repo_type, private=private) auth_check = getattr(api, "auth_check", None) if callable(auth_check): _call_with_retry( f"verify write access for {repo_type} repository", lambda: auth_check(repo_id=repo_id, repo_type=repo_type, write=True), policy, ) return info def _current_verification( api: Any, snapshot_download: Callable[..., str], *, repo_id: str, repo_type: str, expected: Mapping[str, Mapping[str, Any]], token: str, policy: RetryPolicy, ) -> tuple[Any, RemoteVerification]: for stability_attempt in range(1, policy.max_attempts + 1): before = _call_with_retry( f"inspect current {repo_type} repository head", lambda: api.repo_info( repo_id=repo_id, repo_type=repo_type, timeout=policy.request_timeout_seconds, ), policy, ) revision = str(getattr(before, "sha", "") or "") if revision: verification = _verify_remote( api, snapshot_download, repo_id=repo_id, repo_type=repo_type, revision=revision, expected=expected, token=token, policy=policy, ) else: verification = RemoteVerification(False, frozenset(), "repository has no commit") after = _call_with_retry( f"recheck current {repo_type} repository head", lambda: api.repo_info( repo_id=repo_id, repo_type=repo_type, timeout=policy.request_timeout_seconds, ), policy, ) after_revision = str(getattr(after, "sha", "") or "") if after_revision == revision: return after, verification if stability_attempt == policy.max_attempts: raise PublishFailure( f"{repo_type} repository head kept changing during pinned verification" ) print( f"{repo_type} repository head changed during pinned verification; " f"retrying {stability_attempt + 1}/{policy.max_attempts}", file=sys.stderr, ) raise AssertionError("unreachable") def _publish_component( api: Any, snapshot_download: Callable[..., str], *, folder: Path, expected: Mapping[str, Mapping[str, Any]], repo_id: str, repo_type: str, private: bool, token: str, replace_existing: bool, release_kind: str, inventory_digest: str, policy: RetryPolicy, space_hardware: str | None = None, ) -> ComponentResult: _ensure_repo( api, repo_id=repo_id, repo_type=repo_type, private=private, space_hardware=space_hardware if repo_type == "space" else None, policy=policy, ) info, initial = _current_verification( api, snapshot_download, repo_id=repo_id, repo_type=repo_type, expected=expected, token=token, policy=policy, ) base_sha = str(getattr(info, "sha", "") or "") if initial.matches: return ComponentResult( repo_id=repo_id, repo_type=repo_type, commit_sha=base_sha, commit_url=_commit_url(repo_id, repo_type, base_sha), action="already_current", ) meaningful_existing = set(initial.paths) - _HUB_MANAGED_FILES if meaningful_existing and not replace_existing: raise PublishFailure( f"refusing to replace divergent existing {repo_type} repository {repo_id!r} " f"({len(meaningful_existing)} remote path(s)); inspect it and pass " "--replace-existing explicitly" ) message_verb = "Deploy" if repo_type == "space" else "Publish" upload_kwargs = { "repo_id": repo_id, "repo_type": repo_type, "folder_path": str(folder), "commit_message": f"{message_verb} Tiny Hinglish Turn Detector {release_kind}", "commit_description": f"release inventory sha256: {inventory_digest}", "delete_patterns": "*", "parent_commit": base_sha or None, } for attempt in range(1, policy.max_attempts + 1): try: commit = api.upload_folder(**upload_kwargs) except Exception as exc: current_info, current = _current_verification( api, snapshot_download, repo_id=repo_id, repo_type=repo_type, expected=expected, token=token, policy=policy, ) current_sha = str(getattr(current_info, "sha", "") or "") if current.matches: return ComponentResult( repo_id=repo_id, repo_type=repo_type, commit_sha=current_sha, commit_url=_commit_url(repo_id, repo_type, current_sha), action="reconciled_after_error", ) if current_sha != base_sha: raise PublishFailure( f"{repo_type} repository head changed during upload and is not an exact mirror; " "refusing to overwrite the concurrent commit" ) from None if not _is_transient(exc) or attempt == policy.max_attempts: raise PublishFailure( f"upload {repo_type} repository failed after {attempt} attempt(s): " f"{_failure_kind(exc)}; remote head was reconciled and remains unchanged" ) from None delay = _delay(policy, attempt) print( f"upload {repo_type} repository: transient {_failure_kind(exc)}; " f"remote head unchanged, retrying {attempt + 1}/{policy.max_attempts} " f"in {delay:.1f}s", file=sys.stderr, ) policy.sleeper(delay) continue returned_commit_sha = str(getattr(commit, "oid", "") or "") if not returned_commit_sha: raise PublishFailure( f"Hub returned no commit SHA after uploading {repo_type} repository" ) current_info, current = _current_verification( api, snapshot_download, repo_id=repo_id, repo_type=repo_type, expected=expected, token=token, policy=policy, ) current_sha = str(getattr(current_info, "sha", "") or "") if current_sha != returned_commit_sha and not current.matches: raise PublishFailure( f"{repo_type} repository head changed after upload and is not an exact mirror" ) if not current.matches: raise PublishFailure( f"uploaded {repo_type} repository failed pinned remote verification: {current.reason}" ) commit_sha = current_sha if commit_sha == returned_commit_sha: commit_url = str(getattr(commit, "commit_url", "") or "") or _commit_url( repo_id, repo_type, commit_sha ) else: commit_url = _commit_url(repo_id, repo_type, commit_sha) return ComponentResult( repo_id=repo_id, repo_type=repo_type, commit_sha=commit_sha, commit_url=commit_url, action="uploaded", ) raise AssertionError("unreachable") def _space_stage(runtime: Any) -> str: stage = getattr(runtime, "stage", runtime) stage = getattr(stage, "value", stage) return str(stage).upper() def _wait_for_space( api: Any, *, repo_id: str, timeout_seconds: float, poll_seconds: float, policy: RetryPolicy, require_build_transition: bool, monotonic: Callable[[], float] = time.monotonic, ) -> SpaceRuntimeEvidence: deadline = monotonic() + timeout_seconds last_stage: str | None = None transition_observed = False while True: runtime = _call_with_retry( "inspect Space runtime", lambda: api.get_space_runtime(repo_id=repo_id), policy, ) last_stage = _space_stage(runtime) if last_stage in _FAILED_SPACE_STAGES: raise SpaceRuntimeFailure( f"Space did not become healthy; terminal runtime stage is {last_stage}", stage=last_stage, ) if last_stage in _INTERMEDIATE_SPACE_STAGES: transition_observed = True if last_stage == "RUNNING" and (transition_observed or not require_build_transition): return SpaceRuntimeEvidence( stage=last_stage, build_transition_observed=transition_observed, ) remaining = deadline - monotonic() if remaining <= 0: transition_note = ( "no new build/start transition was observed" if require_build_transition and not transition_observed else f"last stage: {last_stage}" ) raise SpaceRuntimeFailure( f"Space content is verified, but runtime did not reach RUNNING within " f"{timeout_seconds:g}s ({transition_note})", stage=last_stage, ) policy.sleeper(min(poll_seconds, remaining)) def _utc_now() -> str: return datetime.now(timezone.utc).isoformat() def _write_receipt(path: Path, receipt: Mapping[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) descriptor, temporary_name = tempfile.mkstemp( prefix=f".{path.name}.", suffix=".tmp", dir=path.parent ) temporary = Path(temporary_name) try: with os.fdopen(descriptor, "w", encoding="utf-8") as handle: json.dump(receipt, handle, indent=2, sort_keys=True) handle.write("\n") handle.flush() os.fsync(handle.fileno()) temporary.chmod(0o600) os.replace(temporary, path) except Exception: temporary.unlink(missing_ok=True) raise def _receipt_path(raw: str, release_dir: Path) -> Path: path = Path(raw) if not path.is_absolute(): path = ROOT / path path = path.resolve() try: path.relative_to(release_dir.resolve()) except ValueError: return path raise SystemExit("publish receipt must be outside the inventoried release directory") def _execute_publish( args: argparse.Namespace, *, release_dir: Path, manifest: dict[str, Any], development_only: bool, token: str, api: Any, snapshot_download: Callable[..., str], policy: RetryPolicy, ) -> dict[str, Any]: receipt_path = _receipt_path(args.receipt, release_dir) receipt: dict[str, Any] = { "format_version": 1, "release_inventory_sha256": manifest["release_inventory_sha256"], "development_only": development_only, "requested_owner": args.username, "targets": { "model": f"{args.username}/{args.model_name}", "space": f"{args.username}/{args.space_name}", }, "space_creation": { "sdk": "gradio", "hardware_if_created": args.space_hardware, }, "status": "starting", "updated_at": _utc_now(), } _write_receipt(receipt_path, receipt) try: identity = _call_with_retry("authenticate with Hugging Face", api.whoami, policy) authenticated_name = str(identity.get("name", "")) if isinstance(identity, dict) else "" if authenticated_name.casefold() != args.username.casefold(): raise PublishFailure( f"authenticated Hugging Face user {authenticated_name!r} does not match " f"requested owner {args.username!r}" ) except PublishFailure as exc: receipt["status"] = "failed" receipt["failure_stage"] = "authentication" receipt["failure"] = str(exc) receipt["updated_at"] = _utc_now() _write_receipt(receipt_path, receipt) raise except Exception as exc: failure = PublishFailure(f"authentication failed unexpectedly: {_failure_kind(exc)}") receipt["status"] = "failed" receipt["failure_stage"] = "authentication" receipt["failure"] = str(failure) receipt["updated_at"] = _utc_now() _write_receipt(receipt_path, receipt) raise failure from None model_id = f"{authenticated_name}/{args.model_name}" space_id = f"{authenticated_name}/{args.space_name}" receipt["authenticated_owner"] = authenticated_name receipt["targets"] = {"model": model_id, "space": space_id} receipt["status"] = "authenticated" receipt["updated_at"] = _utc_now() _write_receipt(receipt_path, receipt) release_kind = "development preview" if development_only else "final release" stage = "model" try: _validate_inventory(release_dir, manifest) model = _publish_component( api, snapshot_download, folder=release_dir / "model", expected=_component_inventory(manifest, "model"), repo_id=model_id, repo_type="model", private=args.private, token=token, replace_existing=args.replace_existing, release_kind=release_kind, inventory_digest=manifest["release_inventory_sha256"], policy=policy, ) receipt["model"] = model.as_dict() receipt["status"] = "model_verified" receipt["updated_at"] = _utc_now() _write_receipt(receipt_path, receipt) stage = "space" _validate_inventory(release_dir, manifest) space = _publish_component( api, snapshot_download, folder=release_dir / "space", expected=_component_inventory(manifest, "space"), repo_id=space_id, repo_type="space", private=args.private, token=token, replace_existing=args.replace_existing, release_kind=release_kind, inventory_digest=manifest["release_inventory_sha256"], policy=policy, space_hardware=args.space_hardware, ) receipt["space"] = space.as_dict() receipt["status"] = "space_content_verified" receipt["updated_at"] = _utc_now() _write_receipt(receipt_path, receipt) stage = "space_runtime" runtime_evidence = _wait_for_space( api, repo_id=space_id, timeout_seconds=args.space_timeout_seconds, poll_seconds=args.space_poll_seconds, policy=policy, require_build_transition=space.action != "already_current", ) # A concurrent push after the build can make RUNNING refer to content other than this release. current_info, current = _current_verification( api, snapshot_download, repo_id=space_id, repo_type="space", expected=_component_inventory(manifest, "space"), token=token, policy=policy, ) if not current.matches: raise PublishFailure("Space head changed before runtime verification completed") current_sha = str(getattr(current_info, "sha", "") or "") receipt["space"]["commit_sha"] = current_sha receipt["space"]["commit_url"] = _commit_url(space_id, "space", current_sha) receipt["space"]["runtime_stage"] = runtime_evidence.stage receipt["space"]["build_transition_observed"] = runtime_evidence.build_transition_observed receipt["space"]["runtime_commit_bound"] = runtime_evidence.commit_bound receipt["status"] = "content_verified_runtime_running" receipt["updated_at"] = _utc_now() _write_receipt(receipt_path, receipt) return receipt except (PublishFailure, SystemExit) as exc: receipt["status"] = "failed" receipt["failure_stage"] = stage receipt["failure"] = str(exc) if isinstance(exc, SpaceRuntimeFailure): receipt.setdefault("space", {})["runtime_stage"] = exc.stage receipt["space"]["runtime_commit_bound"] = False receipt["updated_at"] = _utc_now() _write_receipt(receipt_path, receipt) raise except Exception as exc: failure = PublishFailure(f"{stage} failed unexpectedly: {_failure_kind(exc)}") receipt["status"] = "failed" receipt["failure_stage"] = stage receipt["failure"] = str(failure) receipt["updated_at"] = _utc_now() _write_receipt(receipt_path, receipt) raise failure from None def _positive_int(raw: str) -> int: value = int(raw) if value < 1: raise argparse.ArgumentTypeError("must be at least 1") return value def _nonnegative_float(raw: str) -> float: value = float(raw) if not math.isfinite(value) or value < 0: raise argparse.ArgumentTypeError("must be finite and non-negative") return value def _positive_float(raw: str) -> float: value = float(raw) if not math.isfinite(value) or value <= 0: raise argparse.ArgumentTypeError("must be finite and greater than zero") return value def _bounded_http_client_factory( httpx_module: Any, *, connect_seconds: float, read_seconds: float, write_seconds: float, pool_seconds: float, ) -> Callable[[], Any]: timeout = httpx_module.Timeout( connect=connect_seconds, read=read_seconds, write=write_seconds, pool=pool_seconds, ) def factory() -> Any: return httpx_module.Client(follow_redirects=True, timeout=timeout) return factory def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--username", default="suvradeepp") parser.add_argument("--model-name", default="tiny-hinglish-turn-detector") parser.add_argument("--space-name", default="tiny-hinglish-turn-detector") parser.add_argument( "--space-hardware", default="zero-a10g", help=( "Hardware requested only when creating a new Space (default: zero-a10g, " "the free-account-compatible ZeroGPU tier)" ), ) parser.add_argument("--release-dir", default="release") parser.add_argument("--receipt", default="reports/hf_publish_receipt.json") parser.add_argument("--private", action="store_true") parser.add_argument( "--replace-existing", action="store_true", help="Explicitly allow replacing a non-empty divergent existing target repository", ) parser.add_argument("--max-attempts", type=_positive_int, default=4) parser.add_argument("--retry-base-seconds", type=_nonnegative_float, default=1.0) parser.add_argument("--space-timeout-seconds", type=_nonnegative_float, default=900.0) parser.add_argument("--space-poll-seconds", type=_positive_float, default=10.0) parser.add_argument("--http-connect-timeout-seconds", type=_positive_float, default=10.0) parser.add_argument("--http-read-timeout-seconds", type=_positive_float, default=120.0) parser.add_argument("--http-write-timeout-seconds", type=_positive_float, default=120.0) parser.add_argument("--http-pool-timeout-seconds", type=_positive_float, default=10.0) parser.add_argument( "--allow-development-release", action="store_true", help="Allow publishing a manifest explicitly marked development_only", ) parser.add_argument( "--execute", action="store_true", help="Perform external repository creation/uploads; otherwise only validate offline", ) return parser.parse_args() def main() -> int: args = parse_args() release_dir = Path(args.release_dir) if not release_dir.is_absolute(): release_dir = ROOT / release_dir model_folder = release_dir / "model" space_folder = release_dir / "space" for path in (model_folder, space_folder, release_dir / "release_manifest.json"): if not path.exists(): raise SystemExit(f"missing release artifact: {path}; run scripts/build_release.py") try: manifest = json.loads((release_dir / "release_manifest.json").read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError) as exc: raise SystemExit("release_manifest.json is missing or invalid") from exc _validate_inventory(release_dir, manifest) _validate_provenance(release_dir) development_only = _validated_release_status(release_dir, manifest) if development_only and not args.allow_development_release: raise SystemExit( "refusing to publish a development-only artifact without --allow-development-release" ) if not args.execute: release_kind = "development" if development_only else "final" print( f"validated {release_kind} release folders for {args.username} entirely offline; " "pass --execute to publish" ) return 0 token = ( os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN") or _load_env_token(ROOT / ".env") ) if not token: raise SystemExit("HF token missing; set HF_TOKEN or keep it in ignored .env") try: import httpx from huggingface_hub import HfApi, close_session, set_client_factory, snapshot_download except ImportError as exc: raise SystemExit( "install the exact publishing dependencies: python -m pip install -r " "requirements-publish.txt" ) from exc set_client_factory( _bounded_http_client_factory( httpx, connect_seconds=args.http_connect_timeout_seconds, read_seconds=args.http_read_timeout_seconds, write_seconds=args.http_write_timeout_seconds, pool_seconds=args.http_pool_timeout_seconds, ) ) api = HfApi(token=token) policy = RetryPolicy( max_attempts=args.max_attempts, base_seconds=args.retry_base_seconds, request_timeout_seconds=args.http_read_timeout_seconds, ) try: receipt = _execute_publish( args, release_dir=release_dir, manifest=manifest, development_only=development_only, token=token, api=api, snapshot_download=snapshot_download, policy=policy, ) except PublishFailure as exc: raise SystemExit(str(exc)) from None finally: close_session() print(f"model: {_repo_url(receipt['targets']['model'], 'model')}") print(f"model commit: {receipt['model']['commit_url']}") print(f"space: {_repo_url(receipt['targets']['space'], 'space')}") print(f"space commit: {receipt['space']['commit_url']}") print(f"publication receipt: {_receipt_path(args.receipt, release_dir)}") return 0 if __name__ == "__main__": raise SystemExit(main())