utils.py

from __future__ import annotations

import json
import os
import re
from datetime import datetime, timezone
from html import escape
from inspect import signature
from typing import Any

import pandas as pd
from cryptography.fernet import Fernet
from gradio_client import Client
from huggingface_hub import HfApi, hf_hub_download
from huggingface_hub.errors import EntryNotFoundError, RepositoryNotFoundError

API = HfApi()
HF_TOKEN = ((os.environ.get("aitx_submissions") or os.environ.get("HF_TOKEN")) or "").strip() or None
SPACE_TOKEN_ENCRYPTION_KEY = (os.environ.get("AITX_SPACE_TOKEN_ENCRYPTION_KEY") or "").strip()


SUBMISSIONS_REPO = "aitxchallenge/aitxchallenge-submissions"
RESULTS_REPO = "aitxchallenge/aitxchallenge-results"
RESULTS_FILE = "results.jsonl"
SUBMISSION_PREFIX = "submissions/submission_"
SAMPLE_QUESTIONS_REPO = "aitxchallenge/Phase1_Model_Validator"
SAMPLE_QUESTIONS_FILE = "sample_questions.json"
PARTICIPANT_API_NAME = "/answer_question"
SPACE_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*/[A-Za-z0-9][A-Za-z0-9._-]*$")
LEADERBOARD_COLUMNS = [
    "Rank",
    "Model",
    "Model Tier",
    "Model Source",
    "Submitter",
    "Overall Score",
    "Submitted At",
]

SUBMISSION_TIERS = {
    "Tier 1: Small": (
        "Models that can run efficiently on low-resource environments, typically "
        "with <=8 billion parameters. All model weights must be open-source."
    ),
    "Tier 2: Large": (
        "Larger models with <=70 billion parameters. All model weights must be "
        "open-source."
    ),
    "Tier 3: Unrestricted": (
        "Any model, including those with >70 billion parameters, closed-source "
        "models, or systems relying on private/commercial APIs."
    ),
}

MODEL_SOURCE_TYPES = {
    "Open source": "Model weights are publicly available under an open-source license.",
    "Closed source": "Model weights are not publicly available or the system relies on private/commercial APIs.",
}


