import functools
import hashlib
import io
import json
import os
import shutil
import tempfile
import uuid
import zipfile
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlparse

import gradio as gr
import pandas as pd
from huggingface_hub import (
    HfApi,
    hf_hub_download,
    snapshot_download,
)

try:
    import spaces
except ImportError:
    class _SpacesFallback:
        @staticmethod
        def GPU(*decorator_args, **decorator_kwargs):
            if (
                len(decorator_args) == 1
                and callable(decorator_args[0])
                and not decorator_kwargs
            ):
                return decorator_args[0]

            def decorator(function):
                return function

            return decorator

    spaces = _SpacesFallback()

from football_trajectory_similarity import (
    FootballTrajectorySimilarity,
)


APP_TITLE = "TacSIm Online Benchmark"
BENCHMARK_VERSION = "v2.0-batch"

OFFICIAL_FRAME_INTERVAL = 0.1
OFFICIAL_HORIZONS = (3.0, 5.0, 10.0)
OFFICIAL_GRIDS = (
    (10, 6),
    (15, 10),
    (20, 12),
    (30, 20),
    (105, 68),
)
OFFICIAL_GRID_LABELS = [
    f"{length}x{width}"
    for length, width in OFFICIAL_GRIDS
]

LOCAL_TEST_DIR = Path(
    os.getenv("LOCAL_TEST_DIR", "Test")
)
TEST_DATASET_REPO_ID = os.getenv(
    "TEST_DATASET_REPO_ID",
    "",
).strip()
TEST_DATASET_SUBDIR = os.getenv(
    "TEST_DATASET_SUBDIR",
    "",
).strip().strip("/")
TEST_DATASET_REVISION = os.getenv(
    "TEST_DATASET_REVISION",
    "main",
).strip()

LEADERBOARD_REPO_ID = os.getenv(
    "LEADERBOARD_REPO_ID",
    "",
).strip()

SHARED_HF_TOKEN = os.getenv("HF_TOKEN", "").strip()
TEST_DATASET_TOKEN = os.getenv(
    "TEST_DATASET_TOKEN",
    SHARED_HF_TOKEN,
).strip()
LEADERBOARD_HF_TOKEN = os.getenv(
    "LEADERBOARD_HF_TOKEN",
    SHARED_HF_TOKEN,
).strip()

MAX_ZIP_FILES = int(
    os.getenv("MAX_ZIP_FILES", "50000")
)
MAX_ZIP_UNCOMPRESSED_BYTES = int(
    os.getenv(
        "MAX_ZIP_UNCOMPRESSED_BYTES",
        str(5 * 1024 * 1024 * 1024),
    )
)

@spaces.GPU(duration=5)
def _zerogpu_registration_probe():
    """Register one short ZeroGPU callback.

    The actual TacSIm evaluation is CPU-only and must not be wrapped
    by spaces.GPU, otherwise large Test splits are rejected by the
    ZeroGPU per-call duration limit.
    """
    return "ZeroGPU callback registered."


LEADERBOARD_GRID_ORDER = [
    "10x6",
    "15x10",
    "20x12",
    "30x20",
    "105x68",
]

LEADERBOARD_HORIZON_ORDER = [
    "3s",
    "5s",
    "10s",
]

SCORE_COLUMN_SPECS = [
    (
        f"{grid.replace('x', '×')} · {horizon}",
        horizon,
        grid,
    )
    for grid in LEADERBOARD_GRID_ORDER
    for horizon in LEADERBOARD_HORIZON_ORDER
]

SCORE_COLUMNS = [
    display_name
    for display_name, _, _ in SCORE_COLUMN_SPECS
]

DEFAULT_RANKING_METRIC = SCORE_COLUMNS[0]

LEADERBOARD_COLUMNS = [
    "Rank",
    "Method",
    *SCORE_COLUMNS,
    "Test Files",
    "Team / Institution",
    "Submitter",
    "Paper / Code",
    "Submitted",
]


def _safe_text(value, field_name, max_length):
    text = str(value or "").strip()

    if len(text) > max_length:
        raise gr.Error(
            f"{field_name} must contain at most "
            f"{max_length} characters."
        )

    return text


def _safe_url(value):
    url = _safe_text(value, "Paper / Code URL", 500)

    if not url:
        return ""

    parsed = urlparse(url)

    if parsed.scheme not in {"http", "https"}:
        raise gr.Error(
            "Paper / Code URL must start with "
            "http:// or https://."
        )

    return url


