src/submission/commit_results.py

import json
import math
import os
import tempfile
from threading import Lock

from huggingface_hub.errors import HfHubHTTPError

from src.display.formatting import styled_error, styled_message
from src.envs import API, LOCAL_DEV, REPO_ID, TOKEN

DIMENSION_FIELDS = ["Layout", "Attribute", "Text", "Knowledge"]
DOMAIN_FIELDS = ["Slides", "Webpage", "Poster", "Chart", "Scientific Figure"]
SCORE_GROUPS = [
    ("Capability Dimensions", "by_dimension", "AverageByDimension", DIMENSION_FIELDS),
    ("Content Domains", "by_domain", "AverageByDomain", DOMAIN_FIELDS),
]

SCORE_FIELD_ORDER = []
for _, group_key, _, labels in SCORE_GROUPS:
    for label in labels:
        SCORE_FIELD_ORDER.append((group_key, label, "hard"))
        SCORE_FIELD_ORDER.append((group_key, label, "easy"))

_COMMIT_RESULTS_LOCK = Lock()


def _load_commit_results(commit_results_path: str) -> list[dict]:
    if not os.path.exists(commit_results_path):
        return []

    entries = []
    with open(commit_results_path, encoding="utf-8") as fp:
        for line in fp:
            line = line.strip()
            if not line:
                continue
            entries.append(json.loads(line))
    return entries


def _write_commit_results(commit_results_path: str, entries: list[dict]) -> None:
    with open(commit_results_path, "w", encoding="utf-8") as fp:
        for entry in entries:
            fp.write(json.dumps(entry, ensure_ascii=False))
            fp.write("\n")


def _coerce_score(raw_value, label: str) -> float:
    if raw_value in (None, ""):
        raise ValueError(f"Please provide a score for {label}.")

    try:
        value = float(raw_value)
    except (TypeError, ValueError) as exc:
        raise ValueError(f"{label} must be a number.") from exc

    if not math.isfinite(value):
        raise ValueError(f"{label} must be a finite number.")
    if value < 0 or value > 100:
        raise ValueError(f"{label} must be between 0 and 100.")

    return round(value, 1)


def _average_score(values: list[float]) -> float:
    return round(sum(values) / len(values), 1)


def build_commit_result_entry(model_name: str, *score_values) -> dict:
    model_name = (model_name or "").strip()
    if not model_name:
        raise ValueError("Please provide a model name.")

    if len(score_values) != len(SCORE_FIELD_ORDER):
        raise ValueError("The submission form is incomplete. Please refresh the page and try again.")

    by_dimension = {label: {} for label in DIMENSION_FIELDS}
    by_domain = {label: {} for label in DOMAIN_FIELDS}
    dimension_hard = []
    dimension_easy = []
    domain_hard = []
    domain_easy = []

    for (group_key, label, track), raw_value in zip(SCORE_FIELD_ORDER, score_values):
        display_label = f"{label} {track}"
        score_value = _coerce_score(raw_value, display_label)

        if group_key == "by_dimension":
            by_dimension[label][track] = score_value
            if track == "hard":
                dimension_hard.append(score_value)
            else:
                dimension_easy.append(score_value)
        else:
            by_domain[label][track] = score_value
            if track == "hard":
                domain_hard.append(score_value)
            else:
                domain_easy.append(score_value)

    return {
        "Model": model_name,
        "SourceType": "user-submitted",
        "PaperLink": "",
        "AverageByDomain": {
            "hard": _average_score(dimension_hard + domain_hard),
            "easy": _average_score(dimension_easy + domain_easy),
        },
        "AverageByDimension": {
            "hard": _average_score(dimension_hard + domain_hard),
            "easy": _average_score(dimension_easy + domain_easy),
        },
        "by_domain": by_domain,
        "by_dimension": by_dimension,
    }


def submit_commit_result(commit_results_path: str, model_name: str, *score_values) -> str:
    try:
        new_entry = build_commit_result_entry(model_name, *score_values)
    except ValueError as exc:
        return styled_error(str(exc))

    if not LOCAL_DEV and not TOKEN:
        return styled_error(
            "This Space is missing a valid HF_TOKEN secret, so it cannot append to commit_results.jsonl."
        )

    with _COMMIT_RESULTS_LOCK:
        existing_entries = _load_commit_results(commit_results_path)
        updated_entries = [*existing_entries, new_entry]

        tmp_fd, tmp_path = tempfile.mkstemp(
            prefix="commit-results-",
            suffix=".jsonl",
            dir=os.path.dirname(commit_results_path) or ".",
        )
        os.close(tmp_fd)

        try:
            _write_commit_results(tmp_path, updated_entries)

            if LOCAL_DEV:
                os.replace(tmp_path, commit_results_path)
                return styled_message(
                    f'Saved "{new_entry["Model"]}" locally. The leaderboard file now includes this submission.'
                )

            API.upload_file(
                path_or_fileobj=tmp_path,
                path_in_repo=os.path.basename(commit_results_path),
                repo_id=REPO_ID,
                repo_type="space",
                commit_message=f'Add BizGenEval result for {new_entry["Model"]}',
            )
            os.replace(tmp_path, commit_results_path)
            return styled_message(
                f'Submitted "{new_entry["Model"]}". The Space will rebuild and the public leaderboard will refresh shortly.'
            )
        except Exception as exc:
            if os.path.exists(tmp_path):
                os.remove(tmp_path)

            if isinstance(exc, HfHubHTTPError) and exc.response is not None and exc.response.status_code in {401, 403}:
                return styled_error(
                    "The Space could not write to commit_results.jsonl because HF_TOKEN is missing, invalid, or lacks"
                    " write access to microsoft/BizGenEval-Leaderboard."
                )

            return styled_error(f"Could not update commit_results.jsonl: {exc}")