def submit_prediction(
    model_name: str,
    submission_tier: str,
    model_source_type: str,
    space_id: str,
    hf_token: str,
    submitted_by: str = "anonymous",
    submitted_by_hf_username: str = "",
    submitted_by_hf_name: str = "",
    submitted_by_hf_profile: str = "",
    no_logging_attestation: bool = False,
    private_evaluation_cost_acknowledgement: bool = False,
) -> str:
    """Upload a Space-based submission metadata record to the submissions dataset."""
    cleaned_model_name = (model_name or "").strip()
    cleaned_submission_tier = (submission_tier or "").strip()
    cleaned_model_source_type = (model_source_type or "").strip()
    cleaned_space_id = (space_id or "").strip()
    cleaned_hf_token = (hf_token or "").strip()
    cleaned_submitted_by = (submitted_by or "").strip()
    cleaned_hf_username = (submitted_by_hf_username or "").strip()
    cleaned_hf_name = (submitted_by_hf_name or "").strip()
    cleaned_hf_profile = (submitted_by_hf_profile or "").strip()

    if not cleaned_model_name:
        raise ValueError("Model name is required.")
    if cleaned_submission_tier not in SUBMISSION_TIERS:
        raise ValueError("Submission tier is required.")
    if cleaned_model_source_type not in MODEL_SOURCE_TYPES:
        raise ValueError("Open source vs. closed source model selection is required.")
    if not cleaned_submitted_by:
        raise ValueError("Team or submitter name is required.")
    if not cleaned_space_id:
        raise ValueError("Space ID is required.")
    if not SPACE_ID_PATTERN.match(cleaned_space_id):
        raise ValueError("Space ID must look like `owner/space-name`.")
    if not cleaned_hf_token:
        raise ValueError("Read token for the submitted Space is required.")
    if not no_logging_attestation:
        raise ValueError(
            "You must agree not to log, store, redistribute, or train on evaluation questions. "
            "Failure to comply will result in expulsion from the competition."
        )
    if not private_evaluation_cost_acknowledgement:
        raise ValueError(
            "You must authorize the organizers to run the private dataset on your evaluation Space "
            "and acknowledge that your Space may incur evaluation costs."
        )

    duplicate_warning = _build_duplicate_submission_warning(cleaned_model_name, cleaned_hf_username)

    timestamp = datetime.now(timezone.utc).replace(microsecond=0).isoformat()
    safe_model_name = re.sub(r"[^A-Za-z0-9._-]+", "_", cleaned_model_name).strip("_") or "submission"
    submission_id = f"{safe_model_name}_{timestamp}".replace(":", "-")

    submission = {
        "submission_id": submission_id,
        "model_name": cleaned_model_name,
        "submission_tier": cleaned_submission_tier,
        "submission_tier_description": SUBMISSION_TIERS[cleaned_submission_tier],
        "model_source_type": cleaned_model_source_type,
        "model_source_type_description": MODEL_SOURCE_TYPES[cleaned_model_source_type],
        "space_id": cleaned_space_id,
        "space_token_encrypted": _encrypt_space_token(cleaned_hf_token),
        "submitted_by": cleaned_submitted_by,
        "submitted_by_hf_username": cleaned_hf_username,
        "submitted_by_hf_name": cleaned_hf_name,
        "submitted_by_hf_profile": cleaned_hf_profile,
        "submission_time": timestamp,
        "no_logging_attestation": True,
        "no_logging_attestation_text": (
            "Submitter agrees not to log, store, redistribute, or train on private "
            "evaluation questions or answers sent to the submitted Space. Failure to "
            "comply will result in expulsion from the competition."
        ),
        "private_evaluation_cost_acknowledgement": True,
        "private_evaluation_cost_acknowledgement_text": (
            "Submitter authorizes the organizers to run the private dataset on the "
            "submitted evaluation Space and acknowledges that the submitted Space may "
            "incur evaluation costs."
        ),
        "status": "pending",
    }

    if not HF_TOKEN:
        raise RuntimeError("Set `aitx_submissions` or `HF_TOKEN` to write submissions.")

    metadata_bytes = (json.dumps(submission, indent=2) + "\n").encode("utf-8")

    API.upload_file(
        path_or_fileobj=metadata_bytes,
        path_in_repo=f"{SUBMISSION_PREFIX}{submission_id}.json",
        repo_id=SUBMISSIONS_REPO,
        repo_type="dataset",
        token=HF_TOKEN,
        commit_message=f"Add submission {submission_id}",
    )

    return _format_submission_confirmation(
        submission_id=submission_id,
        model_name=cleaned_model_name,
        space_id=cleaned_space_id,
        submitted_by=cleaned_submitted_by,
        submitted_by_hf_username=cleaned_hf_username,
        duplicate_warning=duplicate_warning,
    )


def check_submission_format(space_id: str, hf_token: str) -> str:
    """Call a submitted Space with a public sample question and validate the response schema."""
    cleaned_space_id = (space_id or "").strip()
    if not cleaned_space_id:
        raise ValueError("Space ID is required.")
    if not SPACE_ID_PATTERN.match(cleaned_space_id):
        raise ValueError("Space ID must look like `owner/space-name`.")

    sample_payload, expected_answer = _load_sample_question()
    client = _make_space_client(cleaned_space_id, hf_token)
    raw_response = client.predict(sample_payload, api_name=PARTICIPANT_API_NAME)
    try:
        response = _validate_api_format_response(raw_response, expected_id=sample_payload["id"])
    except ValueError as exc:
        return _format_api_check_report(
            cleaned_space_id=cleaned_space_id,
            sample_payload=sample_payload,
            expected_answer=expected_answer,
            raw_response=raw_response,
            validation_error=str(exc),
        )

    return _format_api_check_report(
        cleaned_space_id=cleaned_space_id,
        sample_payload=sample_payload,
        expected_answer=expected_answer,
        raw_response=response,
        validation_error="",
    )


def load_results() -> pd.DataFrame:
    """Load evaluation results for the public leaderboard."""
    try:
        results_path = hf_hub_download(
            repo_id=RESULTS_REPO,
            filename=RESULTS_FILE,
            repo_type="dataset",
            token=HF_TOKEN,
            force_download=True,
        )
    except (EntryNotFoundError, RepositoryNotFoundError):
        return pd.DataFrame(columns=LEADERBOARD_COLUMNS)

    records = []
    with open(results_path, "r", encoding="utf-8") as handle:
        for line in handle:
            stripped = line.strip()
            if stripped:
                records.append(json.loads(stripped))

    records = _backfill_result_submission_fields(records)
    return _build_leaderboard_dataframe(records)


