scripts/publish_hf.py

"""Publish the verified AUREOLE-R release. Default: offline integrity check.

The token stays in process memory; no login, credential file, shell command,
or token argument is used. SDK calls permit an explicit parent-commit guard
and verification of the public, immutable result. No existing file is deleted.
"""
from __future__ import annotations
import argparse
import getpass
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import re
import shutil
import sys
import tempfile
import warnings
from datetime import datetime, timezone

ROOT = Path(__file__).resolve().parents[1]
ENDPOINT = "https://huggingface.co"
DEFAULT_REPO = "PureOne/AUREOLE-R-v3"
EDITION = "3.0.0-hf.1"


class ReleaseError(Exception):
    """Safe-to-display error containing no credentials or HTTP payloads."""


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def safe_name(name: str) -> bool:
    parts = PurePosixPath(name).parts
    return (bool(parts) and not name.startswith("/") and
            not any(part in (".", "..", "") for part in name.split("/")) and
            not any(c in name for c in "\\:\x00\r\n") and
            not any(part.endswith((" ", ".")) for part in parts) and
            not any(part.casefold() in {".env", ".git", ".venv", ".venv-publish"}
                    or part.casefold().endswith(".token") for part in parts) and
            name not in {"publication_receipt.json", "CHECKSUMS.sha256"})


def release_files(root: Path = ROOT) -> dict[str, Path]:
    root = root.resolve()
    manifest = root / "CHECKSUMS.sha256"
    if manifest.is_symlink() or not manifest.is_file():
        raise ReleaseError("CHECKSUMS.sha256 is missing or is a symbolic link.")
    files: dict[str, Path] = {}
    seen = set()
    for line in manifest.read_text(encoding="utf-8").splitlines():
        match = re.fullmatch(r"([0-9a-f]{64})  (.+)", line)
        if not match:
            raise ReleaseError("Malformed release checksum entry.")
        expected, name = match.groups()
        if not safe_name(name) or name.casefold() in seen:
            raise ReleaseError("Unsafe or duplicate release path.")
        seen.add(name.casefold())
        path = root / name
        if (not path.is_file() or not path.resolve().is_relative_to(root) or
                any(p.is_symlink() for p in [path, *path.parents] if p != root)):
            raise ReleaseError("A release file is missing or uses a symbolic link.")
        if sha256(path) != expected:
            raise ReleaseError(f"Checksum mismatch: {name}. Extract a fresh release ZIP.")
        files[name] = path
    if not files:
        raise ReleaseError("The release checksum manifest is empty.")
    files["CHECKSUMS.sha256"] = manifest
    return files


def copy_release(files: dict[str, Path], stage: Path) -> dict[str, Path]:
    for name, source in files.items():
        dest = stage / name
        dest.parent.mkdir(parents=True, exist_ok=True)
        shutil.copyfile(source, dest)
    return release_files(stage)


def remote_matches(sibling, path: Path) -> bool:
    if sibling.size != path.stat().st_size:
        return False
    lfs = getattr(sibling, "lfs", None)
    if lfs is not None:
        digest = lfs.get("sha256") if isinstance(lfs, dict) else lfs.sha256
        return digest == sha256(path)
    payload = path.read_bytes()
    git_blob = hashlib.sha1(b"blob " + str(len(payload)).encode() + b"\0" + payload).hexdigest()
    return getattr(sibling, "blob_id", None) == git_blob


def inspect_remote(info, files: dict[str, Path], *, complete: bool) -> set[str]:
    if info.private or getattr(info, "gated", False) not in (False, None):
        raise ReleaseError("The destination is private or gated; its visibility was not changed.")
    present = set()
    for sibling in info.siblings or []:
        name = sibling.rfilename
        if name == ".gitattributes":  # Server-managed storage configuration.
            continue
        if name not in files:
            raise ReleaseError("The destination contains unrelated files. Choose a new repository with --repo.")
        if not remote_matches(sibling, files[name]):
            raise ReleaseError("The destination contains different release bytes. No overwrite was attempted; choose a new repository with --repo.")
        present.add(name)
    if complete and present != set(files):
        raise ReleaseError("The public commit is missing release files. Rerun this same package to resume.")
    return present


def check_owner(identity: dict, repo: str) -> None:
    if identity.get("name", "").casefold() != repo.split("/", 1)[0].casefold():
        raise ReleaseError("This token belongs to a different account than the repository owner. Use your own account token or --repo YOUR_ACCOUNT/NAME.")


