admin.py

"""Admin-tab handlers for the CADGenBench leaderboard Space.

Bundle 7: promote a row into the validated tier (recording the evidence
type) and demote it back. Gating is the ``CADGENBENCH_ADMINS`` Space
variable (comma-separated HF usernames); :func:`is_admin` is the single
predicate the UI uses to enable or disable the controls.

Row writes go through :func:`submit._hub_rmw_results`, so the
``_HUB_LOCK`` + read-modify-write semantics match the submit path
exactly. There is no second writer of ``results.jsonl`` with its own
locking story.
"""
from __future__ import annotations

import logging
import os
from typing import Any, Iterable

import gradio as gr
from huggingface_hub.errors import EntryNotFoundError

from submit import (
    HF_SUBMISSIONS_REPO,
    REPORTS_DIR,
    SUBMISSIONS_DIR,
    _HF_API,
    _hub_rmw_results,
)

logger = logging.getLogger(__name__)

ADMINS_ENV = "CADGENBENCH_ADMINS"

# The evidence types accepted on promotion. Mirrors the
# `validation_method` enum in cadgenbench-submissions/schema.md and the
# validation policy doc.
VALID_METHODS: tuple[str, ...] = ("code", "traces", "api", "manual")


def admin_usernames() -> set[str]:
    """Parse ``CADGENBENCH_ADMINS`` into a set of HF usernames.

    Comma-separated, whitespace-trimmed, empties dropped. Read fresh on
    each call so flipping the Space variable takes effect without a code
    deploy. Empty or unset yields an empty set, which means no one is an
    admin and the controls stay inert.
    """
    raw = os.environ.get(ADMINS_ENV, "")
    return {part.strip() for part in raw.split(",") if part.strip()}


def is_admin(profile: gr.OAuthProfile | None) -> bool:
    """Return whether *profile* is a logged-in user in the admin set.

    Logged-out users (``profile is None``) are never admins. With an
    empty admin set no profile qualifies, so the admin controls remain
    disabled for everyone until ``CADGENBENCH_ADMINS`` is populated.
    """
    if profile is None:
        return False
    return profile.username in admin_usernames()


def _clean_id_set(submission_ids: Iterable[str]) -> set[str]:
    """Normalise an id iterable to a non-empty set, else raise.

    Guards every bulk helper: a no-op call (nothing selected) is a
    caller error, surfaced as ``ValueError`` rather than a silent
    empty write.
    """
    ids = {str(s) for s in submission_ids if s}
    if not ids:
        raise ValueError("No submissions selected.")
    return ids


def promote_rows(submission_ids: Iterable[str], method: str) -> None:
    """Move every listed row into the validated tier with *method*.

    One ``results.jsonl`` write for the whole batch. Idempotent on rows
    already validated (their method is set to *method*).

    Raises:
        ValueError: *method* is unknown, or no ids were given.
        LookupError: one or more ids are absent from ``results.jsonl``
            (no partial write happens; the helper raises inside the
            read-modify-write before the upload).
    """
    if method not in VALID_METHODS:
        raise ValueError(
            f"Unknown validation_method {method!r}; expected one of "
            f"{', '.join(VALID_METHODS)}."
        )
    ids = _clean_id_set(submission_ids)

    def mutate(rows: list[dict[str, Any]]) -> None:
        seen = set()
        for row in rows:
            if row.get("submission_id") in ids:
                row["validation_status"] = "validated"
                row["validation_method"] = method
                seen.add(row["submission_id"])
        _raise_for_missing(ids, seen)

    _hub_rmw_results(
        mutate,
        commit_message=f"promote {len(ids)} row(s) to validated ({method})",
    )


def demote_rows(submission_ids: Iterable[str]) -> None:
    """Return every listed row to the unvalidated tier, clearing method.

    One ``results.jsonl`` write for the whole batch. Idempotent on rows
    already unvalidated.

    Raises:
        ValueError: no ids were given.
        LookupError: one or more ids are absent from ``results.jsonl``.
    """
    ids = _clean_id_set(submission_ids)

    def mutate(rows: list[dict[str, Any]]) -> None:
        seen = set()
        for row in rows:
            if row.get("submission_id") in ids:
                row["validation_status"] = "unvalidated"
                row["validation_method"] = None
                seen.add(row["submission_id"])
        _raise_for_missing(ids, seen)

    _hub_rmw_results(
        mutate,
        commit_message=f"demote {len(ids)} row(s) to unvalidated",
    )


def delete_rows(submission_ids: Iterable[str]) -> None:
    """Permanently delete every listed submission: artifacts then row.

    Irreversible. For each id, best-effort deletes the companion blobs
    (``submissions/<id>.zip``, ``reports/<id>.{html,json}``) and then
    drops the row from ``results.jsonl`` in a single write. A blob that
    does not exist is skipped (a failed / pending row may never have
    had a report). Missing ``results.jsonl`` rows are tolerated too, so
    a re-run after a partial failure still converges.

    Raises:
        ValueError: no ids were given.
    """
    ids = _clean_id_set(submission_ids)

    for sid in sorted(ids):
        for path in (
            f"{SUBMISSIONS_DIR}/{sid}.zip",
            f"{REPORTS_DIR}/{sid}.html",
            f"{REPORTS_DIR}/{sid}.json",
        ):
            try:
                _HF_API.delete_file(
                    path_in_repo=path,
                    repo_id=HF_SUBMISSIONS_REPO,
                    repo_type="dataset",
                    commit_message=f"delete artifact {path}",
                )
            except EntryNotFoundError:
                pass
            except Exception as e:  # noqa: BLE001 - keep deleting the rest
                logger.warning(
                    "Failed to delete artifact %s (%s: %s)",
                    path, type(e).__name__, e,
                )

    def mutate(rows: list[dict[str, Any]]) -> None:
        rows[:] = [r for r in rows if r.get("submission_id") not in ids]

    _hub_rmw_results(
        mutate, commit_message=f"delete {len(ids)} submission(s)",
    )


def _raise_for_missing(requested: set[str], seen: set[str]) -> None:
    """Raise ``LookupError`` if any requested id was not found in the rows."""
    missing = requested - seen
    if missing:
        raise LookupError(
            f"submission_id(s) not in results.jsonl: {', '.join(sorted(missing))}."
        )


def promote_row(submission_id: str, method: str) -> None:
    """Single-row convenience wrapper over :func:`promote_rows`."""
    promote_rows([submission_id], method)


def demote_row(submission_id: str) -> None:
    """Single-row convenience wrapper over :func:`demote_rows`."""
    demote_rows([submission_id])