def _sha256_file(file_path):
    digest = hashlib.sha256()

    with open(file_path, "rb") as file_object:
        for block in iter(
            lambda: file_object.read(1024 * 1024),
            b"",
        ):
            digest.update(block)

    return digest.hexdigest()


def _combined_file_hash(file_map):
    digest = hashlib.sha256()

    for file_name in sorted(file_map):
        digest.update(file_name.encode("utf-8"))
        digest.update(b"\0")
        digest.update(
            _sha256_file(file_map[file_name]).encode(
                "ascii"
            )
        )
        digest.update(b"\n")

    return digest.hexdigest()


def _collect_csv_files(root_directory):
    root_directory = Path(root_directory)

    if not root_directory.exists():
        raise gr.Error(
            f"Test directory was not found: "
            f"{root_directory}"
        )

    csv_files = sorted(
        path
        for path in root_directory.rglob("*.csv")
        if path.is_file()
    )

    if not csv_files:
        raise gr.Error(
            f"No CSV files were found under "
            f"{root_directory}."
        )

    file_map = {}

    for file_path in csv_files:
        file_name = file_path.name

        if file_name in file_map:
            raise gr.Error(
                "Duplicate CSV basenames were found in the "
                f"test set: {file_name}. Filenames must be "
                "unique across all subdirectories."
            )

        file_map[file_name] = file_path

    return file_map


@functools.lru_cache(maxsize=1)
def _resolve_test_file_map():
    if TEST_DATASET_REPO_ID:
        # Download all CSV files from the private Dataset repository.
        # This is robust to repositories where "Test" is a split name
        # rather than a physical folder.
        snapshot_root = Path(
            snapshot_download(
                repo_id=TEST_DATASET_REPO_ID,
                repo_type="dataset",
                revision=TEST_DATASET_REVISION,
                token=TEST_DATASET_TOKEN or None,
                allow_patterns=[
                    "*.csv",
                    "**/*.csv",
                ],
            )
        )

        configured_root = (
            snapshot_root / TEST_DATASET_SUBDIR
            if TEST_DATASET_SUBDIR
            else snapshot_root
        )

        # Prefer the configured subdirectory when it physically exists.
        # Otherwise, fall back to recursively scanning the repository
        # snapshot. This handles Dataset Viewer split names and files
        # stored directly at the repository root.
        if (
            TEST_DATASET_SUBDIR
            and configured_root.exists()
            and any(configured_root.rglob("*.csv"))
        ):
            test_root = configured_root
        else:
            test_root = snapshot_root
    else:
        test_root = LOCAL_TEST_DIR

    return _collect_csv_files(test_root)


@functools.lru_cache(maxsize=1)
def _test_set_metadata():
    file_map = _resolve_test_file_map()

    return {
        "number_of_files": len(file_map),
        "sha256": _combined_file_hash(file_map),
        "file_names": sorted(file_map),
        "source": (
            (
                f"dataset:{TEST_DATASET_REPO_ID}/"
                f"{TEST_DATASET_SUBDIR or '<auto>'}"
            )
            if TEST_DATASET_REPO_ID
            else f"local:{LOCAL_TEST_DIR}"
        ),
    }


def _is_zip_symlink(zip_info):
    unix_mode = zip_info.external_attr >> 16
    return (unix_mode & 0o170000) == 0o120000


def _safe_extract_submission(zip_path, output_directory):
    output_directory = Path(output_directory)
    output_directory.mkdir(
        parents=True,
        exist_ok=True,
    )

    try:
        archive = zipfile.ZipFile(zip_path, "r")
    except zipfile.BadZipFile as error:
        raise gr.Error(
            "The uploaded file is not a valid ZIP archive."
        ) from error

    with archive:
        members = [
            member
            for member in archive.infolist()
            if not member.is_dir()
        ]

        if len(members) > MAX_ZIP_FILES:
            raise gr.Error(
                f"The ZIP contains too many files "
                f"({len(members)}). Maximum: "
                f"{MAX_ZIP_FILES}."
            )

        total_size = sum(
            member.file_size
            for member in members
        )

        if total_size > MAX_ZIP_UNCOMPRESSED_BYTES:
            raise gr.Error(
                "The uncompressed ZIP content is too large."
            )

        csv_count = 0
        output_root = output_directory.resolve()

        for member in members:
            member_path = Path(member.filename)

            if (
                member_path.is_absolute()
                or ".." in member_path.parts
                or _is_zip_symlink(member)
            ):
                raise gr.Error(
                    "The ZIP archive contains an unsafe path."
                )

            if "__MACOSX" in member_path.parts:
                continue

            if member_path.suffix.lower() != ".csv":
                continue

            destination = (
                output_directory / member_path
            ).resolve()

            try:
                destination.relative_to(output_root)
            except ValueError as error:
                raise gr.Error(
                    "The ZIP archive contains an unsafe path."
                ) from error

            destination.parent.mkdir(
                parents=True,
                exist_ok=True,
            )

            with archive.open(member, "r") as source:
                with open(destination, "wb") as target:
                    shutil.copyfileobj(source, target)

            csv_count += 1

    if csv_count == 0:
        raise gr.Error(
            "The ZIP archive does not contain any CSV files."
        )


