miniapp_leaderboard.py

import datetime
import io
import json
import os
import re
import uuid
from urllib.parse import urlparse

import gradio as gr
import numpy as np
import pandas as pd
import requests
from huggingface_hub import HfApi, hf_hub_download
from huggingface_hub.errors import RepositoryNotFoundError

APP_NAME = "miniapp"

HF_TOKEN = os.environ.get("HF_TOKEN") or os.environ.get("TOKEN") or os.environ.get("HUGGINGFACE_TOKEN")
LEADERBOARD_DATASET = (os.environ.get("LEADERBOARD_DATASET") or "").strip()
MAX_ENTRIES = int(os.environ.get("MAX_ENTRIES", "500"))

ENTRIES_PREFIX = "entries/"

LEADERBOARD_COLUMNS = [
    "Model name",
    "Avg",
    "Easy",
    "Mid",
    "Hard",
    "Games",
    "Science",
    "Tools",
    "Humanities",
    "Viz",
    "Lifestyle",
    "Submitted at",
    "Submitter",
]

NUMERIC_COLS = [
    "Avg",
    "Easy",
    "Mid",
    "Hard",
    "Games",
    "Science",
    "Tools",
    "Humanities",
    "Viz",
    "Lifestyle",
]

# 展示顺序：Avg 最左
DISPLAY_ORDER = [
    "Avg",
    "Model name",
    "Easy",
    "Mid",
    "Hard",
    "Games",
    "Science",
    "Tools",
    "Humanities",
    "Viz",
    "Lifestyle",
]
SORTABLE_COLS = DISPLAY_ORDER[:]

IN_SPACES = bool(
    os.environ.get("SPACE_ID")
    or os.environ.get("SPACE_REPO_NAME")
    or os.environ.get("SPACE_AUTHOR_NAME")
    or os.environ.get("system", "") == "spaces"
)


def _api() -> HfApi:
    return HfApi(token=HF_TOKEN)


def _is_valid_http_url(url: str) -> bool:
    try:
        parsed = urlparse(url)
        return parsed.scheme in ("http", "https") and bool(parsed.netloc)
    except Exception:
        return False


def _slug(s: str, max_len: int = 60) -> str:
    s = (s or "").strip().lower()
    s = re.sub(r"[^a-z0-9]+", "-", s)
    s = re.sub(r"-{2,}", "-", s).strip("-")
    return (s[:max_len] or "x")


def _empty_df() -> pd.DataFrame:
    return pd.DataFrame(columns=LEADERBOARD_COLUMNS)


def _ensure_dataset_readable() -> tuple[bool, str]:
    if not HF_TOKEN:
        return False, "Space is missing HF_TOKEN (Secrets)."
    if not LEADERBOARD_DATASET:
        return False, "Space is missing LEADERBOARD_DATASET (Secrets)."
    api = _api()
    try:
        api.repo_info(repo_id=LEADERBOARD_DATASET, repo_type="dataset")
        return True, ""
    except RepositoryNotFoundError:
        return False, (
            f"Dataset repo not found: {LEADERBOARD_DATASET}. "
            "Create it first (as a dataset) or fix LEADERBOARD_DATASET."
        )
    except Exception:
        return False, "Cannot access the dataset repo. Check token permissions."


def _list_entry_files() -> list[str]:
    ok, _ = _ensure_dataset_readable()
    if not ok:
        return []
    api = _api()
    try:
        files = api.list_repo_files(repo_id=LEADERBOARD_DATASET, repo_type="dataset")
    except Exception:
        return []

    entry_files = [f for f in files if f.startswith(ENTRIES_PREFIX) and f.endswith(".json")]
    entry_files.sort(reverse=True)
    return entry_files[:MAX_ENTRIES]


def _load_entries_df() -> pd.DataFrame:
    ok, _ = _ensure_dataset_readable()
    if not ok:
        return _empty_df()

    rows: list[dict] = []
    for filename in _list_entry_files():
        try:
            path = hf_hub_download(
                repo_id=LEADERBOARD_DATASET,
                repo_type="dataset",
                filename=filename,
                token=HF_TOKEN,
            )
            with open(path, "r", encoding="utf-8") as fp:
                row = json.load(fp)
                rows.append(row)
        except Exception:
            continue

    if not rows:
        return _empty_df()

    df = pd.DataFrame(rows)
    for c in LEADERBOARD_COLUMNS:
        if c not in df.columns:
            df[c] = ""

    df = df[LEADERBOARD_COLUMNS]
    for c in NUMERIC_COLS:
        df[c] = pd.to_numeric(df[c], errors="coerce")

    df = df.sort_values(by=["Submitted at"], ascending=False, kind="stable")
    return df


