app.py

"""RelBench leaderboard submission API (Hugging Face Docker Space).

A plain FastAPI service — no Gradio. The submission form lives on tabular.stanford.edu and
POSTs here. On each submission this service:

  1. unzips the upload and runs the official validator (`relbench.leaderboard`),
  2. if at least one leaderboard family is validated:
       - commits the raw submission (zip + metadata + report) to the PRIVATE submissions
         dataset as an append-only audit record, and
       - opens a PR on the PUBLIC leaderboard dataset adding one `entries/<slug>.json`,
  3. returns the validation report so the form can show the verdict.

Maintainers are notified by Hugging Face natively (they watch the leaderboard repo) and
merge the PR to publish. The website reads entries straight from the public repo, so a
merge goes live with no redeploy.

Config (Space secrets / variables):
  HF_TOKEN          write token for both datasets (required to write; omit for dry-run)
  LEADERBOARD_REPO  default stanford-star/relbench-leaderboard   (public)
  SUBMISSIONS_REPO  default stanford-star/relbench-submissions    (private)
  ALLOW_ORIGINS     comma-separated CORS origins (default https://tabular.stanford.edu)
  EVAL_WORKERS      validator process-pool size (default 1 — lowest memory)
  MAX_UPLOAD_MB     reject uploads larger than this (default 200)
"""
from __future__ import annotations

import io
import json
import os
import re
import tempfile
import uuid
import zipfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional

from fastapi import FastAPI, File, HTTPException, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse

# --------------------------------------------------------------------------- #
# Config
# --------------------------------------------------------------------------- #
HF_TOKEN = os.getenv("HF_TOKEN") or None
LEADERBOARD_REPO = os.getenv("LEADERBOARD_REPO", "stanford-star/relbench-leaderboard")
SUBMISSIONS_REPO = os.getenv("SUBMISSIONS_REPO", "stanford-star/relbench-submissions")
ALLOW_ORIGINS = [o.strip() for o in
                 os.getenv("ALLOW_ORIGINS", "https://tabular.stanford.edu").split(",") if o.strip()]
EVAL_WORKERS = int(os.getenv("EVAL_WORKERS", "1"))
MAX_UPLOAD_MB = int(os.getenv("MAX_UPLOAD_MB", "200"))

# Validator family name -> the site's board key (assets/js/leaderboard.js / boards.json).
SITE_BOARD = {
    "classification": "binary_classification",
    "regression": "regression",
    "recommendation": "link_prediction",
}

# Public leaderboard-row fields, copied verbatim from the submission's metadata.yaml.
# `url` links the name; `type` drives the In-context/Fine-tuned sub-tabs. The submission
# `date` is stamped server-side (see build_entry); `email` is recorded privately, never here.
ENTRY_FIELDS = ["name", "type", "url", "note"]

app = FastAPI(title="RelBench Leaderboard Submission")
app.add_middleware(
    CORSMiddleware, allow_origins=ALLOW_ORIGINS, allow_methods=["POST", "GET"],
    allow_headers=["*"],
)


def slug(s: str) -> str:
    s = s.lower().replace("+", " plus ")
    s = re.sub(r"[^a-z0-9]+", "-", s).strip("-")
    return re.sub(r"-+", "-", s) or "entry"


def find_pred_dir(root: Path) -> Path:
    """Locate the directory that holds the prediction CSVs (zip may wrap them in a folder)."""
    best, best_n = root, -1
    for d, _, files in os.walk(root):
        n = sum(1 for f in files if f.endswith(".csv"))
        if n > best_n:
            best, best_n = Path(d), n
    if best_n <= 0:
        raise HTTPException(400, "No prediction CSVs (*.csv) found in the upload.")
    return best


def build_entry(mf: Dict[str, Any], result: Dict[str, Any]) -> Dict[str, Any]:
    """Assemble the per-entry JSON in the site's schema from the metadata fields + report."""
    entry: Dict[str, Any] = {f: mf.get(f) or None for f in ENTRY_FIELDS}
    entry["date"] = datetime.now(timezone.utc).strftime("%Y-%m")   # submission date, server-stamped
    boards: Dict[str, Any] = {}
    for fam in result["validated"]:
        fam_res = result["families"][fam]
        boards[SITE_BOARD[fam]] = {
            "results": {t: result["tasks"][t]["metric"] for t in fam_res["valid"]},
            "mean": fam_res["aggregate"],
            "cov": fam_res["num_valid"],
        }
    entry["boards"] = boards
    return entry


def report_markdown(mf: Dict[str, Any], result: Dict[str, Any],
                    metadata: Dict[str, Any]) -> str:
    lines = [f"### Submission: {mf.get('name', '(no name)')}", ""]
    for fam, f in result["families"].items():
        agg = "–" if f["aggregate"] is None else f"{f['aggregate']:.4f}"
        lines.append(f"- **{fam}** — {f['verdict']} · mean {f['metric_name']} = {agg}")
    if mf.get("url"):
        lines.append(f"\nURL: {mf['url']}")
    for w in metadata.get("warnings", []):
        lines.append(f"\n> metadata warning: {w}")
    return "\n".join(lines)