def _validate_submission_file_set(
    reference_file_map,
    submission_file_map,
):
    reference_names = set(reference_file_map)
    submission_names = set(submission_file_map)

    missing = sorted(
        reference_names - submission_names
    )
    extra = sorted(
        submission_names - reference_names
    )

    if missing or extra:
        messages = []

        if missing:
            preview = ", ".join(missing[:20])
            messages.append(
                f"Missing {len(missing)} required file(s): "
                f"{preview}"
                + (" ..." if len(missing) > 20 else "")
            )

        if extra:
            preview = ", ".join(extra[:20])
            messages.append(
                f"Found {len(extra)} unexpected file(s): "
                f"{preview}"
                + (" ..." if len(extra) > 20 else "")
            )

        raise gr.Error(
            "\n".join(messages)
        )


def _evaluate_file_pair(
    file_name,
    reference_file,
    generated_file,
):
    evaluator = FootballTrajectorySimilarity(
        reference_file=reference_file,
        generated_file=generated_file,
        frame_interval=OFFICIAL_FRAME_INTERVAL,
        horizons=OFFICIAL_HORIZONS,
        grid_resolutions=OFFICIAL_GRIDS,
    )

    rows = evaluator.evaluate()

    for row in rows:
        row["File"] = file_name
        row["TacSIm Score"] = (
            row["Spatial Occupancy Similarity"]
            + row["Movement Vector Similarity"]
        ) / 2.0

    return rows


def _aggregate_batch_results(detail_table):
    ball_metrics = [
        "TacSIm Score",
        "Spatial Occupancy Similarity",
        "Movement Vector Similarity",
    ]

    aggregate_ball = (
        detail_table.groupby(
            ["Horizon", "Grid"],
            as_index=False,
        )[ball_metrics]
        .mean()
    )

    team_detail = (
        detail_table[
            [
                "File",
                "Horizon",
                "Team Formation Compactness Similarity",
                "Team Speed Similarity",
            ]
        ]
        .drop_duplicates(
            subset=["File", "Horizon"]
        )
    )

    aggregate_team = (
        team_detail.groupby(
            "Horizon",
            as_index=False,
        )[
            [
                "Team Formation Compactness Similarity",
                "Team Speed Similarity",
            ]
        ]
        .mean()
    )

    horizon_scores = (
        detail_table.groupby(
            "Horizon"
        )["TacSIm Score"]
        .mean()
        .to_dict()
    )

    summary = {
        "overall_score": (
            float(
                detail_table["TacSIm Score"].mean()
            )
            * 100.0
        ),
        "score_3s": (
            float(
                horizon_scores.get(
                    "3s",
                    float("nan"),
                )
            )
            * 100.0
        ),
        "score_5s": (
            float(
                horizon_scores.get(
                    "5s",
                    float("nan"),
                )
            )
            * 100.0
        ),
        "score_10s": (
            float(
                horizon_scores.get(
                    "10s",
                    float("nan"),
                )
            )
            * 100.0
        ),
        "spatial_score": (
            float(
                detail_table[
                    "Spatial Occupancy Similarity"
                ].mean()
            )
            * 100.0
        ),
        "movement_score": (
            float(
                detail_table[
                    "Movement Vector Similarity"
                ].mean()
            )
            * 100.0
        ),
        "compactness_score": (
            float(
                team_detail[
                    "Team Formation Compactness Similarity"
                ].mean()
            )
            * 100.0
        ),
        "speed_score": (
            float(
                team_detail[
                    "Team Speed Similarity"
                ].mean()
            )
            * 100.0
        ),
    }

    summary = {
        key: round(value, 2)
        for key, value in summary.items()
    }

    display_ball = aggregate_ball.copy()
    display_team = aggregate_team.copy()

    display_ball[ball_metrics] = (
        display_ball[ball_metrics] * 100.0
    )

    team_metrics = [
        "Team Formation Compactness Similarity",
        "Team Speed Similarity",
    ]
    display_team[team_metrics] = (
        display_team[team_metrics] * 100.0
    )

    display_ball = display_ball.round(2)
    display_team = display_team.round(2)

    summary_table = pd.DataFrame(
        [
            {
                "3s Mean": summary["score_3s"],
                "5s Mean": summary["score_5s"],
                "10s Mean": summary["score_10s"],
                "Spatial Mean": summary[
                    "spatial_score"
                ],
                "Movement Mean": summary[
                    "movement_score"
                ],
                "Compactness Mean": summary[
                    "compactness_score"
                ],
                "Speed Mean": summary["speed_score"],
            }
        ]
    )

    return (
        summary,
        summary_table,
        aggregate_ball,
        aggregate_team,
        display_ball,
        display_team,
    )