def _parse_hf_created_at(created_at: str) -> datetime.datetime | None:
    try:
        if created_at.endswith("Z"):
            created_at = created_at[:-1] + "+00:00"
        return datetime.datetime.fromisoformat(created_at)
    except Exception:
        return None


def _check_user_eligibility(username: str) -> tuple[bool, str]:
    try:
        r = requests.get(f"https://huggingface.co/api/users/{username}/overview", timeout=10)
        r.raise_for_status()
        created_at = r.json().get("createdAt")
        if not created_at:
            return False, "Cannot verify account creation date."
        dt = _parse_hf_created_at(created_at)
        if not dt:
            return False, "Cannot parse account creation date."
        now = datetime.datetime.now(datetime.timezone.utc)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=datetime.timezone.utc)
        if (now - dt).days < 120:
            return False, "Account must be older than 4 months to submit."
        return True, ""
    except Exception:
        return False, "Cannot verify Hugging Face account. Please try again later."


def _submitted_today(username: str) -> bool:
    df = _load_entries_df()
    if df.empty:
        return False
    today = datetime.datetime.utcnow().date().isoformat()
    user_rows = df[df["Submitter"].astype(str) == username]
    if user_rows.empty:
        return False
    return any(str(v).startswith(today) for v in user_rows["Submitted at"].tolist())


# ---------- HTML Leaderboard ----------
def _fmt_cell(v):
    if v is None or (isinstance(v, float) and pd.isna(v)):
        return ""
    if isinstance(v, (int, float, np.number)):
        return f"{float(v):.2f}"
    return str(v)


def _apply_search_and_sort(df: pd.DataFrame, search_text: str, sort_col: str, sort_dir: str) -> pd.DataFrame:
    s = (search_text or "").strip().lower()
    if s:
        df = df[df["Model name"].astype(str).str.lower().str.contains(s, na=False)]

    sort_col = sort_col if sort_col in df.columns else "Avg"
    asc = sort_dir == "asc"
    df = df.sort_values(by=[sort_col], ascending=asc, kind="stable", na_position="last")
    return df


def _render_leaderboard_html(df: pd.DataFrame, sort_col: str, sort_dir: str) -> str:
    import html as _html

    def th(label, col=None, align_left=False, cls=""):
        if col:
            arrow = ""
            if col == sort_col:
                arrow = " ▲" if sort_dir == "asc" else " ▼"
            al = " left" if align_left else ""
            return f'<th class="th clickable{al} {cls}" data-col="{_html.escape(col)}">{_html.escape(label)}{arrow}</th>'
        al = " left" if align_left else ""
        return f'<th class="th{al} {cls}">{_html.escape(label)}</th>'

    trs = []
    for _, r in df.iterrows():
        tds = []
        for c in DISPLAY_ORDER:
            val = _fmt_cell(r.get(c, ""))
            if c == "Model name":
                tds.append(f'<td class="td model">{_html.escape(val)}</td>')
            else:
                tds.append(f'<td class="td num">{_html.escape(val)}</td>')
        trs.append("<tr class='tr'>" + "".join(tds) + "</tr>")

    return f"""
    <div class="table-wrap">
      <div class="table-scroll">
        <table class="table" id="lb_table">
          <thead>
            <tr class="r1">
              {th("Avg. (%)", "Avg", cls="avg")}
              {th("Model", "Model name", align_left=True, cls="model")}
              <th class="th group" colspan="9">Pass Rate (%)</th>
            </tr>
            <tr class="r2">
              <th class="th"></th>
              <th class="th"></th>
              <th class="th group" colspan="3">Difficulty</th>
              <th class="th group" colspan="6">Domain</th>
            </tr>
            <tr class="r3">
              <th class="th"></th>
              <th class="th"></th>
              {th("Easy", "Easy")}
              {th("Mid", "Mid")}
              {th("Hard", "Hard")}
              {th("Games", "Games")}
              {th("Science", "Science")}
              {th("Tools", "Tools")}
              {th("Humanities", "Humanities")}
              {th("Viz.", "Viz")}
              {th("Lifestyle", "Lifestyle")}
            </tr>
          </thead>
          <tbody>
            {''.join(trs)}
          </tbody>
        </table>
      </div>
    </div>
    """