def _build_leaderboard_dataframe(records: list[dict[str, Any]]) -> pd.DataFrame:
    """Build a public, presentation-ready leaderboard from raw evaluator records."""
    df = pd.DataFrame(records)
    if df.empty:
        return pd.DataFrame(columns=LEADERBOARD_COLUMNS)
    if "overall_score" not in df.columns:
        df["overall_score"] = pd.NA
    if "status" in df.columns:
        df = df[df["status"].fillna("completed") == "completed"].copy()
    if df.empty:
        return pd.DataFrame(columns=LEADERBOARD_COLUMNS)

    if "submitted_by_hf_username" not in df.columns:
        df["submitted_by_hf_username"] = ""
    else:
        df["submitted_by_hf_username"] = df["submitted_by_hf_username"].fillna("")
    df["overall_score_sort"] = pd.to_numeric(df["overall_score"], errors="coerce").fillna(-1.0)
    df["submission_time_sort"] = pd.to_datetime(df.get("submission_time"), errors="coerce", utc=True)

    oauth_rows = df["submitted_by_hf_username"].astype(str).str.strip() != ""
    if oauth_rows.any():
        deduped_oauth = (
            df[oauth_rows]
            .sort_values(by=["submission_time_sort", "overall_score_sort"], ascending=[False, False], na_position="last")
            .drop_duplicates(subset=["submitted_by_hf_username"], keep="first")
        )
        df = pd.concat([deduped_oauth, df[~oauth_rows]], ignore_index=True)

    df = df.sort_values(
        by=["overall_score_sort", "submission_time_sort"],
        ascending=[False, True],
        na_position="last",
    ).reset_index(drop=True)

    display_df = pd.DataFrame()
    display_df["Rank"] = range(1, len(df) + 1)
    display_df["Model"] = _display_column(df, "model_name")
    display_df["Model Tier"] = _display_column(df, "submission_tier")
    display_df["Model Source"] = _display_column(df, "model_source_type")
    display_df["Submitter"] = [_format_submitter(row) for _, row in df.iterrows()]
    display_df["Overall Score"] = df["overall_score"].map(_format_score)

    for category in _category_score_names(df):
        column_name = f"{_humanize_label(category)} Score"
        display_df[column_name] = [
            _format_score(_get_category_score(row, category)) for _, row in df.iterrows()
        ]

    display_df["Submitted At"] = [
        _format_timestamp(value) for value in df.get("submission_time", pd.Series(dtype="object"))
    ]

    return display_df


def resolve_oauth_submitter(oauth_profile: Any) -> tuple[str, str, str]:
    """Extract the public HF identity fields from a Gradio OAuth profile."""
    if oauth_profile is None:
        raise ValueError("Please sign in with Hugging Face before submitting.")

    username = (getattr(oauth_profile, "username", "") or oauth_profile.get("preferred_username", "")).strip()
    name = (getattr(oauth_profile, "name", "") or oauth_profile.get("name", "")).strip()
    profile_url = (getattr(oauth_profile, "profile", "") or oauth_profile.get("profile", "")).strip()

    if not username:
        raise ValueError("Please sign in with Hugging Face before submitting.")

    return username, name, profile_url


def _format_submission_confirmation(
    submission_id: str,
    model_name: str,
    space_id: str,
    submitted_by: str,
    submitted_by_hf_username: str,
    duplicate_warning: str,
) -> str:
    submitter_label = _public_submitter_label(submitted_by, submitted_by_hf_username)
    lines = [
        "### Submission recorded",
        "",
        f"- Submission ID: `{submission_id}`",
        f"- Model: `{model_name}`",
        f"- Space: `{space_id}`",
    ]
    if submitted_by_hf_username:
        lines.append(f"- Signed in as: `@{submitted_by_hf_username}`")
        lines.append(f"- Public leaderboard attribution: `{submitter_label}`")
    lines.append("- The private evaluator will pick it up on the next run.")

    if duplicate_warning:
        lines.extend(
            [
                "",
                f"**Warning:** {duplicate_warning}",
            ]
        )

    return "\n".join(lines)