def _write_batch_reports(
    detail_table,
    aggregate_ball,
    aggregate_team,
    summary,
    metadata,
):
    report_directory = Path(
        tempfile.mkdtemp(
            prefix="tacsim_batch_report_"
        )
    )

    detail_path = (
        report_directory
        / "per_file_detailed_results.csv"
    )
    aggregate_ball_path = (
        report_directory
        / "aggregate_ball_results.csv"
    )
    aggregate_team_path = (
        report_directory
        / "aggregate_team_results.csv"
    )
    summary_path = (
        report_directory
        / "summary.json"
    )
    report_zip_path = (
        report_directory
        / "tacsim_evaluation_report.zip"
    )

    detail_table.to_csv(
        detail_path,
        index=False,
        encoding="utf-8-sig",
    )
    aggregate_ball.to_csv(
        aggregate_ball_path,
        index=False,
        encoding="utf-8-sig",
    )
    aggregate_team.to_csv(
        aggregate_team_path,
        index=False,
        encoding="utf-8-sig",
    )

    summary_payload = {
        "metadata": metadata,
        "summary": summary,
    }

    summary_path.write_text(
        json.dumps(
            summary_payload,
            indent=2,
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )

    with zipfile.ZipFile(
        report_zip_path,
        "w",
        zipfile.ZIP_DEFLATED,
    ) as archive:
        archive.write(
            detail_path,
            detail_path.name,
        )
        archive.write(
            aggregate_ball_path,
            aggregate_ball_path.name,
        )
        archive.write(
            aggregate_team_path,
            aggregate_team_path.name,
        )
        archive.write(
            summary_path,
            summary_path.name,
        )

    return (
        str(aggregate_ball_path),
        str(detail_path),
        str(summary_path),
        str(report_zip_path),
    )


def _run_batch_evaluation(
    submission_zip,
    progress,
):
    if not submission_zip:
        raise gr.Error(
            "A submission ZIP file is required."
        )

    reference_file_map = _resolve_test_file_map()
    test_metadata = _test_set_metadata()

    with tempfile.TemporaryDirectory(
        prefix="tacsim_submission_"
    ) as temporary_directory:
        extraction_root = (
            Path(temporary_directory) / "submission"
        )

        progress(
            0.01,
            desc="Extracting submission ZIP",
        )

        _safe_extract_submission(
            submission_zip,
            extraction_root,
        )

        submission_file_map = (
            _collect_csv_files(extraction_root)
        )

        _validate_submission_file_set(
            reference_file_map,
            submission_file_map,
        )

        submission_hash = _combined_file_hash(
            submission_file_map
        )

        rows = []
        total_files = len(reference_file_map)

        for file_index, file_name in enumerate(
            sorted(reference_file_map),
            start=1,
        ):
            progress(
                file_index / total_files,
                desc=(
                    f"Evaluating {file_index}/"
                    f"{total_files}: {file_name}"
                ),
            )

            try:
                rows.extend(
                    _evaluate_file_pair(
                        file_name=file_name,
                        reference_file=(
                            reference_file_map[
                                file_name
                            ]
                        ),
                        generated_file=(
                            submission_file_map[
                                file_name
                            ]
                        ),
                    )
                )
            except Exception as error:
                raise gr.Error(
                    f"Evaluation failed for "
                    f"{file_name}: {error}"
                ) from error

    detail_table = pd.DataFrame(rows)

    (
        summary,
        summary_table,
        aggregate_ball,
        aggregate_team,
        display_ball,
        display_team,
    ) = _aggregate_batch_results(detail_table)

    metadata = {
        "benchmark_version": BENCHMARK_VERSION,
        "frame_interval": OFFICIAL_FRAME_INTERVAL,
        "horizons": list(OFFICIAL_HORIZONS),
        "grids": OFFICIAL_GRID_LABELS,
        "number_of_test_files": (
            test_metadata["number_of_files"]
        ),
        "test_set_sha256": (
            test_metadata["sha256"]
        ),
        "test_source": test_metadata["source"],
        "submission_sha256": submission_hash,
        "movement_vector_definition":
            "direct cosine similarity",
        "goalkeeper_excluded_from_team_metrics": True,
        "aggregation": (
            "Arithmetic mean across all matched "
            "test trajectories."
        ),
    }

    (
        aggregate_csv,
        detail_csv,
        summary_json,
        report_zip,
    ) = _write_batch_reports(
        detail_table=detail_table,
        aggregate_ball=aggregate_ball,
        aggregate_team=aggregate_team,
        summary=summary,
        metadata=metadata,
    )

    evaluation_state = {
        "metadata": metadata,
        "summary": summary,
        "aggregate_ball_results": (
            aggregate_ball.to_dict(
                orient="records"
            )
        ),
        "aggregate_team_results": (
            aggregate_team.to_dict(
                orient="records"
            )
        ),
    }

    status = (
        f"Evaluation completed for "
        f"{test_metadata['number_of_files']} "
        f"matched test trajectories. Scores are "
        f"displayed on a 0–100 scale."
    )

    return (
        status,
        summary_table,
        display_ball,
        display_team,
        aggregate_csv,
        detail_csv,
        summary_json,
        report_zip,
        evaluation_state,
    )


def run_official_benchmark(
    submission_zip,
    progress=gr.Progress(),
):
    try:
        return _run_batch_evaluation(
            submission_zip,
            progress,
        )
    except gr.Error:
        raise
    except Exception as error:
        raise gr.Error(str(error)) from error


def _leaderboard_api():
    if not LEADERBOARD_REPO_ID:
        raise gr.Error(
            "LEADERBOARD_REPO_ID is not configured "
            "in Space Settings → Variables."
        )

    if not LEADERBOARD_HF_TOKEN:
        raise gr.Error(
            "LEADERBOARD_HF_TOKEN or HF_TOKEN is "
            "not configured in Space Settings → Secrets."
        )

    return HfApi(
        token=LEADERBOARD_HF_TOKEN
    )


def _submission_record_files():
    if not LEADERBOARD_REPO_ID:
        return []

    try:
        api = HfApi(
            token=LEADERBOARD_HF_TOKEN or None
        )
        files = api.list_repo_files(
            repo_id=LEADERBOARD_REPO_ID,
            repo_type="dataset",
        )
    except Exception:
        return []

    return [
        file_name
        for file_name in files
        if (
            file_name.startswith("submissions/")
            and file_name.endswith(".json")
        )
    ]


def _load_submission_records():
    records = []

    try:
        current_test_hash = (
            _test_set_metadata()["sha256"]
        )
    except Exception:
        current_test_hash = None

    for file_name in _submission_record_files():
        try:
            local_path = hf_hub_download(
                repo_id=LEADERBOARD_REPO_ID,
                filename=file_name,
                repo_type="dataset",
                token=(
                    LEADERBOARD_HF_TOKEN
                    or None
                ),
            )

            payload = json.loads(
                Path(local_path).read_text(
                    encoding="utf-8"
                )
            )

            if (
                payload.get("benchmark_version")
                != BENCHMARK_VERSION
            ):
                continue

            if (
                current_test_hash
                and payload.get("test_set_sha256")
                != current_test_hash
            ):
                continue

            records.append(payload)
        except Exception:
            continue

    return records


def _normalize_grid_label(value):
    return (
        str(value or "")
        .strip()
        .lower()
        .replace("×", "x")
        .replace(" ", "")
    )


def _score_value_to_percent(value):
    try:
        numeric_value = float(value)
    except (TypeError, ValueError):
        return float("nan")

    if not pd.notna(numeric_value):
        return float("nan")

    # Stored batch aggregate values are normally in [0, 1].
    # This also supports future records already stored on a 0–100 scale.
    if -1.000001 <= numeric_value <= 1.000001:
        numeric_value *= 100.0

    return round(numeric_value, 2)


def _extract_scale_scores(record):
    score_lookup = {}

    for row in record.get(
        "aggregate_ball_results",
        [],
    ):
        horizon = str(
            row.get("Horizon", "")
        ).strip()
        grid = _normalize_grid_label(
            row.get("Grid", "")
        )

        key = (horizon, grid)
        score_lookup[key] = _score_value_to_percent(
            row.get("TacSIm Score")
        )

    extracted = {}

    for display_name, horizon, grid in SCORE_COLUMN_SPECS:
        extracted[display_name] = score_lookup.get(
            (horizon, grid),
            float("nan"),
        )

    return extracted


def _leaderboard_dataframe(
    records,
    ranking_metric=DEFAULT_RANKING_METRIC,
):
    if ranking_metric not in SCORE_COLUMNS:
        ranking_metric = DEFAULT_RANKING_METRIC

    if not records:
        return pd.DataFrame(
            columns=LEADERBOARD_COLUMNS
        )

    rows = []

    for record in records:
        row = {
            "Method": record.get(
                "method",
                "",
            ),
            "Test Files": record.get(
                "number_of_test_files",
                "",
            ),
            "Team / Institution": record.get(
                "team",
                "",
            ),
            "Submitter": record.get(
                "submitter",
                "",
            ),
            "Paper / Code": record.get(
                "paper_url",
                "",
            ),
            "Submitted": record.get(
                "submitted_at",
                "",
            ),
        }

        row.update(
            _extract_scale_scores(record)
        )
        rows.append(row)

    dataframe = pd.DataFrame(rows)

    for score_column in SCORE_COLUMNS:
        if score_column not in dataframe.columns:
            dataframe[score_column] = float("nan")

        dataframe[score_column] = pd.to_numeric(
            dataframe[score_column],
            errors="coerce",
        ).round(2)

    valid_mask = dataframe[
        ranking_metric
    ].notna()

    dataframe["_SelectedMetricRank"] = pd.NA

    if valid_mask.any():
        dataframe.loc[
            valid_mask,
            "_SelectedMetricRank",
        ] = (
            dataframe.loc[
                valid_mask,
                ranking_metric,
            ]
            .rank(
                method="min",
                ascending=False,
            )
            .astype("Int64")
        )

    dataframe = dataframe.sort_values(
        by=[
            ranking_metric,
            "Method",
            "Submitted",
        ],
        ascending=[
            False,
            True,
            True,
        ],
        na_position="last",
    ).reset_index(drop=True)

    dataframe.insert(
        0,
        "Rank",
        dataframe.pop(
            "_SelectedMetricRank"
        ),
    )

    return dataframe[LEADERBOARD_COLUMNS]


def refresh_leaderboard(
    ranking_metric=DEFAULT_RANKING_METRIC,
):
    if ranking_metric not in SCORE_COLUMNS:
        ranking_metric = DEFAULT_RANKING_METRIC

    records = _load_submission_records()
    dataframe = _leaderboard_dataframe(
        records,
        ranking_metric,
    )

    if not LEADERBOARD_REPO_ID:
        status = (
            "Leaderboard storage is not configured. "
            "Set LEADERBOARD_REPO_ID and a write token."
        )
    else:
        status = (
            f"Loaded {len(dataframe)} valid "
            f"submission(s) for {BENCHMARK_VERSION}. "
            f"Ranked independently by: {ranking_metric}. "
            "Higher is better; tied scores share the same rank."
        )

    return dataframe, status


def submit_to_leaderboard(
    method_name,
    team_name,
    submitter_name,
    paper_url,
    confirmation,
    evaluation_state,
    ranking_metric,
):
    if not evaluation_state:
        raise gr.Error(
            "Run the official benchmark before "
            "submitting."
        )

    if not confirmation:
        raise gr.Error(
            "Confirm that this is a genuine model "
            "submission before continuing."
        )

    method = _safe_text(
        method_name,
        "Method name",
        120,
    )
    team = _safe_text(
        team_name,
        "Team / Institution",
        160,
    )
    submitter = _safe_text(
        submitter_name,
        "Submitter",
        120,
    )
    url = _safe_url(paper_url)

    if not method:
        raise gr.Error(
            "Method name is required."
        )

    if not submitter:
        raise gr.Error(
            "Submitter name is required."
        )

    api = _leaderboard_api()
    metadata = evaluation_state["metadata"]
    submission_hash = metadata[
        "submission_sha256"
    ]

    for existing in _load_submission_records():
        if (
            existing.get("submission_sha256")
            == submission_hash
            and existing.get(
                "method",
                "",
            ).casefold() == method.casefold()
        ):
            raise gr.Error(
                "This method and submission have "
                "already been added."
            )

    submission_id = str(uuid.uuid4())
    submitted_at = datetime.now(
        timezone.utc
    ).replace(microsecond=0).isoformat()

    payload = {
        "submission_id": submission_id,
        "benchmark_version": BENCHMARK_VERSION,
        "method": method,
        "team": team,
        "submitter": submitter,
        "paper_url": url,
        "submitted_at": submitted_at,
        "submission_sha256": submission_hash,
        "test_set_sha256": metadata[
            "test_set_sha256"
        ],
        "number_of_test_files": metadata[
            "number_of_test_files"
        ],
        "summary": evaluation_state["summary"],
        "evaluation_metadata": metadata,
        "aggregate_ball_results":
            evaluation_state[
                "aggregate_ball_results"
            ],
        "aggregate_team_results":
            evaluation_state[
                "aggregate_team_results"
            ],
    }

    payload_bytes = json.dumps(
        payload,
        indent=2,
        ensure_ascii=False,
    ).encode("utf-8")

    api.upload_file(
        path_or_fileobj=io.BytesIO(
            payload_bytes
        ),
        path_in_repo=(
            f"submissions/{submission_id}.json"
        ),
        repo_id=LEADERBOARD_REPO_ID,
        repo_type="dataset",
        commit_message=(
            f"Add leaderboard submission: "
            f"{method}"
        ),
    )

    leaderboard, _ = refresh_leaderboard(
        ranking_metric
    )

    return (
        f"Submission accepted: {method}. "
        f"Submission ID: {submission_id}",
        leaderboard,
    )


def benchmark_status():
    try:
        metadata = _test_set_metadata()

        status = (
            f"Benchmark {BENCHMARK_VERSION} is ready. "
            f"Backend test files: "
            f"{metadata['number_of_files']}. "
            f"Test-set hash: "
            f"{metadata['sha256'][:12]}..."
        )
    except Exception as error:
        status = (
            f"Benchmark test set is not ready: "
            f"{error}"
        )

    return status


def build_interface():
    with gr.Blocks(title=APP_TITLE) as demo:
        evaluation_state = gr.State(value=None)

        # Hidden event used only to register a valid ZeroGPU callback.
        # The official full-test evaluation itself runs on CPU.
        zerogpu_probe_button = gr.Button(
            "ZeroGPU registration",
            visible=False,
        )
        zerogpu_probe_output = gr.Textbox(
            visible=False,
        )
        zerogpu_probe_button.click(
            fn=_zerogpu_registration_probe,
            inputs=[],
            outputs=[zerogpu_probe_output],
            api_visibility="private",
        )

        gr.Markdown(
            f"""
# {APP_TITLE}

The backend contains the complete hidden **Test** split. Upload one ZIP
containing a generated CSV for every test trajectory.

The submission CSV filenames must exactly match the backend test
filenames. Folder nesting inside the ZIP is allowed, but CSV basenames
must be unique.

**Official settings**

- Benchmark version: `{BENCHMARK_VERSION}`
- Frame interval: `0.1s`
- Horizons: `3s`, `5s`, `10s`
- Grids: `10×6`, `15×10`, `20×12`, `30×20`, `105×68`
- TacSIm Score: `(Spatial Occupancy + Movement Vector) / 2`
- Each scale-and-horizon score is reported separately
- Display scale: `0–100`
"""
        )

        backend_status = gr.Textbox(
            label="Backend Test-Set Status",
            value=benchmark_status(),
            interactive=False,
        )

        with gr.Tab("Official Batch Benchmark"):
            gr.Markdown(
                """
## Submission ZIP format

```text
submission.zip
├── 00001.csv
├── 00002.csv
├── 00003.csv
└── ...
```

A wrapper folder is also accepted:

```text
submission.zip
└── MyMethod/
    ├── 00001.csv
    ├── 00002.csv
    └── ...
```
"""
            )

            submission_zip = gr.File(
                label="Generated Test Trajectories (.zip)",
                file_types=[".zip"],
                type="filepath",
            )

            run_button = gr.Button(
                "Run Full Test-Set Evaluation",
                variant="primary",
            )

            evaluation_status = gr.Textbox(
                label="Evaluation Status",
                interactive=False,
            )

            summary_table = gr.Dataframe(
                label="Aggregate Submission Summary",
                interactive=False,
            )

            ball_table = gr.Dataframe(
                label="Average Ball Metrics by Horizon and Grid",
                interactive=False,
            )

            team_table = gr.Dataframe(
                label="Average Team Metrics by Horizon",
                interactive=False,
            )

            with gr.Row():
                aggregate_csv = gr.File(
                    label="Aggregate Ball CSV"
                )
                detail_csv = gr.File(
                    label="Per-File Detailed CSV"
                )

            with gr.Row():
                summary_json = gr.File(
                    label="Summary JSON"
                )
                report_zip = gr.File(
                    label="Complete Evaluation Report"
                )

            run_button.click(
                fn=run_official_benchmark,
                inputs=[submission_zip],
                outputs=[
                    evaluation_status,
                    summary_table,
                    ball_table,
                    team_table,
                    aggregate_csv,
                    detail_csv,
                    summary_json,
                    report_zip,
                    evaluation_state,
                ],
                show_progress="full",
            )

            gr.Markdown(
                """
## Submit evaluated result to the leaderboard

Only the result currently stored in this evaluation session can be
submitted. Synthetic, manually edited, or incomplete submissions must
not be entered as model results.
"""
            )

            with gr.Row():
                method_name = gr.Textbox(
                    label="Method Name",
                    placeholder="Example: Foresight-MAIL",
                )
                team_name = gr.Textbox(
                    label="Team / Institution",
                    placeholder="Optional",
                )

            with gr.Row():
                submitter_name = gr.Textbox(
                    label="Submitter",
                    placeholder=(
                        "Name or Hugging Face username"
                    ),
                )
                paper_url = gr.Textbox(
                    label="Paper / Code URL",
                    placeholder="https://...",
                )

            confirmation = gr.Checkbox(
                label=(
                    "I confirm that this ZIP contains "
                    "genuine model outputs for the complete "
                    "test split and has not been manually "
                    "edited to alter benchmark scores."
                ),
                value=False,
            )

            submit_button = gr.Button(
                "Submit to Leaderboard"
            )

            submission_status = gr.Textbox(
                label="Submission Status",
                interactive=False,
            )

        with gr.Tab("Leaderboard"):
            gr.Markdown(
                f"""
## TacSIm Leaderboard — {BENCHMARK_VERSION}

The leaderboard reports a separate TacSIm **Score** for every official
grid resolution and prediction horizon. There is no Overall score.

Select any score column below to rank methods independently by that
column. Higher is better. Tied values receive the same rank.
"""
            )

            ranking_metric = gr.Dropdown(
                choices=SCORE_COLUMNS,
                value=DEFAULT_RANKING_METRIC,
                label="Ranking Metric",
                info=(
                    "Rank and sort the leaderboard by one "
                    "scale-and-horizon score."
                ),
            )

            leaderboard_table = gr.Dataframe(
                headers=LEADERBOARD_COLUMNS,
                value=pd.DataFrame(
                    columns=LEADERBOARD_COLUMNS
                ),
                interactive=False,
                wrap=True,
            )

            leaderboard_status = gr.Textbox(
                label="Leaderboard Status",
                interactive=False,
            )

            refresh_button = gr.Button(
                "Refresh Leaderboard"
            )

            refresh_button.click(
                fn=refresh_leaderboard,
                inputs=[ranking_metric],
                outputs=[
                    leaderboard_table,
                    leaderboard_status,
                ],
                api_name="refresh_leaderboard",
            )

            ranking_metric.change(
                fn=refresh_leaderboard,
                inputs=[ranking_metric],
                outputs=[
                    leaderboard_table,
                    leaderboard_status,
                ],
            )

            demo.load(
                fn=refresh_leaderboard,
                inputs=[ranking_metric],
                outputs=[
                    leaderboard_table,
                    leaderboard_status,
                ],
            )

        submit_button.click(
            fn=submit_to_leaderboard,
            inputs=[
                method_name,
                team_name,
                submitter_name,
                paper_url,
                confirmation,
                evaluation_state,
                ranking_metric,
            ],
            outputs=[
                submission_status,
                leaderboard_table,
            ],
            api_visibility="private",
        )

        gr.Markdown(
            """
## Required columns in every generated CSV

`player_1_x`, `player_1_y`, ..., `player_11_x`, `player_11_y`,
`ball_x`, and `ball_y`.

Additional columns, including action columns, are allowed and ignored
by the trajectory evaluator.
"""
        )

    return demo


if __name__ == "__main__":
    app = build_interface()
    app.queue(
        default_concurrency_limit=1
    ).launch()