| import base64 |
| import json |
| import os |
| import re |
| from datetime import date |
| from html import escape |
| from http import HTTPStatus |
| from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer |
| from pathlib import Path |
| from typing import Any |
| from urllib.parse import parse_qs, urlencode, urlparse |
|
|
|
|
| ROOT = Path(__file__).resolve().parent |
| DATASETS_PATH = ROOT / "data" / "datasets.json" |
| RESULTS_PATH = Path(os.getenv("RUSBEIR_RESULTS_PATH", ROOT / "data" / "results.jsonl")) |
| LOGO_PATH = ROOT / "assets" / "rusBeIR_logo.png" |
| PORT = int(os.getenv("PORT", "7860")) |
|
|
| DEFAULT_METRIC = "NDCG@10" |
| METRICS = ["NDCG@10", "MAP@10", "Recall@10", "P@10", "MRR@10"] |
| TRAILING_COLUMNS = ["Date", "Source URL"] |
| DISPLAY_COLUMN_NAMES = { |
| "Model ID": "Model<br>ID", |
| "Organization": "Org.", |
| "Source URL": "Source<br>URL", |
| "sberquad-retrieval": "sberquad<br>retrieval", |
| "ruscibench-retrieval": "ruscibench<br>retrieval", |
| "wikifacts-articles": "wikifacts<br>articles", |
| "wikifacts-para": "wikifacts<br>para", |
| "wikifacts-sents": "wikifacts<br>sents", |
| "wikifacts-window_2": "wikifacts<br>window 2", |
| "wikifacts-window_3": "wikifacts<br>window 3", |
| "wikifacts-window_4": "wikifacts<br>window 4", |
| "wikifacts-window_5": "wikifacts<br>window 5", |
| "wikifacts-window_6": "wikifacts<br>window 6", |
| } |
|
|
| CSS = """ |
| :root { |
| --bg: #f9fafb; |
| --panel: #ffffff; |
| --panel-soft: #f6f7f9; |
| --text: #1f2937; |
| --muted: #6b7280; |
| --line: #e5e7eb; |
| --line-strong: #d1d5db; |
| --accent: #ff6f00; |
| --accent-soft: #fff7ed; |
| --shadow: 0 1px 2px rgba(0, 0, 0, 0.04); |
| } |
| * { box-sizing: border-box; } |
| body { |
| margin: 0; |
| background: var(--bg); |
| color: var(--text); |
| font-family: Inter, ui-sans-serif, system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif; |
| font-size: 14px; |
| } |
| a { color: #c2410c; font-weight: 600; text-decoration: none; } |
| a:hover { text-decoration: underline; } |
| .page { |
| width: min(1480px, 100%); |
| margin: 0 auto; |
| padding: 24px; |
| } |
| .shell { display: flex; flex-direction: column; gap: 16px; } |
| .hero, .card, .panel { |
| background: var(--panel); |
| border: 1px solid var(--line); |
| border-radius: 8px; |
| box-shadow: var(--shadow); |
| } |
| .hero { |
| display: grid; |
| grid-template-columns: 1fr auto; |
| gap: 18px; |
| align-items: start; |
| padding: 22px; |
| } |
| .logo { width: 132px; max-width: 24vw; height: auto; object-fit: contain; } |
| .kicker { |
| color: #9a3412; |
| font-size: 12px; |
| font-weight: 800; |
| letter-spacing: 0.08em; |
| text-transform: uppercase; |
| margin-bottom: 8px; |
| } |
| h1 { font-size: 36px; line-height: 1.12; margin: 0 0 10px; letter-spacing: 0; } |
| .subtitle { color: var(--muted); font-size: 15px; line-height: 1.55; margin: 0; max-width: 860px; } |
| .badges { display: flex; flex-wrap: wrap; gap: 8px; margin-top: 16px; } |
| .badge { |
| display: inline-flex; |
| align-items: center; |
| border: 1px solid var(--line); |
| border-radius: 8px; |
| background: var(--panel-soft); |
| color: #374151; |
| padding: 5px 10px; |
| font-size: 13px; |
| font-weight: 600; |
| } |
| .cards { display: grid; grid-template-columns: repeat(4, minmax(0, 1fr)); gap: 12px; } |
| .card { padding: 15px; } |
| .card-label { color: var(--muted); font-size: 12px; font-weight: 700; text-transform: uppercase; letter-spacing: 0.04em; } |
| .card-value { font-size: 24px; line-height: 1.2; font-weight: 750; margin-top: 8px; overflow-wrap: anywhere; } |
| .card-note { color: var(--muted); font-size: 13px; margin-top: 6px; } |
| .nav { |
| display: flex; |
| gap: 4px; |
| flex-wrap: wrap; |
| border-bottom: 1px solid var(--line); |
| padding-left: 4px; |
| } |
| .nav a { |
| border: 1px solid transparent; |
| border-bottom: 0; |
| border-radius: 8px 8px 0 0; |
| background: transparent; |
| color: #4b5563; |
| padding: 9px 12px; |
| font-weight: 650; |
| } |
| .nav a.active { |
| background: var(--panel); |
| border-color: var(--line); |
| color: #111827; |
| margin-bottom: -1px; |
| } |
| .panel { padding: 16px; overflow-x: visible; } |
| .section-title { font-size: 18px; font-weight: 750; margin: 0 0 4px; } |
| .section-note { color: var(--muted); font-size: 13px; margin: 0 0 14px; } |
| .filters { |
| display: grid; |
| grid-template-columns: minmax(150px, 0.8fr) minmax(190px, 1fr) minmax(320px, 2fr) minmax(110px, 0.7fr) auto; |
| gap: 12px; |
| align-items: end; |
| margin: 14px 0; |
| } |
| label { display: grid; gap: 6px; color: #374151; font-size: 13px; font-weight: 650; } |
| label.checkbox-label { align-items: center; grid-template-columns: auto 1fr; gap: 8px; min-height: 38px; } |
| label.checkbox-label input { min-height: auto; width: 16px; height: 16px; padding: 0; } |
| input, select, textarea, button { |
| border: 1px solid var(--line-strong); |
| border-radius: 8px; |
| background: var(--panel); |
| color: var(--text); |
| font: inherit; |
| padding: 9px 11px; |
| min-height: 38px; |
| outline: none; |
| } |
| input:focus, select:focus, textarea:focus { |
| border-color: #fb923c; |
| box-shadow: 0 0 0 3px rgba(251, 146, 60, 0.18); |
| } |
| textarea { width: 100%; min-height: 260px; font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; font-size: 13px; } |
| input[type="file"] { width: 100%; background: var(--panel-soft); } |
| button { |
| cursor: pointer; |
| font-weight: 700; |
| background: var(--panel-soft); |
| color: #111827; |
| } |
| button:hover { background: #eef0f4; } |
| .status { border: 1px solid var(--line); border-radius: 8px; background: var(--panel-soft); padding: 10px 12px; margin-bottom: 12px; color: #374151; } |
| .submit-grid { display: grid; gap: 14px; } |
| .submit-actions { display: flex; gap: 10px; flex-wrap: wrap; align-items: end; } |
| .file-control { flex: 1 1 320px; } |
| .table-scroll { |
| width: 100%; |
| max-height: 720px; |
| overflow: auto; |
| border: 1px solid var(--line); |
| border-radius: 8px; |
| background: var(--panel); |
| } |
| table { border-collapse: separate; border-spacing: 0; min-width: 2600px; width: max-content; table-layout: fixed; font-size: 13px; } |
| th, td { |
| border-right: 1px solid var(--line); |
| border-bottom: 1px solid #eef0f3; |
| padding: 9px 10px; |
| background: var(--panel); |
| color: #1f2937; |
| vertical-align: middle; |
| overflow-wrap: anywhere; |
| } |
| th { |
| position: sticky; |
| top: 0; |
| z-index: 4; |
| background: #f6f7f9; |
| color: #374151; |
| font-weight: 750; |
| white-space: normal; |
| line-height: 1.15; |
| } |
| td { height: 46px; font-weight: 600; } |
| tr:nth-child(even) td { background: #fcfcfd; } |
| .col-rank { width: 58px; min-width: 58px; max-width: 58px; text-align: center; } |
| .col-model { width: 260px; min-width: 260px; max-width: 260px; } |
| .col-average { width: 104px; min-width: 104px; max-width: 104px; text-align: right; } |
| .col-meta { width: 110px; min-width: 110px; max-width: 110px; } |
| .col-model-id { width: 260px; min-width: 260px; max-width: 260px; } |
| .col-dataset { width: 96px; min-width: 96px; max-width: 96px; text-align: right; } |
| .col-date { width: 112px; min-width: 112px; max-width: 112px; } |
| .col-source { width: 220px; min-width: 220px; max-width: 220px; } |
| .sticky-rank, .sticky-model, .sticky-average { position: sticky; z-index: 3; } |
| th.sticky-rank, th.sticky-model, th.sticky-average { z-index: 6; } |
| .sticky-rank { left: 0; } |
| .sticky-model { left: 58px; } |
| .sticky-average { left: 318px; box-shadow: 8px 0 12px rgba(31, 41, 55, 0.06); } |
| .datasets { min-width: 100%; } |
| .datasets th, .datasets td { width: auto; min-width: 120px; } |
| @media (max-width: 900px) { |
| .page { padding: 12px; } |
| .hero { grid-template-columns: 1fr; } |
| h1 { font-size: 30px; } |
| .cards { grid-template-columns: 1fr; } |
| .filters { grid-template-columns: 1fr; } |
| } |
| """ |
|
|
|
|
| def read_json(path: Path, default: Any) -> Any: |
| if not path.exists(): |
| return default |
| with path.open("r", encoding="utf-8") as file: |
| return json.load(file) |
|
|
|
|
| def read_jsonl(path: Path) -> list[dict[str, Any]]: |
| if not path.exists(): |
| return [] |
| records = [] |
| with path.open("r", encoding="utf-8") as file: |
| for line in file: |
| line = line.strip() |
| if line: |
| records.append(json.loads(line)) |
| return records |
|
|
|
|
| def metric_value(metrics: dict[str, Any], metric: str) -> float | None: |
| value = metrics.get(metric) |
| if value is None: |
| return None |
| try: |
| return float(value) |
| except (TypeError, ValueError): |
| return None |
|
|
|
|
| def load_datasets() -> list[dict[str, Any]]: |
| return read_json(DATASETS_PATH, []) |
|
|
|
|
| def load_results() -> list[dict[str, Any]]: |
| return read_jsonl(RESULTS_PATH) |
|
|
|
|
| def logo_html() -> str: |
| if not LOGO_PATH.exists(): |
| return "" |
| data = base64.b64encode(LOGO_PATH.read_bytes()).decode("ascii") |
| return f'<img class="logo" src="data:image/png;base64,{data}" alt="RusBEIR logo">' |
|
|
|
|
| def dataset_names_for_task(task_filter: str) -> set[str]: |
| datasets = load_datasets() |
| if task_filter != "All": |
| datasets = [dataset for dataset in datasets if dataset.get("task") == task_filter] |
| return {dataset["name"] for dataset in datasets if dataset.get("official", True)} |
|
|
|
|
| def compute_average(record: dict[str, Any], metric: str, dataset_names: set[str]) -> float | None: |
| scores = record.get("scores", {}) |
| explicit = metric_value(scores.get("average", {}), metric) |
| if explicit is not None: |
| return explicit |
| values = [] |
| for dataset_name, dataset_metrics in scores.get("datasets", {}).items(): |
| if dataset_name in dataset_names: |
| value = metric_value(dataset_metrics, metric) |
| if value is not None: |
| values.append(value) |
| return sum(values) / len(values) if values else None |
|
|
|
|
| def leaderboard_rows(metric: str, task_filter: str, verified_only: bool, query: str) -> tuple[list[dict[str, Any]], set[str]]: |
| dataset_names = dataset_names_for_task(task_filter) |
| rows = [] |
| for record in load_results(): |
| if verified_only and not record.get("verified", False): |
| continue |
| model_text = f"{record.get('model_id', '')} {record.get('model_name', '')}".lower() |
| if query and query.lower() not in model_text: |
| continue |
| dataset_scores = record.get("scores", {}).get("datasets", {}) |
| row = { |
| "Rank": None, |
| "Model": record.get("model_name") or record.get("model_id"), |
| metric: compute_average(record, metric, dataset_names), |
| "Model ID": record.get("model_id", ""), |
| "Organization": record.get("organization", ""), |
| "Type": record.get("type", ""), |
| "Verified": "yes" if record.get("verified", False) else "no", |
| "Date": record.get("date", ""), |
| "Source URL": record.get("source_url", ""), |
| } |
| for dataset_name in sorted(dataset_names): |
| row[dataset_name] = metric_value(dataset_scores.get(dataset_name, {}), metric) |
| rows.append(row) |
| rows.sort(key=lambda row: row[metric] if row[metric] is not None else -1, reverse=True) |
| for index, row in enumerate(rows, start=1): |
| row["Rank"] = index |
| return rows, dataset_names |
|
|
|
|
| def format_score(value: Any) -> str: |
| if value is None: |
| return "" |
| try: |
| return f"{float(value):.4f}" |
| except (TypeError, ValueError): |
| return escape(str(value)) |
|
|
|
|
| def column_class(column: str, metric: str, index: int) -> str: |
| if index == 0: |
| return "col-rank sticky-rank" |
| if index == 1: |
| return "col-model sticky-model" |
| if column == metric: |
| return "col-average sticky-average" |
| if column == "Model ID": |
| return "col-model-id" |
| if column in {"Organization", "Type", "Verified"}: |
| return "col-meta" |
| if column == "Date": |
| return "col-date" |
| if column == "Source URL": |
| return "col-source" |
| return "col-dataset" |
|
|
|
|
| def render_table(metric: str, task_filter: str, verified_only: bool, query: str) -> str: |
| rows, dataset_names = leaderboard_rows(metric, task_filter, verified_only, query) |
| columns = [ |
| "Rank", |
| "Model", |
| metric, |
| "Model ID", |
| "Organization", |
| "Type", |
| "Verified", |
| *sorted(dataset_names), |
| *TRAILING_COLUMNS, |
| ] |
| headers = "".join( |
| f'<th class="{column_class(column, metric, index)}">{DISPLAY_COLUMN_NAMES.get(column, escape(column))}</th>' |
| for index, column in enumerate(columns) |
| ) |
| body = [] |
| for row in rows: |
| cells = [] |
| for index, column in enumerate(columns): |
| value = row.get(column, "") |
| if column == "Source URL" and value: |
| cell = f'<a href="{escape(str(value), quote=True)}" target="_blank" rel="noopener noreferrer">source</a>' |
| elif column == "Rank" or column in {"Model", "Model ID", "Organization", "Type", "Verified", "Date"}: |
| cell = escape(str(value)) |
| else: |
| cell = format_score(value) |
| cells.append(f'<td class="{column_class(column, metric, index)}">{cell}</td>') |
| body.append(f"<tr>{''.join(cells)}</tr>") |
| return f'<div class="table-scroll"><table><thead><tr>{headers}</tr></thead><tbody>{"".join(body)}</tbody></table></div>' |
|
|
|
|
| def normalize_submission_record(record: dict[str, Any]) -> dict[str, Any]: |
| model_id = str(record.get("model_id", "")).strip() |
| if not model_id: |
| raise ValueError("`model_id` is required.") |
| scores = record.get("scores") |
| if not isinstance(scores, dict): |
| raise ValueError("`scores` must be an object.") |
| average_scores = scores.get("average") or {} |
| dataset_scores = scores.get("datasets") or {} |
| if not isinstance(average_scores, dict) or not isinstance(dataset_scores, dict): |
| raise ValueError("`scores.average` and `scores.datasets` must be objects.") |
| has_metric = any(metric_value(average_scores, metric) is not None for metric in METRICS) |
| if not has_metric: |
| has_metric = any( |
| isinstance(metrics, dict) and any(metric_value(metrics, metric) is not None for metric in METRICS) |
| for metrics in dataset_scores.values() |
| ) |
| if not has_metric: |
| raise ValueError(f"At least one numeric metric is required: {', '.join(METRICS)}.") |
| normalized = dict(record) |
| normalized["model_id"] = model_id |
| normalized["model_name"] = str(record.get("model_name") or model_id.split("/")[-1]).strip() |
| normalized["organization"] = str(record.get("organization") or (model_id.split("/", 1)[0] if "/" in model_id else "")).strip() |
| normalized["type"] = str(record.get("type") or "dense").strip() |
| normalized["date"] = str(record.get("date") or date.today().isoformat()).strip() |
| normalized["verified"] = bool(record.get("verified", False)) |
| normalized["source_url"] = str(record.get("source_url", "")).strip() |
| normalized["scores"] = {"average": average_scores, "datasets": dataset_scores} |
| return normalized |
|
|
|
|
| def add_submission(record_text: str) -> str: |
| parsed = json.loads(record_text) |
| if not isinstance(parsed, dict): |
| raise ValueError("Submission must be a single JSON object.") |
| record = normalize_submission_record(parsed) |
| serialized = json.dumps(record, ensure_ascii=False, sort_keys=True, separators=(",", ":")) |
| existing = { |
| json.dumps(normalize_submission_record(item), ensure_ascii=False, sort_keys=True, separators=(",", ":")) |
| for item in load_results() |
| } |
| if serialized in existing: |
| return f"`{record['model_id']}` is already present with the same scores." |
| RESULTS_PATH.parent.mkdir(parents=True, exist_ok=True) |
| needs_newline = RESULTS_PATH.exists() and RESULTS_PATH.stat().st_size > 0 |
| with RESULTS_PATH.open("a", encoding="utf-8") as file: |
| if needs_newline: |
| with RESULTS_PATH.open("rb") as check_file: |
| check_file.seek(-1, os.SEEK_END) |
| if check_file.read(1) != b"\n": |
| file.write("\n") |
| file.write(serialized) |
| return f"Added `{record['model_id']}` to `{RESULTS_PATH.name}`." |
|
|
|
|
| def add_submissions(record_text: str) -> str: |
| record_text = record_text.strip() |
| if not record_text: |
| raise ValueError("Paste JSONL content or upload a JSONL file first.") |
|
|
| if record_text.startswith("["): |
| parsed = json.loads(record_text) |
| if not isinstance(parsed, list): |
| raise ValueError("JSON array submission must contain result objects.") |
| lines = [json.dumps(item, ensure_ascii=False) for item in parsed] |
| else: |
| lines = [line.strip() for line in record_text.splitlines() if line.strip()] |
|
|
| added = 0 |
| skipped = 0 |
| messages = [] |
| for line_no, line in enumerate(lines, start=1): |
| try: |
| message = add_submission(line) |
| except (json.JSONDecodeError, ValueError) as exc: |
| raise ValueError(f"Line {line_no}: {exc}") from exc |
| if "already present" in message: |
| skipped += 1 |
| else: |
| added += 1 |
| messages.append(message) |
|
|
| summary = f"Added {added} record(s)" |
| if skipped: |
| summary += f"; skipped {skipped} duplicate(s)" |
| if len(messages) == 1: |
| return messages[0] |
| return summary + "." |
|
|
|
|
| def parse_multipart_form(body: bytes, content_type: str) -> tuple[dict[str, str], dict[str, tuple[str, str]]]: |
| boundary_match = re.search(r'boundary="?([^";]+)"?', content_type) |
| if not boundary_match: |
| raise ValueError("Missing multipart boundary.") |
|
|
| boundary = ("--" + boundary_match.group(1)).encode("utf-8") |
| fields: dict[str, str] = {} |
| files: dict[str, tuple[str, str]] = {} |
|
|
| for part in body.split(boundary): |
| part = part.strip() |
| if not part or part == b"--": |
| continue |
| if part.endswith(b"--"): |
| part = part[:-2].rstrip() |
| header_blob, separator, value = part.partition(b"\r\n\r\n") |
| if not separator: |
| continue |
| headers = header_blob.decode("utf-8", errors="replace") |
| value = value.rstrip(b"\r\n") |
| disposition = next((line for line in headers.splitlines() if line.lower().startswith("content-disposition:")), "") |
| name_match = re.search(r'name="([^"]+)"', disposition) |
| if not name_match: |
| continue |
| name = name_match.group(1) |
| filename_match = re.search(r'filename="([^"]*)"', disposition) |
| text = value.decode("utf-8-sig", errors="replace") |
| if filename_match and filename_match.group(1): |
| files[name] = (filename_match.group(1), text) |
| else: |
| fields[name] = text |
| return fields, files |
|
|
|
|
| def render_filters(metric: str, task: str, verified: bool, query: str) -> str: |
| tasks = ["All", *sorted({dataset["task"] for dataset in load_datasets() if dataset.get("official", True)})] |
| metric_options = "".join(f'<option value="{escape(item)}"{" selected" if item == metric else ""}>{escape(item)}</option>' for item in METRICS) |
| task_options = "".join(f'<option value="{escape(item)}"{" selected" if item == task else ""}>{escape(item)}</option>' for item in tasks) |
| checked = " checked" if verified else "" |
| return f""" |
| <form class="filters" method="get" action="/"> |
| <label>Metric<select name="metric">{metric_options}</select></label> |
| <label>Task<select name="task">{task_options}</select></label> |
| <label>Model<input name="q" value="{escape(query, quote=True)}" placeholder="intfloat, BGE, FRIDA..."></label> |
| <label class="checkbox-label"><input type="checkbox" name="verified" value="1"{checked}>Verified</label> |
| <button type="submit">Apply</button> |
| </form> |
| """ |
|
|
|
|
| def render_summary(metric: str) -> str: |
| datasets = [dataset for dataset in load_datasets() if dataset.get("official", True)] |
| records = load_results() |
| rows, _ = leaderboard_rows(metric, "All", False, "") |
| best_model = rows[0]["Model"] if rows else "No results yet" |
| best_score = f"{rows[0][metric] * 100:.2f}" if rows and rows[0].get(metric) is not None else "n/a" |
| types = sorted({str(record.get("type", "")).strip() for record in records if record.get("type")}) |
| type_text = ", ".join(types) if types else "n/a" |
| return f""" |
| <section class="hero"> |
| <div> |
| <div class="kicker">Russian Information Retrieval Benchmark</div> |
| <h1>RusBEIR Leaderboard</h1> |
| <p class="subtitle">Compare dense retrievers, sparse baselines, and reranker pipelines on official RusBEIR datasets. The default ranking is the macro-average of <strong>{escape(metric)}</strong>.</p> |
| <div class="badges"> |
| <span class="badge">Metric: {escape(metric)}</span> |
| <span class="badge">Official datasets: {len(datasets)}</span> |
| <span class="badge">Rows: {len(records)}</span> |
| <span class="badge">Types: {escape(type_text)}</span> |
| </div> |
| </div> |
| {logo_html()} |
| </section> |
| <section class="cards"> |
| <div class="card"><div class="card-label">Best Model</div><div class="card-value">{escape(str(best_model))}</div><div class="card-note">Highest average {escape(metric)}</div></div> |
| <div class="card"><div class="card-label">Best Score</div><div class="card-value">{best_score}</div><div class="card-note">Shown as percentage points</div></div> |
| <div class="card"><div class="card-label">Models</div><div class="card-value">{len(records)}</div><div class="card-note">Imported and reviewable JSONL rows</div></div> |
| <div class="card"><div class="card-label">Datasets</div><div class="card-value">{len(datasets)}</div><div class="card-note">Official benchmark tasks</div></div> |
| </section> |
| """ |
|
|
|
|
| def render_datasets() -> str: |
| rows = [] |
| for dataset in load_datasets(): |
| if not dataset.get("official", True): |
| continue |
| rows.append( |
| "<tr>" |
| f"<td>{escape(str(dataset.get('name', '')))}</td>" |
| f"<td>{escape(str(dataset.get('task', '')))}</td>" |
| f"<td>{escape(str(dataset.get('split', '')))}</td>" |
| f"<td>{escape(str(dataset.get('hf_repo', '')))}</td>" |
| f"<td>{escape(str(dataset.get('qrels_repo', '')))}</td>" |
| f"<td>{escape(str(dataset.get('origin', '')))}</td>" |
| "</tr>" |
| ) |
| return f""" |
| <table class="datasets"> |
| <thead><tr><th>Dataset</th><th>Task</th><th>Split</th><th>Corpus repo</th><th>Qrels repo</th><th>Origin</th></tr></thead> |
| <tbody>{''.join(rows)}</tbody> |
| </table> |
| """ |
|
|
|
|
| def render_nav(active_page: str) -> str: |
| items = [ |
| ("leaderboard", "/", "Leaderboard"), |
| ("datasets", "/datasets", "Datasets"), |
| ("submit", "/submit", "Submit"), |
| ("about", "/about", "About"), |
| ] |
| links = [] |
| for page, href, label in items: |
| active = " active" if page == active_page else "" |
| links.append(f'<a class="{active.strip()}" href="{href}">{label}</a>') |
| return f'<nav class="nav">{"".join(links)}</nav>' |
|
|
|
|
| def render_page_content(page: str, metric: str, task: str, verified: bool, query: str, query_string: str) -> str: |
| if page == "datasets": |
| return f""" |
| <section class="panel"> |
| <h2 class="section-title">Official Datasets</h2> |
| <p class="section-note">RusBEIR tasks used for the default macro-average ranking.</p> |
| <div class="table-scroll">{render_datasets()}</div> |
| </section> |
| """ |
|
|
| if page == "submit": |
| return f""" |
| <section class="panel"> |
| <h2 class="section-title">Submit Results</h2> |
| <p class="section-note">Paste JSONL content or upload a ready JSONL file. Each record must include model_id, scores, and at least one numeric metric.</p> |
| <form class="submit-grid" method="post" action="/submit?{query_string}" enctype="multipart/form-data"> |
| <label>JSONL record or records<textarea name="record" placeholder='{{"model_id":"org/model","scores":{{"average":{{"NDCG@10":0.5}},"datasets":{{}}}}}}'></textarea></label> |
| <div class="submit-actions"> |
| <label class="file-control">JSONL file<input type="file" name="results_file" accept=".jsonl,.json,application/json,application/x-ndjson"></label> |
| <button type="submit">Add to leaderboard</button> |
| </div> |
| </form> |
| </section> |
| """ |
|
|
| if page == "about": |
| return """ |
| <section class="panel"> |
| <h2 class="section-title">About RusBEIR</h2> |
| <p class="section-note">RusBEIR is a Russian BEIR-style benchmark for zero-shot information retrieval. The leaderboard is backed by a plain JSONL file.</p> |
| <p class="section-note">Verified rows should point to reproducible logs or a commit with generated retrieval results.</p> |
| </section> |
| """ |
|
|
| return f""" |
| <section class="panel"> |
| <h2 class="section-title">Model Rankings</h2> |
| <p class="section-note">Filter by task family, model name, or verification status. Scores are stored as fractions.</p> |
| {render_filters(metric, task, verified, query)} |
| {render_table(metric, task, verified, query)} |
| </section> |
| """ |
|
|
|
|
| def render_page(params: dict[str, list[str]], page: str = "leaderboard", status: str = "") -> str: |
| metric = params.get("metric", [DEFAULT_METRIC])[0] |
| if metric not in METRICS: |
| metric = DEFAULT_METRIC |
| task = params.get("task", ["All"])[0] |
| verified = params.get("verified", [""])[0] == "1" |
| query = params.get("q", [""])[0].strip() |
| status_html = f'<div class="status">{escape(status)}</div>' if status else "" |
| query_string = urlencode({"metric": metric, "task": task, "q": query, "verified": "1" if verified else ""}) |
| active_page = page if page in {"leaderboard", "datasets", "submit", "about"} else "leaderboard" |
| return f"""<!doctype html> |
| <html lang="en"> |
| <head> |
| <meta charset="utf-8"> |
| <meta name="viewport" content="width=device-width, initial-scale=1"> |
| <title>RusBEIR Leaderboard</title> |
| <style>{CSS}</style> |
| </head> |
| <body> |
| <main class="page"> |
| <div class="shell"> |
| {render_summary(metric)} |
| {render_nav(active_page)} |
| {status_html} |
| {render_page_content(active_page, metric, task, verified, query, query_string)} |
| </div> |
| </main> |
| </body> |
| </html>""" |
|
|
|
|
| class Handler(BaseHTTPRequestHandler): |
| def send_html(self, html: str, status: HTTPStatus = HTTPStatus.OK) -> None: |
| body = html.encode("utf-8") |
| self.send_response(status) |
| self.send_header("Content-Type", "text/html; charset=utf-8") |
| self.send_header("Content-Length", str(len(body))) |
| self.end_headers() |
| self.wfile.write(body) |
|
|
| def do_GET(self) -> None: |
| parsed = urlparse(self.path) |
| pages = { |
| "/": "leaderboard", |
| "/index.html": "leaderboard", |
| "/datasets": "datasets", |
| "/submit": "submit", |
| "/about": "about", |
| } |
| if parsed.path not in pages: |
| self.send_error(HTTPStatus.NOT_FOUND) |
| return |
| self.send_html(render_page(parse_qs(parsed.query), page=pages[parsed.path])) |
|
|
| def do_POST(self) -> None: |
| parsed = urlparse(self.path) |
| if parsed.path != "/submit": |
| self.send_error(HTTPStatus.NOT_FOUND) |
| return |
| length = int(self.headers.get("Content-Length", "0")) |
| content_type = self.headers.get("Content-Type", "") |
| body = self.rfile.read(length) |
| try: |
| if content_type.startswith("multipart/form-data"): |
| fields, files = parse_multipart_form(body, content_type) |
| parts = [] |
| if fields.get("record", "").strip(): |
| parts.append(fields["record"].strip()) |
| if "results_file" in files: |
| filename, file_text = files["results_file"] |
| if file_text.strip(): |
| parts.append(file_text.strip()) |
| elif filename: |
| raise ValueError(f"`{filename}` is empty.") |
| status = add_submissions("\n".join(parts)) |
| else: |
| form = parse_qs(body.decode("utf-8")) |
| status = add_submissions(form.get("record", [""])[0].strip()) |
| except (json.JSONDecodeError, ValueError) as exc: |
| status = f"Submission was not added: {exc}" |
| self.send_html(render_page(parse_qs(parsed.query), page="submit", status=status)) |
|
|
| def log_message(self, format: str, *args: Any) -> None: |
| print(f"{self.address_string()} - {format % args}") |
|
|
|
|
| if __name__ == "__main__": |
| server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler) |
| print(f"Serving RusBEIR leaderboard on 0.0.0.0:{PORT}") |
| server.serve_forever() |
|
|