def _public_submitter_label(submitted_by: str, submitted_by_hf_username: str) -> str:
    cleaned_submitter = (submitted_by or "").strip()
    cleaned_hf_username = (submitted_by_hf_username or "").strip()
    if cleaned_submitter and cleaned_hf_username:
        return f"{cleaned_submitter} (@{cleaned_hf_username})"
    if cleaned_hf_username:
        return f"@{cleaned_hf_username}"
    return cleaned_submitter or "anonymous"


def _build_duplicate_submission_warning(model_name: str, submitted_by_hf_username: str) -> str:
    cleaned_hf_username = (submitted_by_hf_username or "").strip()
    if not cleaned_hf_username:
        return ""

    try:
        submission_records = _load_submission_records()
    except Exception:
        return ""

    normalized_username = _normalize_lookup_text(cleaned_hf_username)
    normalized_model_name = _normalize_lookup_text(model_name)

    prior_user_records = [
        record
        for record in submission_records
        if _normalize_lookup_text(record.get("submitted_by_hf_username", "")) == normalized_username
    ]
    if not prior_user_records:
        return ""

    prior_model_records = [
        record
        for record in prior_user_records
        if _normalize_lookup_text(record.get("model_name", "")) == normalized_model_name
    ]
    if prior_model_records:
        latest_record = max(prior_model_records, key=lambda record: str(record.get("submission_time", "")))
        latest_submission_id = str(latest_record.get("submission_id", "")).strip()
        latest_time = _format_timestamp(latest_record.get("submission_time"))
        latest_detail = ""
        if latest_submission_id and latest_time:
            latest_detail = f" Most recent earlier submission: `{latest_submission_id}` at `{latest_time}`."
        return (
            f"`@{cleaned_hf_username}` already has {len(prior_model_records)} earlier submission(s) "
            f"with the model name `{model_name}`.{latest_detail} "
            "This new submission was still recorded, and only the latest completed signed-in submission "
            "will remain public on the leaderboard."
        )

    return (
        f"`@{cleaned_hf_username}` already has earlier submission(s) under a different model name. "
        "Only the latest completed signed-in submission will remain public on the leaderboard."
    )


def _load_submission_records() -> list[dict[str, Any]]:
    if not HF_TOKEN:
        return []

    try:
        submission_files = [
            filename
            for filename in API.list_repo_files(SUBMISSIONS_REPO, repo_type="dataset", token=HF_TOKEN)
            if filename.startswith(SUBMISSION_PREFIX) and filename.endswith(".json")
        ]
    except Exception:
        return []

    records: list[dict[str, Any]] = []
    for filename in submission_files:
        try:
            record_path = hf_hub_download(
                repo_id=SUBMISSIONS_REPO,
                filename=filename,
                repo_type="dataset",
                token=HF_TOKEN,
                force_download=True,
            )
        except Exception:
            continue

        with open(record_path, "r", encoding="utf-8") as handle:
            try:
                records.append(json.load(handle))
            except json.JSONDecodeError:
                continue
    return records


