src/hf_sync.py

from __future__ import annotations

import os
import shutil
import sys
from pathlib import Path

from .config import HF_EVENTS_FILE, HF_REPO_ID, events_path, load_env

load_env()

ASSETS_EVENTS_PATH = events_path()


def hf_repo_id() -> str:
    return HF_REPO_ID


def hf_token() -> str | None:
    return (os.environ.get("HF_TOKEN") or os.environ.get("HUGGINGFACE_HUB_TOKEN") or "").strip() or None

def hf_enabled() -> bool:
    return bool(hf_token())


def pull_events(path: str | Path = ASSETS_EVENTS_PATH) -> None:
    from huggingface_hub import hf_hub_download

    downloaded = hf_hub_download(
        repo_id=hf_repo_id(),
        filename=HF_EVENTS_FILE,
        repo_type="dataset",
        token=hf_token(),
    )
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(downloaded, path)


def push_events(path: str | Path = ASSETS_EVENTS_PATH) -> None:
    from huggingface_hub import HfApi

    HfApi(token=hf_token()).upload_file(
        path_or_fileobj=str(path),
        path_in_repo=HF_EVENTS_FILE,
        repo_id=hf_repo_id(),
        repo_type="dataset",
        commit_message="Update robotic seminars events",
    )


if __name__ == "__main__":
    command = sys.argv[1] if len(sys.argv) > 1 else ""
    if command == "pull":
        pull_events()
    if command == "push":
        push_events()