def publish(files, repo, api, public_api, not_found, add_operation, download, work: Path):
    """Explicit dependencies allow testing the complete control flow offline."""
    identity = api.whoami()
    check_owner(identity, repo)
    try:
        before = api.repo_info(repo, repo_type="model", files_metadata=True)
    except not_found:
        api.create_repo(repo_id=repo, repo_type="model", private=False, exist_ok=False)
        before = api.repo_info(repo, repo_type="model", files_metadata=True)
    present = inspect_remote(before, files, complete=False)
    api.auth_check(repo_id=repo, repo_type="model", write=True)
    if present != set(files):
        operations = [add_operation(path_in_repo=name, path_or_fileobj=str(path))
                      for name, path in sorted(files.items()) if name not in present]
        result = api.create_commit(
            repo_id=repo, repo_type="model", operations=operations,
            commit_message=f"AUREOLE-R {EDITION}: standalone public research release",
            commit_description="Verified paper, source, prior, raw evidence, protocols and machine-readable research indexes. Finite CPU reference; full unified SR/RR/FG remains unvalidated.",
            parent_commit=before.sha,
        )
        revision = result.oid
    else:
        revision = before.sha
    public = public_api.repo_info(repo, repo_type="model", revision=revision, files_metadata=True)
    if public.sha != revision:
        raise ReleaseError("The public server did not return the uploaded commit.")
    inspect_remote(public, files, complete=True)
    remote_manifest = Path(download(
        repo_id=repo, filename="CHECKSUMS.sha256", repo_type="model",
        revision=revision, token=False, endpoint=ENDPOINT,
        cache_dir=str(work / "public-verification"), force_download=True,
    ))
    if sha256(remote_manifest) != sha256(files["CHECKSUMS.sha256"]):
        raise ReleaseError("The anonymously downloaded manifest did not match the release.")
    return {
        "schema_version": "1.0", "publication_verified": True,
        "verified_utc": datetime.now(timezone.utc).isoformat(),
        "repo_id": repo, "repo_type": "model", "revision": revision,
        "repository_url": f"{ENDPOINT}/{repo}",
        "immutable_url": f"{ENDPOINT}/{repo}/tree/{revision}",
        "paper_url": f"{ENDPOINT}/{repo}/resolve/{revision}/AUREOLE_R_v3.0.0_Certified_Innovation_Rendering.pdf",
        "scientific_version": "3.0.0", "publication_edition": EDITION,
        "verified_release_files": len(files),
        "checksum_manifest_sha256": sha256(files["CHECKSUMS.sha256"]),
        "verification": "Anonymous commit read, size and Git-blob/LFS-hash verification for every release file, and anonymous SHA-256 manifest download.",
        "scientific_scope": "Finite direct-light CPU reference. No GPU or full SR/RR/FG validation.",
    }


def main(argv=None, root: Path = ROOT) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repo", default=DEFAULT_REPO, help="Personal Hugging Face account/repository")
    parser.add_argument("--publish", action="store_true", help="Create or safely resume this PUBLIC research release")
    args = parser.parse_args(argv)
    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*/[A-Za-z0-9][A-Za-z0-9._-]*", args.repo):
        raise ReleaseError("Repository must be ACCOUNT/NAME.")
    files = release_files(root)
    print(f"PUBLIC destination: {ENDPOINT}/{args.repo}")
    print(f"Verified {len(files)} release files ({sum(p.stat().st_size for p in files.values()):,} bytes).")
    if not args.publish:
        print("Offline check complete. No authentication or network calls. Add --publish to upload.")
        return 0
    os.environ["HF_DEBUG"] = "0"
    os.environ["HF_HUB_VERBOSITY"] = "warning"
    os.environ["HF_HUB_DISABLE_TELEMETRY"] = "1"
    try:
        from huggingface_hub import HfApi, CommitOperationAdd, hf_hub_download
        from huggingface_hub.errors import RepositoryNotFoundError
    except ImportError:
        raise ReleaseError("Install requirements-publish.txt, or use PUBLISH_TO_HUGGINGFACE.bat.") from None
    with tempfile.TemporaryDirectory(prefix="aureole-publication-") as temp:
        work = Path(temp)
        staged = copy_release(files, work / "release")
        with warnings.catch_warnings():
            warnings.simplefilter("error", getpass.GetPassWarning)
            token = os.environ.get("HF_TOKEN") or getpass.getpass("Hugging Face write token (paste; input is hidden): ")
        token = token.strip()
        if not token:
            raise ReleaseError("No token supplied. Nothing was uploaded.")
        print("Checking account and public destination, then uploading the verified release...")
        receipt = publish(
            staged, args.repo, HfApi(endpoint=ENDPOINT, token=token),
            HfApi(endpoint=ENDPOINT, token=False), RepositoryNotFoundError,
            CommitOperationAdd, hf_hub_download, work,
        )
        del token
    receipt_path = root / "publication_receipt.json"
    receipt_path.write_text(json.dumps(receipt, indent=2) + "\n", encoding="utf-8")
    print("PUBLICATION VERIFIED: " + receipt["repository_url"])
    print("Immutable research release: " + receipt["immutable_url"])
    print("Receipt saved to publication_receipt.json (contains no token).")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (KeyboardInterrupt, EOFError, getpass.GetPassWarning):
        print("Cancelled, or secure token input unavailable. Use the Windows launcher in a terminal.", file=sys.stderr)
        raise SystemExit(1)
    except ReleaseError as exc:
        print(str(exc), file=sys.stderr)
        raise SystemExit(1)
    except Exception as exc:
        status = getattr(getattr(exc, "response", None), "status_code", None)
        suffix = f" HTTP status: {status}." if isinstance(status, int) else ""
        print("Publication was not verified." + suffix + " Check your connection and write-token permissions, then rerun the same package. A remote commit may already exist; matching bytes are safe to resume. No exception payload or token was logged.", file=sys.stderr)
        raise SystemExit(1)