def render_lb(search_text: str, sort_col: str, sort_dir: str) -> str:
    df = _load_entries_df()
    df = _apply_search_and_sort(df, search_text, sort_col, sort_dir)
    return _render_leaderboard_html(df, sort_col, sort_dir)


def toggle_sort(clicked_col: str, current_col: str, current_dir: str):
    clicked_col = (clicked_col or "").strip()
    if clicked_col not in SORTABLE_COLS:
        return current_col, current_dir
    if clicked_col == current_col:
        return current_col, ("asc" if current_dir == "desc" else "desc")
    return clicked_col, "desc"


# ---------- Submit ----------
def submit(
    model_api: str,
    api_key: str,
    search_text: str,
    sort_col: str,
    sort_dir: str,
    profile: gr.OAuthProfile | None,
):
    if IN_SPACES and (profile is None or not getattr(profile, "username", None)):
        return "You must log in to submit.", render_lb(search_text, sort_col, sort_dir)

    submitter = (getattr(profile, "username", None) if profile is not None else "local") or "anonymous"

    model_api = (model_api or "").strip()
    api_key = (api_key or "").strip()

    if not model_api:
        return "Model API URL is required.", render_lb(search_text, sort_col, sort_dir)
    if not _is_valid_http_url(model_api):
        return "Model API must be a valid http(s) URL.", render_lb(search_text, sort_col, sort_dir)
    if not api_key:
        return "API key is required.", render_lb(search_text, sort_col, sort_dir)

    ok, msg = _ensure_dataset_readable()
    if not ok:
        return msg, render_lb(search_text, sort_col, sort_dir)

    if IN_SPACES:
        ok, msg = _check_user_eligibility(submitter)
        if not ok:
            return msg, render_lb(search_text, sort_col, sort_dir)

    if _submitted_today(submitter):
        return "You have already submitted today. Please try again tomorrow.", render_lb(search_text, sort_col, sort_dir)

    now = datetime.datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
    nonce = uuid.uuid4().hex[:8]
    safe_user = _slug(submitter)

    host = urlparse(model_api).netloc or "unknown"
    model_name = host

    safe_model = _slug(model_name)
    path_in_repo = f"{ENTRIES_PREFIX}{now[:10]}/{now}-{safe_user}-{safe_model}-{nonce}.json"

    payload = {
        "Model name": model_name,
        "Avg": None,
        "Easy": None,
        "Mid": None,
        "Hard": None,
        "Games": None,
        "Science": None,
        "Tools": None,
        "Humanities": None,
        "Viz": None,
        "Lifestyle": None,
        "Submitted at": now,
        "Submitter": submitter,
        "Model API": model_api,
    }

    api = _api()
    data = (json.dumps(payload, ensure_ascii=False, indent=2) + "\n").encode("utf-8")
    api.upload_file(
        repo_id=LEADERBOARD_DATASET,
        repo_type="dataset",
        path_or_fileobj=io.BytesIO(data),
        path_in_repo=path_in_repo,
        commit_message=f"miniapp: submit {submitter}/{model_name}",
        token=HF_TOKEN,
    )

    return "Submitted successfully.", render_lb(search_text, sort_col, sort_dir)


