feat: initial commit of arena-leaderboard-parser & omniroute benchmark tools
This commit is contained in:
@@ -0,0 +1,496 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Arena.ai Leaderboard Parser
|
||||
Парсер лидербордов arena.ai: подаёшь модель — получаешь рейтинг.
|
||||
Кеш на 24 часа.
|
||||
|
||||
Разделы:
|
||||
text — text/overall: Arena Score (ELO), голоса, цена, контекст (413 моделей)
|
||||
agent — agent/overall: tool-calling метрики (Net Improvement, Bash Recovery,
|
||||
Tool Hallucination, Steerability, Cost/Task) — 51 модель
|
||||
|
||||
Standalone: только stdlib (urllib + re + json).
|
||||
|
||||
CLI:
|
||||
python3 arena_leaderboard.py # text: топ-10
|
||||
python3 arena_leaderboard.py glm-5.3-max # text: рейтинг модели
|
||||
python3 arena_leaderboard.py --board agent # agent: топ-10
|
||||
python3 arena_leaderboard.py --board agent claude # agent: модель/поиск
|
||||
python3 arena_leaderboard.py --refresh glm # игнорировать кеш
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
BOARDS = {
|
||||
"text": "https://arena.ai/leaderboard/text/overall",
|
||||
"agent": "https://arena.ai/leaderboard/agent",
|
||||
}
|
||||
CACHE_DIR = Path("/opt/data/cache")
|
||||
CACHE_TTL = 86400 # 24h
|
||||
|
||||
HEADERS = {
|
||||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||||
"(KHTML, like Gecko) Chrome/131.0 Safari/537.36",
|
||||
"Accept": "text/html,application/xhtml+xml",
|
||||
"Accept-Language": "en-US,en;q=0.9",
|
||||
}
|
||||
|
||||
|
||||
def _cache_file(board: str) -> Path:
|
||||
return CACHE_DIR / f"arena_leaderboard_{board}.json"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- fetching
|
||||
|
||||
def _fetch_content(board: str) -> tuple:
|
||||
"""Получает контент страницы (markdown/html).
|
||||
Сначала пробует firecrawl (даёт полный Markdown таблицы на 416 строк),
|
||||
затем failover-кольцо Hermes (обход Cloudflare WAF).
|
||||
При сбое — fallback на прямой urllib запрос.
|
||||
"""
|
||||
url = BOARDS[board]
|
||||
# 1. Попытка через Firecrawl / Keyless ring Hermes
|
||||
try:
|
||||
if "/opt/hermes" not in sys.path:
|
||||
sys.path.insert(0, "/opt/hermes")
|
||||
from plugins.web.keyless_mcp import firecrawl_extract_keyless, extract_with_failover
|
||||
# Firecrawl стабильно рендерит полную markdown таблицу (68KB, 416 моделей)
|
||||
try:
|
||||
res = firecrawl_extract_keyless([url])
|
||||
if res and not res[0].get("error"):
|
||||
c = res[0].get("content", "")
|
||||
if c and len(c) > 20000:
|
||||
return "markdown", c
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
res = extract_with_failover("firecrawl", [url])
|
||||
if res and not res[0].get("error"):
|
||||
c = res[0].get("content", "")
|
||||
if c and len(c) > 500:
|
||||
return "markdown", c
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 2. Fallback на прямой urllib
|
||||
try:
|
||||
req = urllib.request.Request(url, headers=HEADERS)
|
||||
with urllib.request.urlopen(req, timeout=30) as r:
|
||||
html = r.read().decode("utf-8", errors="replace")
|
||||
return "html", html
|
||||
except Exception as e:
|
||||
raise RuntimeError(f"Не удалось получить {url} (Cloudflare block?): {e}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- parsing helpers
|
||||
|
||||
def _strip_tags(s: str) -> str:
|
||||
return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", s)).strip()
|
||||
|
||||
|
||||
def _pct(cell_html: str) -> str:
|
||||
"""Процентная ячейка agent-борда: '13.82%' + '±2.17%' -> '13.82%±2.17%'."""
|
||||
txt = re.sub(r"<!-- -->", "", cell_html)
|
||||
m = re.search(r"(-?\d+(?:\.\d+)?)\s*%", txt)
|
||||
if not m:
|
||||
return ""
|
||||
out = f"{m.group(1)}%"
|
||||
ci = re.search(r"±\s*(\d+(?:\.\d+)?)\s*%", txt)
|
||||
if ci:
|
||||
out += f"±{ci.group(1)}%"
|
||||
return out
|
||||
|
||||
|
||||
def _model_from_cell(model_cell: str) -> tuple:
|
||||
"""(name, provider) из ячейки модели."""
|
||||
mm = (re.search(r'<span[^>]*title="([^"]+)"', model_cell)
|
||||
or re.search(r"<a[^>]*>(.*?)</a>", model_cell, re.S))
|
||||
if not mm:
|
||||
return "", ""
|
||||
name = mm.group(1)
|
||||
if "<" in name:
|
||||
name = _strip_tags(name)
|
||||
pm = re.search(r'<span[^>]*class="[^"]*text-text-secondary[^"]*"[^>]*>(.*?)</span>',
|
||||
model_cell, re.S)
|
||||
provider = _strip_tags(pm.group(1)) if pm else ""
|
||||
return name.strip(), provider
|
||||
|
||||
|
||||
def _parse_text_rows(html: str) -> list:
|
||||
"""text/overall: Rank | Rank Spread | Model | Score | Votes | Price | Context."""
|
||||
start = html.find("</thead>")
|
||||
if start == -1:
|
||||
raise ValueError("Таблица text не найдена (нет </thead>)")
|
||||
|
||||
tr_re = re.compile(r"<tr\b[^>]*>(.*?)</tr>", re.S)
|
||||
td_re = re.compile(r"<td\b[^>]*>(.*?)</td>", re.S)
|
||||
rows = []
|
||||
|
||||
for tr_html in tr_re.findall(html[start:]):
|
||||
cells = td_re.findall(tr_html)
|
||||
if len(cells) < 6:
|
||||
continue
|
||||
|
||||
rm = re.search(r">\s*(\d+)\s*<", cells[0])
|
||||
if not rm:
|
||||
continue
|
||||
|
||||
model, provider = _model_from_cell(cells[2])
|
||||
if not model:
|
||||
continue
|
||||
|
||||
score_cell = cells[3]
|
||||
sm = re.search(r">\s*(\d{3,4})\s*<", score_cell)
|
||||
score = sm.group(1) if sm else ""
|
||||
if score:
|
||||
ci = re.search(r"±\s*(\d+)", score_cell)
|
||||
if ci:
|
||||
score += f"±{ci.group(1)}"
|
||||
|
||||
rows.append({
|
||||
"rank": int(rm.group(1)),
|
||||
"model": model,
|
||||
"provider": provider,
|
||||
"score": score,
|
||||
"preliminary": "Preliminary" in score_cell,
|
||||
"rank_spread": " ".join(re.findall(r">\s*(\d+)\s*<", cells[1])),
|
||||
"votes": int(re.sub(r"[^\d]", "", _strip_tags(cells[4])) or 0),
|
||||
"price": _strip_tags(cells[5]),
|
||||
"context": _strip_tags(cells[6]) if len(cells) > 6 else "",
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
def _parse_agent_rows(html: str) -> list:
|
||||
"""agent/overall: Rank | Model | Net Improvement | Confirmed Success |
|
||||
Praise vs Complaint | Steerability | Bash Recovery | Tool Hallucination |
|
||||
Sessions | Cost/Task | Tokens/Task | Price."""
|
||||
start = html.find("</thead>")
|
||||
if start == -1:
|
||||
raise ValueError("Таблица agent не найдена (нет </thead>)")
|
||||
|
||||
tr_re = re.compile(r"<tr\b[^>]*>(.*?)</tr>", re.S)
|
||||
td_re = re.compile(r"<td\b[^>]*>(.*?)</td>", re.S)
|
||||
rows = []
|
||||
|
||||
for tr_html in tr_re.findall(html[start:]):
|
||||
cells = td_re.findall(tr_html)
|
||||
if len(cells) < 9:
|
||||
continue
|
||||
|
||||
rm = re.search(r">\s*(\d+)\s*<", cells[0])
|
||||
if not rm:
|
||||
continue
|
||||
|
||||
model, provider = _model_from_cell(cells[1])
|
||||
if not model:
|
||||
continue
|
||||
|
||||
def pct(i):
|
||||
return _pct(cells[i]) if len(cells) > i else ""
|
||||
|
||||
def plain(i):
|
||||
return _strip_tags(re.sub(r"<!-- -->", "", cells[i])) if len(cells) > i else ""
|
||||
|
||||
rows.append({
|
||||
"rank": int(rm.group(1)),
|
||||
"model": model,
|
||||
"provider": provider,
|
||||
"net_improvement": pct(2),
|
||||
"confirmed_success": pct(3),
|
||||
"praise_vs_complaint": pct(4),
|
||||
"steerability": pct(5),
|
||||
"bash_recovery": pct(6),
|
||||
"tool_hallucination": pct(7),
|
||||
"sessions": int(re.sub(r"[^\d]", "", plain(8)) or 0),
|
||||
"cost_per_task": plain(9),
|
||||
"output_tokens_per_task": plain(10),
|
||||
"price": plain(11),
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
def _parse_text_markdown(md: str) -> list:
|
||||
"""Парсинг text/overall из Markdown таблицы."""
|
||||
rows = []
|
||||
for line in md.splitlines():
|
||||
if line.startswith("|") and not line.startswith("| ---") and not line.startswith("| Rank"):
|
||||
parts = [p.strip() for p in line.split("|")[1:-1]]
|
||||
if len(parts) >= 6 and parts[0].isdigit():
|
||||
rank = int(parts[0])
|
||||
rank_spread = parts[1] if len(parts) > 1 else ""
|
||||
raw_model = parts[2]
|
||||
m = re.search(r"\[([^\]]+)\]", raw_model)
|
||||
model_name = m.group(1).strip() if m else raw_model.split("<br>")[0].strip()
|
||||
|
||||
provider = ""
|
||||
if "<br>" in raw_model:
|
||||
p_part = raw_model.split("<br>")[-1].strip()
|
||||
provider = re.sub(r"<[^>]+>", "", p_part).strip()
|
||||
if "[" in provider:
|
||||
provider = raw_model.split("<br>")[0].strip()
|
||||
|
||||
score_cell = parts[3]
|
||||
sm = re.search(r">?\s*(\d{3,4})\s*<?", score_cell)
|
||||
score = sm.group(1) if sm else ""
|
||||
ci = re.search(r"±\s*(\d+)", score_cell)
|
||||
if score and ci:
|
||||
score += f"±{ci.group(1)}"
|
||||
|
||||
votes_clean = re.sub(r"[^\d]", "", parts[4]) if len(parts) > 4 else "0"
|
||||
votes = int(votes_clean) if votes_clean else 0
|
||||
price = parts[5] if len(parts) > 5 else ""
|
||||
context = parts[6] if len(parts) > 6 else ""
|
||||
|
||||
rows.append({
|
||||
"rank": rank,
|
||||
"model": model_name,
|
||||
"provider": provider,
|
||||
"score": score,
|
||||
"preliminary": "Preliminary" in score_cell,
|
||||
"rank_spread": rank_spread,
|
||||
"votes": votes,
|
||||
"price": price,
|
||||
"context": context,
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
def _parse_agent_markdown(md: str) -> list:
|
||||
"""Парсинг agent/overall из Markdown таблицы."""
|
||||
rows = []
|
||||
for line in md.splitlines():
|
||||
if line.startswith("|") and not line.startswith("| ---") and "Rank" not in line:
|
||||
parts = [p.strip() for p in line.split("|")[1:-1]]
|
||||
if len(parts) >= 8:
|
||||
rm = re.search(r"^\d+", parts[0])
|
||||
if not rm:
|
||||
continue
|
||||
rank = int(rm.group(0))
|
||||
raw_model = parts[1]
|
||||
m = re.search(r"\[([^\]]+)\]", raw_model)
|
||||
model_name = m.group(1).strip() if m else raw_model
|
||||
|
||||
provider = ""
|
||||
if "<br>" in raw_model:
|
||||
provider = re.sub(r"<[^>]+>", "", raw_model.split("<br>")[-1]).strip()
|
||||
elif "·" in raw_model:
|
||||
# e.g. 'Claude Opus 5.5 (Max) Anthropic · Proprietary'
|
||||
provider = raw_model.split("(", 1)[-1].split(")", 1)[-1].strip() if ")" in raw_model else ""
|
||||
|
||||
def clean_val(idx):
|
||||
return parts[idx] if len(parts) > idx else ""
|
||||
|
||||
sessions_clean = re.sub(r"[^\d]", "", clean_val(8))
|
||||
sessions = int(sessions_clean) if sessions_clean else 0
|
||||
|
||||
rows.append({
|
||||
"rank": rank,
|
||||
"model": model_name,
|
||||
"provider": provider,
|
||||
"net_improvement": clean_val(2),
|
||||
"confirmed_success": clean_val(3),
|
||||
"praise_vs_complaint": clean_val(4),
|
||||
"steerability": clean_val(5),
|
||||
"bash_recovery": clean_val(6),
|
||||
"tool_hallucination": clean_val(7),
|
||||
"sessions": sessions,
|
||||
"cost_per_task": clean_val(9),
|
||||
"output_tokens_per_task": clean_val(10),
|
||||
"price": clean_val(11),
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
_PARSERS = {"text": _parse_text_rows, "agent": _parse_agent_rows}
|
||||
|
||||
|
||||
def _parse_page(board: str, content_type: str, content: str) -> dict:
|
||||
if content_type == "markdown":
|
||||
if board == "text":
|
||||
models = _parse_text_markdown(content)
|
||||
elif board == "agent":
|
||||
models = _parse_agent_markdown(content)
|
||||
else:
|
||||
models = []
|
||||
else:
|
||||
models = _PARSERS[board](content)
|
||||
|
||||
if not models:
|
||||
raise ValueError(f"Таблица {board} распарсилась пустой — страница изменилась?")
|
||||
|
||||
dm = re.search(r"(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)\s+\d{1,2},\s+\d{4}", content)
|
||||
return {
|
||||
"board": board,
|
||||
"updated": dm.group(0) if dm else "",
|
||||
"models": models,
|
||||
"total_models": len(models),
|
||||
"source_url": BOARDS[board],
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- cache
|
||||
|
||||
def _load_cache(board: str) -> Optional[dict]:
|
||||
f = _cache_file(board)
|
||||
if not f.exists():
|
||||
return None
|
||||
try:
|
||||
with open(f) as fh:
|
||||
data = json.load(fh)
|
||||
if time.time() - data.get("timestamp", 0) < CACHE_TTL:
|
||||
return data
|
||||
except (json.JSONDecodeError, OSError, ValueError):
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def _save_cache(board: str, data: dict) -> None:
|
||||
f = _cache_file(board)
|
||||
f.parent.mkdir(parents=True, exist_ok=True)
|
||||
data["timestamp"] = time.time()
|
||||
data["fetched_at"] = time.strftime("%Y-%m-%d %H:%M:%S UTC", time.gmtime())
|
||||
with open(f, "w") as fh:
|
||||
json.dump(data, fh, ensure_ascii=False, indent=1)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- public API
|
||||
|
||||
def get_leaderboard(board: str = "text", force_refresh: bool = False) -> dict:
|
||||
"""Полный лидерборд ('text' или 'agent'): {'updated', 'models', ...}."""
|
||||
if board not in BOARDS:
|
||||
raise ValueError(f"board должен быть одним из {list(BOARDS)}")
|
||||
if not force_refresh:
|
||||
cached = _load_cache(board)
|
||||
if cached:
|
||||
out = dict(cached)
|
||||
out["cached"] = True
|
||||
return out
|
||||
content_type, content = _fetch_content(board)
|
||||
data = _parse_page(board, content_type, content)
|
||||
_save_cache(board, data)
|
||||
out = dict(data)
|
||||
out["cached"] = False
|
||||
return out
|
||||
|
||||
|
||||
def get_model_rating(model_name: str, board: str = "text",
|
||||
force_refresh: bool = False) -> Optional[dict]:
|
||||
"""Рейтинг модели: точное совпадение, потом подстрока."""
|
||||
lb = get_leaderboard(board, force_refresh)
|
||||
q = model_name.strip().lower()
|
||||
for m in lb["models"]:
|
||||
if m["model"].lower() == q:
|
||||
return {**m, "cached": lb["cached"]}
|
||||
for m in lb["models"]:
|
||||
if q in m["model"].lower():
|
||||
return {**m, "cached": lb["cached"]}
|
||||
return None
|
||||
|
||||
|
||||
def search_models(query: str, board: str = "text",
|
||||
force_refresh: bool = False) -> list:
|
||||
"""Все модели борда, чьё имя содержит подстроку."""
|
||||
lb = get_leaderboard(board, force_refresh)
|
||||
q = query.strip().lower()
|
||||
return [{**m, "cached": lb["cached"]}
|
||||
for m in lb["models"] if q in m["model"].lower()]
|
||||
|
||||
|
||||
def clear_cache(board: Optional[str] = None) -> None:
|
||||
"""Стереть кеш (конкретного борда или весь)."""
|
||||
files = [_cache_file(board)] if board else \
|
||||
[_cache_file(b) for b in BOARDS]
|
||||
for f in files:
|
||||
if f.exists():
|
||||
f.unlink()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- CLI
|
||||
|
||||
def _fmt(m: dict) -> str:
|
||||
if m.get("net_improvement") is not None and "net_improvement" in m:
|
||||
return (f"#{m['rank']:>3} net={m['net_improvement']:<14}"
|
||||
f"succ={m['confirmed_success']:<14}"
|
||||
f"halluc={m['tool_hallucination']:<10}{m['model']}")
|
||||
pre = " (Preliminary)" if m.get("preliminary") else ""
|
||||
return (f"#{m['rank']:>3} {m['score']:<10}{m['model']:<40}"
|
||||
f"votes={m['votes']:>7,} {m['price']:<18}ctx={m['context']}{pre}")
|
||||
|
||||
|
||||
def _print_model(r: dict, board: str) -> None:
|
||||
src = "кеш" if r["cached"] else "свежая выгрузка"
|
||||
print(f"{r['model']} [{r['provider']}] ({src}, board={board})")
|
||||
print(f" Ранг: #{r['rank']}")
|
||||
if board == "agent":
|
||||
print(f" Net Improvement: {r['net_improvement']}")
|
||||
print(f" Confirmed Success: {r['confirmed_success']}")
|
||||
print(f" Praise vs Complaint: {r['praise_vs_complaint']}")
|
||||
print(f" Steerability: {r['steerability']}")
|
||||
print(f" Bash Recovery: {r['bash_recovery']}")
|
||||
print(f" Tool Hallucination: {r['tool_hallucination']}")
|
||||
print(f" Sessions: {r['sessions']:,}")
|
||||
print(f" Cost/Task: {r['cost_per_task']}")
|
||||
print(f" Tokens/Task: {r['output_tokens_per_task']}")
|
||||
print(f" Цена $/M: {r['price']}")
|
||||
else:
|
||||
print(f" Скор: {r['score']}" + (" Preliminary" if r["preliminary"] else ""))
|
||||
print(f" Спред: {r['rank_spread']}")
|
||||
print(f" Голосов: {r['votes']:,}")
|
||||
print(f" Цена $/M: {r['price']}")
|
||||
print(f" Контекст: {r['context']}")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
args = sys.argv[1:]
|
||||
board = "text"
|
||||
refresh = False
|
||||
if "--refresh" in args:
|
||||
refresh = True
|
||||
args.remove("--refresh")
|
||||
if "--board" in args:
|
||||
i = args.index("--board")
|
||||
board = args[i + 1] if len(args) > i + 1 else "text"
|
||||
del args[i:i + 2]
|
||||
if board not in BOARDS:
|
||||
print(f"Неизвестный board '{board}', доступны: {', '.join(BOARDS)}")
|
||||
sys.exit(1)
|
||||
|
||||
if "--search" in args:
|
||||
i = args.index("--search")
|
||||
query = args[i + 1] if len(args) > i + 1 else ""
|
||||
found = search_models(query, board=board, force_refresh=refresh)
|
||||
if not found:
|
||||
print(f"По '{query}' в {board}-лидерборде ничего нет")
|
||||
sys.exit(1)
|
||||
print(f"Найдено {len(found)} по '{query}' (board={board}):")
|
||||
for m in found:
|
||||
print(" " + _fmt(m))
|
||||
return
|
||||
|
||||
if args:
|
||||
name = " ".join(args)
|
||||
r = get_model_rating(name, board=board, force_refresh=refresh)
|
||||
if not r:
|
||||
print(f"'{name}' не найдена в {board}-лидерборде")
|
||||
sys.exit(1)
|
||||
_print_model(r, board)
|
||||
return
|
||||
|
||||
lb = get_leaderboard(board, force_refresh=refresh)
|
||||
src = "кеш" if lb["cached"] else "свежая выгрузка"
|
||||
title = "Agent Arena (tool calling)" if board == "agent" else "Text Overall"
|
||||
print(f"Arena.ai {title} — обновлено {lb['updated'] or '?'} ({src})")
|
||||
print(f"Всего моделей: {lb['total_models']}\n")
|
||||
for m in lb["models"][:10]:
|
||||
print(_fmt(m))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user