source/scripts/publishing/hub.py

"""Upload an inventoried release to its existing Hugging Face model repository."""

import argparse
import importlib.util
import json
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--release", type=Path, required=True)
    parser.add_argument("--report", type=Path, required=True)
    args = parser.parse_args()
    if args.report.exists():
        parser.error("Upload receipt exists; inspect it before repeating an upload.")
    spec = importlib.util.spec_from_file_location("release_verifier", args.release / "verify_release.py")
    verifier = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(verifier)
    manifest, files = verifier.verify(args.release, hashes=False)
    from huggingface_hub import CommitOperationAdd, HfApi
    api = HfApi()
    before = api.model_info(manifest["repo_id"], revision="main", files_metadata=True)
    unexpected = {f.rfilename for f in before.siblings} - set(files) - {".gitattributes"}
    if unexpected:
        raise ValueError(f"Target contains unrelated files: {sorted(unexpected)}")
    result = api.create_commit(
        repo_id=manifest["repo_id"], repo_type="model", revision="main", parent_commit=before.sha,
        operations=[CommitOperationAdd(path_in_repo=name, path_or_fileobj=str(args.release / name)) for name in files],
        commit_message="Release VimeML V2.1 step40000 FP32 and Core ML INT8 (GPL-2.0)",
    )
    receipt = {"repo_id": manifest["repo_id"], "hub_commit": result.oid,
               "commit_url": result.commit_url, "uploaded_files": len(files),
               "source_commit": manifest["source_commit"]}
    args.report.parent.mkdir(parents=True, exist_ok=True)
    args.report.write_text(json.dumps(receipt, indent=2) + "\n", encoding="utf-8", newline="\n")
    after = api.model_info(manifest["repo_id"], revision=result.oid, files_metadata=True)
    actual = {f.rfilename: f.size for f in after.siblings if f.rfilename != ".gitattributes"}
    expected = {name: (args.release / name).stat().st_size for name in files}
    if actual != expected:
        raise ValueError("Committed remote file inventory differs; preserve upload receipt for diagnosis.")
    receipt["remote_inventory"] = "passed"
    args.report.write_text(json.dumps(receipt, indent=2) + "\n", encoding="utf-8", newline="\n")
    print(json.dumps(receipt, indent=2))


if __name__ == "__main__":
    main()