@app.get("/")
def health() -> Dict[str, Any]:
    return {"ok": True, "leaderboard_repo": LEADERBOARD_REPO,
            "can_write": bool(HF_TOKEN), "eval_workers": EVAL_WORKERS}


@app.post("/submit")
async def submit(file: UploadFile = File(...)) -> JSONResponse:
    """Validate an uploaded submission zip (CSVs + metadata.yaml) and, on success, open a PR."""
    if not file.filename or not file.filename.lower().endswith(".zip"):
        raise HTTPException(400, "Upload must be a .zip of the submission directory.")
    raw = await file.read()
    if len(raw) > MAX_UPLOAD_MB * 1024 * 1024:
        raise HTTPException(413, f"Upload exceeds {MAX_UPLOAD_MB} MB.")

    # Late import: relbench (+ torch/pandas) is heavy; keep health checks instant and surface
    # an import failure as a clean 500 rather than a worker crash at boot.
    from relbench.leaderboard import evaluate_submission

    with tempfile.TemporaryDirectory() as tmp:
        tmp = Path(tmp)
        try:
            with zipfile.ZipFile(io.BytesIO(raw)) as zf:
                zf.extractall(tmp / "unzipped")
        except zipfile.BadZipFile:
            raise HTTPException(400, "Upload is not a valid zip file.")

        pred_dir = find_pred_dir(tmp / "unzipped")
        try:
            # Reads predictions AND metadata.yaml from the submission dir.
            result = evaluate_submission(pred_dir, num_workers=EVAL_WORKERS, verbose=False)
        except Exception as e:  # validation/scoring error -> actionable message for the form
            raise HTTPException(422, f"Validation failed: {e}")

    metadata = result.get("metadata") or {"fields": {}, "errors": ["metadata not parsed"],
                                           "warnings": []}
    mf: Dict[str, Any] = metadata["fields"]
    validated: List[str] = result["validated"]
    report_md = report_markdown(mf, result, metadata)

    # Gate: a clean directory (only CSVs + metadata.yaml), at least one validated leaderboard,
    # and clean metadata. The local CLI (--submit/--package) strips extra files before upload;
    # a zip that still contains them is rejected here.
    problems: List[str] = []
    if result.get("extra_files"):
        problems.append("submission contains files other than prediction CSVs + "
                        "metadata.yaml: " + ", ".join(result["extra_files"]))
    if not validated:
        problems.append("no leaderboard was validated")
    if metadata["errors"]:
        problems.append("metadata.yaml — " + "; ".join(metadata["errors"]))
    if problems:
        return JSONResponse(status_code=200, content={
            "status": "rejected", "validated": validated, "report": result, "report_md": report_md,
            "message": "Not submitted — " + "; ".join(problems)
                       + ". Fix the issues below and resubmit.",
        })

    entry = build_entry(mf, result)
    if not HF_TOKEN:
        # Dry-run (no token configured): show what *would* be submitted without writing.
        return JSONResponse(status_code=200, content={
            "status": "dry_run", "validated": validated, "report": result,
            "report_md": report_md, "entry": entry,
            "message": "Validation passed. (Server has no HF_TOKEN, so nothing was written.)",
        })

    from huggingface_hub import CommitOperationAdd, HfApi
    api = HfApi(token=HF_TOKEN)
    sid = uuid.uuid4().hex[:8]
    base = f"{slug(entry['name'])}-{sid}"
    contact = mf.get("email", "")

    # 1. Raw audit record -> PRIVATE submissions repo (direct commit, append-only).
    record = {**mf, "validated": validated, "date": entry["date"]}
    api.create_commit(
        repo_id=SUBMISSIONS_REPO, repo_type="dataset",
        operations=[
            CommitOperationAdd(f"submissions/{base}/submission.zip", raw),
            CommitOperationAdd(f"submissions/{base}/metadata.json",
                               json.dumps(record, indent=2).encode()),
            CommitOperationAdd(f"submissions/{base}/report.json",
                               json.dumps(result, indent=2).encode()),
        ],
        commit_message=f"Submission: {entry['name']} ({', '.join(validated)})",
    )

    # 2. Proposed leaderboard row -> PR on the PUBLIC leaderboard repo for maintainer review.
    pr = api.create_commit(
        repo_id=LEADERBOARD_REPO, repo_type="dataset",
        operations=[CommitOperationAdd(
            f"entries/{base}.json", json.dumps(entry, indent=2).encode() + b"\n")],
        commit_message=f"Add leaderboard entry: {entry['name']}",
        commit_description=report_md + (f"\n\nContact: {contact}" if contact else ""),
        create_pr=True,
    )
    pr_url = getattr(pr, "pr_url", None)

    return JSONResponse(status_code=200, content={
        "status": "pending_review", "validated": validated, "report": result,
        "report_md": report_md, "pr_url": pr_url,
        "message": "Validation passed. Your submission is now a pull request awaiting "
                   "maintainer approval; it appears on the leaderboard once merged.",
    })