"""Hugging Face Hub sync for code pull and checkpoint push (Kaggle-friendly).""" import os import subprocess from pathlib import Path from typing import Optional def clone_or_pull_repo( repo_id: str, local_dir: str, token: Optional[str] = None, revision: Optional[str] = None, ) -> str: """ Clone HF repo via git if missing; otherwise git pull. Requires `git` and `git-lfs` on the machine (Kaggle has both). """ token = token or os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN") local = Path(local_dir) local.parent.mkdir(parents=True, exist_ok=True) if local.is_dir() and (local / ".git").is_dir(): subprocess.run(["git", "-C", str(local), "pull"], check=False) return str(local) clone_url = f"https://huggingface.co/{repo_id}" if token: clone_url = f"https://user:{token}@huggingface.co/{repo_id}" cmd = ["git", "clone", clone_url, str(local)] if revision: cmd.extend(["--branch", revision]) subprocess.run(cmd, check=True) return str(local) def snapshot_download_repo(repo_id: str, local_dir: str, token: Optional[str] = None) -> str: """Download repo snapshot without git (good for read-only code pull).""" from huggingface_hub import snapshot_download token = token or os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN") path = snapshot_download( repo_id=repo_id, local_dir=local_dir, token=token, local_dir_use_symlinks=False, ) return path def push_folder_to_hub( folder_path: str, repo_id: str, path_in_repo: str = "runs", token: Optional[str] = None, commit_message: str = "Upload training artifacts", ) -> str: """Upload a local folder (checkpoints, metrics) to an HF model/dataset repo.""" from huggingface_hub import HfApi token = token or os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN") api = HfApi(token=token) api.upload_folder( folder_path=folder_path, path_in_repo=path_in_repo, repo_id=repo_id, repo_type="model", commit_message=commit_message, ) return f"https://huggingface.co/{repo_id}/tree/main/{path_in_repo}"