app/fetch.py

"""Pull the last 7 daily `run_models_gpu` reports from the CI dataset.

The CI dataset is `hf-internal-testing/transformers_daily_ci`. Each day has:
  YYYY-MM-DD/
    ci_results_run_models_gpu/
      model_results.json
      new_failures_with_bad_commit_grouped_by_authors.json
"""

from __future__ import annotations

import datetime
import json
import os
from pathlib import Path

from huggingface_hub import HfApi, hf_hub_download
from huggingface_hub.utils import EntryNotFoundError


CI_DATASET = "hf-internal-testing/transformers_daily_ci"
JOB_DIR = "ci_results_run_models_gpu"
MODEL_RESULTS = "model_results.json"
NEW_FAILURES = "new_failures_with_bad_commit_grouped_by_authors.json"


def list_recent_dates(api: HfApi, n: int = 7) -> list[str]:
    """Top-level dirs under the dataset look like YYYY-MM-DD. Return the n most recent."""
    files = api.list_repo_files(repo_id=CI_DATASET, repo_type="dataset")
    dates = set()
    for f in files:
        head = f.split("/", 1)[0]
        try:
            datetime.date.fromisoformat(head)
        except ValueError:
            continue
        dates.add(head)
    return sorted(dates, reverse=True)[:n]


def fetch_day(date: str, cache_dir: str | Path | None = None) -> dict[str, dict | None]:
    """Download both JSONs for a given day; missing files return None instead of raising."""
    out: dict[str, dict | None] = {}
    for label, fname in (("model_results", MODEL_RESULTS), ("new_failures", NEW_FAILURES)):
        try:
            path = hf_hub_download(
                repo_id=CI_DATASET,
                repo_type="dataset",
                filename=f"{date}/{JOB_DIR}/{fname}",
                cache_dir=str(cache_dir) if cache_dir else None,
            )
            with open(path) as f:
                out[label] = json.load(f)
        except EntryNotFoundError:
            out[label] = None
    return out


def fetch_last_n(n: int = 7, cache_dir: str | Path | None = None) -> dict[str, dict[str, dict | None]]:
    api = HfApi(token=os.environ.get("HF_TOKEN"))
    dates = list_recent_dates(api, n)
    return {d: fetch_day(d, cache_dir=cache_dir) for d in dates}


if __name__ == "__main__":
    import sys

    cache = sys.argv[1] if len(sys.argv) > 1 else None
    data = fetch_last_n(7, cache_dir=cache)
    for d, payload in data.items():
        mr = payload.get("model_results")
        nf = payload.get("new_failures")
        print(
            f"{d}: model_results={'OK' if mr else 'MISSING'} "
            f"({len(mr) if mr else 0} models), "
            f"new_failures={'OK' if nf else 'MISSING'} "
            f"({sum(len(v) for v in nf.values()) if nf else 0} authors)"
        )