CSS = r"""
/* 全宽 */
.gradio-container { max-width: 100% !important; }
#page { padding: 16px; }

/* 顶部一行：搜索弱化 */
#topbar { display:flex; align-items:center; justify-content:space-between; gap:12px; margin-bottom: 10px; }
#titleline { font-weight: 700; font-size: 18px; }
#searchbox { width: 280px; }
#searchbox label { display:none !important; }
#searchbox textarea, #searchbox input {
  height: 34px !important;
  border-radius: 8px !important;
  border: 1px solid #e5e7eb !important;
  background: #fff !important;
  box-shadow: none !important;
}
#searchbox textarea::placeholder, #searchbox input::placeholder { color: #9ca3af; }

/* 表格：浅灰分割线风格 */
.table-wrap{
  width: 100%;
  border: 1px solid #e5e7eb;
  border-radius: 8px;
  background: #fff;
}
.table-scroll{ width: 100%; overflow-x: auto; }
table.table{
  width: 100%;
  border-collapse: separate;
  border-spacing: 0;
  min-width: 1100px;
}

/* 表头 */
th.th{
  font-family: ui-sans-serif, system-ui, -apple-system, Segoe UI, Roboto, Helvetica, Arial;
  font-weight: 600;
  font-size: 13px;
  color: #111827;
  padding: 10px 12px;
  text-align: center;
  background: #f9fafb;
  border-bottom: 1px solid #e5e7eb;
  border-right: 1px solid #e5e7eb;
  white-space: nowrap;
}
thead tr.r1 th.th, thead tr.r2 th.th { background: #f9fafb; }
thead tr.r3 th.th { background: #ffffff; }

th.th.left{ text-align:left; }
th.group{ color:#374151; font-weight:600; }
th.th:last-child{ border-right: none; }

/* body */
td.td{
  font-family: ui-sans-serif, system-ui, -apple-system, Segoe UI, Roboto, Helvetica, Arial;
  font-size: 13px;
  color: #111827;
  padding: 10px 12px;
  border-bottom: 1px solid #f0f1f3;
  border-right: 1px solid #f0f1f3;
  background: #fff;
}
td.td:last-child{ border-right: none; }
td.num{ text-align:right; }
td.model{ text-align:left; min-width: 280px; }
tr.tr:hover td.td{ background: #fafafa; }

/* 可点击排序 */
th.clickable{ cursor:pointer; user-select:none; }
th.clickable:hover{ background:#f3f4f6; }

/* 提交：全宽 */
#submit_card{
  width: 100%;
  border: 1px solid #e5e7eb;
  border-radius: 8px;
  padding: 12px;
  background: #fff;
  margin-top: 14px;
}
#submit_card .hint{
  margin: 0 0 10px 0;
  color: #6b7280;
  font-size: 13px;
}
"""

with gr.Blocks(title=f"{APP_NAME} leaderboard") as demo:
    with gr.Column(elem_id="page"):
        with gr.Row(elem_id="topbar"):
            gr.Markdown(f"<div id='titleline'>{APP_NAME} leaderboard</div>")
            with gr.Row():
                search_text = gr.Textbox(
                    elem_id="searchbox",
                    placeholder="Search model…",
                    show_label=False,
                    container=False,
                    scale=1,
                )
                refresh_btn = gr.Button("Refresh", scale=0)

        sort_col = gr.State("Avg")
        sort_dir = gr.State("desc")

        lb_html = gr.HTML(value=render_lb("", "Avg", "desc"))

        clicked_col = gr.Textbox(visible=False, elem_id="clicked_col")

        gr.HTML(
            """
<script>
(function(){
  function bindClicks(){
    const table = document.getElementById("lb_table");
    const hidden = document.getElementById("clicked_col");
    if(!table || !hidden) return;

    table.querySelectorAll("th.clickable").forEach(th=>{
      th.onclick = () => {
        const col = th.getAttribute("data-col") || "";
        hidden.value = col;
        hidden.dispatchEvent(new Event("input", {bubbles:true}));
        hidden.dispatchEvent(new Event("change", {bubbles:true}));
      };
    });
  }

  const obs = new MutationObserver(()=>bindClicks());
  obs.observe(document.body, {subtree:true, childList:true});
  setTimeout(bindClicks, 250);
})();
</script>
"""
        )

        search_text.change(render_lb, inputs=[search_text, sort_col, sort_dir], outputs=[lb_html])
        refresh_btn.click(render_lb, inputs=[search_text, sort_col, sort_dir], outputs=[lb_html])

        def _on_click(col, cur_col, cur_dir, s):
            new_col, new_dir = toggle_sort(col, cur_col, cur_dir)
            return new_col, new_dir, render_lb(s, new_col, new_dir)

        clicked_col.change(
            _on_click,
            inputs=[clicked_col, sort_col, sort_dir, search_text],
            outputs=[sort_col, sort_dir, lb_html],
        )

        # 提交模块：全宽
        gr.HTML(
            """
<div id="submit_card">
  <div class="hint">
    <b>Submission</b> — Submit <b>Model API URL</b> and <b>API key</b> only.
    Requires login (Spaces). One submission per user per day. Account must be older than 4 months.
    API key will <b>not</b> be stored.
  </div>
</div>
"""
        )

        with gr.Column():
            with gr.Row():
                model_api = gr.Textbox(label="Model API URL", placeholder="https://...", scale=3)
                api_key = gr.Textbox(label="API key", type="password", placeholder="Will not be stored", scale=2)
            with gr.Row():
                gr.LoginButton()
                submit_btn = gr.Button("Submit", variant="primary")
            status = gr.Markdown()

        submit_btn.click(
            submit,
            inputs=[model_api, api_key, search_text, sort_col, sort_dir],
            outputs=[status, lb_html],
        )

demo.launch(css=CSS, ssr_mode=False)