def _backfill_result_submission_fields(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
    submission_ids_to_backfill = {
        str(record.get("submission_id", "")).strip()
        for record in records
        if _needs_submission_backfill(record)
    }
    if not submission_ids_to_backfill:
        return records

    submission_metadata = {
        submission_id: _load_submission_record(submission_id)
        for submission_id in submission_ids_to_backfill
    }

    enriched_records: list[dict[str, Any]] = []
    for record in records:
        enriched_record = dict(record)
        submission_id = str(record.get("submission_id", "")).strip()
        submission = submission_metadata.get(submission_id) or {}
        for field_name in (
            "submitted_by",
            "submitted_by_hf_username",
            "submitted_by_hf_name",
            "submitted_by_hf_profile",
            "submission_tier",
            "model_source_type",
        ):
            if not str(enriched_record.get(field_name, "") or "").strip():
                replacement = str(submission.get(field_name, "") or "").strip()
                if replacement:
                    enriched_record[field_name] = replacement
        enriched_records.append(enriched_record)
    return enriched_records


def _backfill_result_submitter_fields(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
    return _backfill_result_submission_fields(records)


def _needs_submission_backfill(record: dict[str, Any]) -> bool:
    submission_id = str(record.get("submission_id", "")).strip()
    if not submission_id:
        return False
    return any(
        not str(record.get(field_name, "") or "").strip()
        for field_name in (
            "submitted_by_hf_username",
            "submitted_by_hf_name",
            "submitted_by_hf_profile",
            "submission_tier",
            "model_source_type",
        )
    )


def _load_submission_record(submission_id: str) -> dict[str, Any] | None:
    cleaned_submission_id = (submission_id or "").strip()
    if not cleaned_submission_id or not HF_TOKEN:
        return None

    try:
        submission_path = hf_hub_download(
            repo_id=SUBMISSIONS_REPO,
            filename=f"{SUBMISSION_PREFIX}{cleaned_submission_id}.json",
            repo_type="dataset",
            token=HF_TOKEN,
            force_download=True,
        )
    except Exception:
        return None

    with open(submission_path, "r", encoding="utf-8") as handle:
        try:
            return json.load(handle)
        except json.JSONDecodeError:
            return None


def _format_submitter(row: pd.Series) -> str:
    team_name = str(row.get("submitted_by", "") or "").strip()
    hf_username = str(row.get("submitted_by_hf_username", "") or "").strip()

    if team_name and hf_username:
        return f"{team_name} (@{hf_username})"
    if hf_username:
        return f"@{hf_username}"
    return team_name


def _display_column(df: pd.DataFrame, column_name: str) -> pd.Series:
    if column_name not in df.columns:
        return pd.Series([""] * len(df), index=df.index, dtype="object")
    return df[column_name].fillna("")


def _category_score_names(df: pd.DataFrame) -> list[str]:
    names: set[str] = set()
    if "category_scores_json" not in df.columns:
        return []
    for value in df["category_scores_json"]:
        names.update(_loads_json_object(value).keys())
    return sorted(names, key=_humanize_label)


def _get_category_score(row: pd.Series, category: str) -> Any:
    scores = _loads_json_object(row.get("category_scores_json"))
    return scores.get(category)


def _loads_json_object(value: Any) -> dict[str, Any]:
    if isinstance(value, dict):
        return value
    if not isinstance(value, str) or not value.strip():
        return {}
    try:
        parsed = json.loads(value)
    except json.JSONDecodeError:
        return {}
    return parsed if isinstance(parsed, dict) else {}


def _format_score(value: Any) -> str:
    if value is None or pd.isna(value):
        return ""
    try:
        return f"{float(value):.1%}"
    except (TypeError, ValueError):
        return ""


def _format_timestamp(value: Any) -> str:
    timestamp = pd.to_datetime(value, errors="coerce", utc=True)
    if pd.isna(timestamp):
        return ""
    return timestamp.strftime("%Y-%m-%d %H:%M UTC")


def _humanize_label(value: str) -> str:
    return str(value).replace("_", " ").replace("-", " ").title()


def _normalize_lookup_text(value: Any) -> str:
    return str(value or "").strip().lower()


def _encrypt_space_token(space_token: str) -> str:
    if not space_token:
        return ""
    if not SPACE_TOKEN_ENCRYPTION_KEY:
        raise RuntimeError("AITX_SPACE_TOKEN_ENCRYPTION_KEY must be set to submit private Space tokens.")
    return Fernet(SPACE_TOKEN_ENCRYPTION_KEY.encode("utf-8")).encrypt(space_token.encode("utf-8")).decode("utf-8")


def _load_sample_question() -> tuple[dict[str, Any], Any]:
    sample_path = hf_hub_download(
        repo_id=SAMPLE_QUESTIONS_REPO,
        filename=SAMPLE_QUESTIONS_FILE,
        repo_type="dataset",
        token=HF_TOKEN,
    )
    with open(sample_path, "r", encoding="utf-8") as handle:
        records = json.load(handle)

    record = _first_sample_record(records)
    sample_payload = {
        "id": record["id"],
        "patient": record.get("patient", {}),
        "question": record["question"],
    }
    return sample_payload, record.get("answer_expected", "No expected answer found in sample question.")


def _first_sample_record(records: Any) -> dict[str, Any]:
    if isinstance(records, list) and records:
        record = records[0]
    elif isinstance(records, dict) and {"id", "question"}.issubset(records):
        record = records
    elif isinstance(records, dict):
        record = next((item[0] for item in records.values() if isinstance(item, list) and item), None)
    else:
        record = None

    if not isinstance(record, dict) or "id" not in record or "question" not in record:
        raise ValueError(f"Could not find a sample question in `{SAMPLE_QUESTIONS_REPO}/{SAMPLE_QUESTIONS_FILE}`.")
    return record


def _make_space_client(space_id: str, space_token: str) -> Client:
    client_kwargs: dict[str, Any] = {}
    token = (space_token or "").strip()
    if token:
        token_parameter = "token" if "token" in signature(Client).parameters else "hf_token"
        client_kwargs[token_parameter] = token
    return Client(space_id, **client_kwargs)


def _validate_api_format_response(raw_response: Any, expected_id: str) -> dict[str, Any]:
    if not isinstance(raw_response, dict):
        raise ValueError("Response must be a JSON object with `id`, `answer`, `explanation`, and optional `evidence_sources`.")

    missing_fields = [field for field in ("id", "answer", "explanation") if field not in raw_response]
    if missing_fields:
        raise ValueError(f"Response is missing required field(s): `{', '.join(missing_fields)}`.")

    response_id = raw_response["id"]
    if not isinstance(response_id, str) or not response_id.strip():
        raise ValueError("`id` must be a non-empty string.")
    if response_id != expected_id:
        raise ValueError(f"`id` must match the input question id `{expected_id}`; got `{response_id}`.")

    answer = raw_response["answer"]
    if not isinstance(answer, str) or not answer.strip():
        raise ValueError("`answer` must be a non-empty string.")

    explanation = raw_response["explanation"]
    if not isinstance(explanation, str):
        raise ValueError("`explanation` must be a string.")

    evidence_sources = raw_response.get("evidence_sources", [])
    if evidence_sources in (None, ""):
        evidence_sources = []
    if not isinstance(evidence_sources, list):
        raise ValueError("`evidence_sources` is optional, but if provided it must be a list of strings.")
    if any(not isinstance(source, str) for source in evidence_sources):
        raise ValueError("Every item in `evidence_sources` must be a string.")

    return {
        "id": response_id,
        "answer": answer,
        "explanation": explanation,
        "evidence_sources": evidence_sources,
    }


def _format_api_check_report(
    cleaned_space_id: str,
    sample_payload: dict[str, Any],
    expected_answer: Any,
    raw_response: Any,
    validation_error: str,
) -> str:
    status = "failed" if validation_error else "passed"
    heading = "Validator failed" if validation_error else "Validator passed"
    summary_html = _format_api_check_summary(
        cleaned_space_id=cleaned_space_id,
        question_id=str(sample_payload["id"]),
        validation_error=validation_error,
    )

    return (
        f'<div class="api-check-summary api-check-summary--{status}">\n'
        f"<h2>{escape(heading)}</h2>\n"
        f"{summary_html}\n"
        "</div>\n\n"
        "### Question Payload Sent\n\n"
        f"{_json_block(sample_payload)}\n\n"
        "### Returned Response\n\n"
        f"{_json_block(raw_response)}\n\n"
        "### Expected Response for This Public Sample\n\n"
        "The `answer` value below is the correct answer for this public sample question. "
        "Your `explanation` and `evidence_sources` may differ.\n\n"
        f"{_json_block(_expected_api_response_schema(sample_payload['id'], expected_answer))}\n\n"
        f"Status: `{status}`"
    )


def _format_api_check_summary(
    cleaned_space_id: str,
    question_id: str,
    validation_error: str,
) -> str:
    rows = [
        (
            "Called Space",
            f'<span class="api-check-pill">{escape(cleaned_space_id)}</span> '
            f'with public sample question <span class="api-check-pill">{escape(question_id)}</span>',
        ),
        (
            "Endpoint",
            f'<span class="api-check-pill">{escape(PARTICIPANT_API_NAME)}</span>',
        ),
        (
            "Scope",
            "This checks that the output conforms to the expected schema only. It does not evaluate answer quality.",
        ),
    ]
    if validation_error:
        rows.append(
            (
                "Validation Error",
                f'<span class="api-check-error">{escape(validation_error)}</span>',
            )
        )

    return "\n".join(
        '<div class="api-check-row">'
        f'<div class="api-check-label">{escape(label)}</div>'
        f'<div class="api-check-value">{value}</div>'
        "</div>"
        for label, value in rows
    )


def _expected_api_response_schema(question_id: str, expected_answer: Any) -> dict[str, Any]:
    return {
        "id": question_id,
        "answer": expected_answer,
        "explanation": "string",
        "evidence_sources": ["optional list of strings"],
    }


def _json_block(value: Any) -> str:
    return f"```json\n{json.dumps(value, indent=2, sort_keys=True, default=str)}\n```"