From f81b866e3468e057f641319c5ff05700663c9342 Mon Sep 17 00:00:00 2001 From: ai_coder Date: Fri, 16 Oct 2026 12:05:02 +0000 Subject: [PATCH] feat: initial commit of arena-leaderboard-parser & omniroute benchmark tools --- .gitignore | 4 + README.md | 72 ++++++ arena_cache_refresh.py | 28 +++ arena_leaderboard.py | 496 ++++++++++++++++++++++++++++++++++++++ omniroute_arena_report.py | 218 +++++++++++++++++ 5 files changed, 818 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 arena_cache_refresh.py create mode 100644 arena_leaderboard.py create mode 100644 omniroute_arena_report.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..500e5d4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +__pycache__/ +*.pyc +*.pyo +.cache/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..a5b7fbb --- /dev/null +++ b/README.md @@ -0,0 +1,72 @@ +# Arena.ai Leaderboard Parser & OmniRoute Benchmark + +Парсер лидерборда [arena.ai](https://arena.ai/) (LMSYS Chatbot Arena) с 24-часовым кешированием и обходом Cloudflare WAF, а также инструмент бенчмаркинга и матчинга моделей со шлюзом **OmniRoute**. + +## 🚀 Возможности + +1. **Обход Cloudflare WAF / Anti-Bot**: + - Автоматически использует stealth-экстрактор (Firecrawl / keyless failover-кольцо) для гарантированного получения актуальных данных таблицы арены без блокировок HTTP 403. + - Fallback на сырой HTML при отсутствии WAF. + +2. **Поддержка бордов**: + - `text` (416 моделей): Rank, Arena Score (ELO ±CI), Spread, Votes, Price, Context. + - `agent` (54 модели): Tool-calling метрики (Net Improvement, Confirmed Success, Steerability, Bash Recovery, Tool Hallucination, Sessions, Cost/Task). + +3. **Интеграция с OmniRoute (`omniroute_arena_report.py`)**: + - Авторизация через Bearer `OMNIROUTE_API_KEY`. + - Liveness-проверка всех моделей провайдера коротким вызовом («2+2»). + - Интеллектуальный матчинг моделей гейтвея с мировым рейтингом LMSYS Arena (учёт reasoning-суффиксов `-high`, `-medium`, `-low`, `-thinking`). + - Формирование наглядного отчёта в Markdown. + +--- + +## 📦 Структура репозитория + +- `arena_leaderboard.py` — основной CLI и Python API парсера лидерборда. +- `omniroute_arena_report.py` — прогон моделей OmniRoute и генерация арена-отчёта. +- `arena_cache_refresh.py` — легковесный скрипт для запуска через cron (обновление кеша раз в сутки). + +--- + +## 🛠 Использование + +### CLI + +```bash +# Топ-10 моделей текстового лидерборда +python3 arena_leaderboard.py + +# Рейтинг конкретной модели +python3 arena_leaderboard.py glm-5.3-max + +# Поиск моделей по подстроке +python3 arena_leaderboard.py --search gemini + +# Принудительное обновление кеша мимо TTL +python3 arena_leaderboard.py --refresh + +# Агентный лидерборд (tool-calling) +python3 arena_leaderboard.py --board agent +python3 arena_leaderboard.py --board agent "Claude Opus 5.5 (Max)" +``` + +### OmniRoute Liveness & Arena Match Report + +```bash +# Проверить всех провайдеров / конкретного провайдера (например, agy, openrouter, zai) +python3 omniroute_arena_report.py --provider agy +python3 omniroute_arena_report.py --provider zai +``` + +### Python API + +```python +from arena_leaderboard import get_leaderboard, get_model_rating, search_models + +# Получить рейтинг модели +rating = get_model_rating("glm-5.3-max") +print(rating) + +# Поиск +gemini_models = search_models("gemini-3.8") +``` diff --git a/arena_cache_refresh.py b/arena_cache_refresh.py new file mode 100644 index 0000000..9bb264c --- /dev/null +++ b/arena_cache_refresh.py @@ -0,0 +1,28 @@ +#!/usr/bin/env python3 +""" +arena_cache_refresh.py — cron no_agent job. +Раз в сутки обновляет кеш лидерборда arena.ai (24h TTL). +Silent watchdog: успех = одна короткая строка в лог крона, в чат ничего не летит. +""" +import sys +import traceback + +sys.path.insert(0, "/opt/data/skills/research/arena-leaderboard-parser") + +try: + from arena_leaderboard import get_leaderboard +except ImportError: + print("FATAL: cannot import arena_leaderboard from " + "/opt/data/skills/research/arena-leaderboard-parser") + sys.exit(1) + +try: + lb = get_leaderboard(force_refresh=True) + top1 = lb["models"][0] + print(f"ok: refreshed {lb['total_models']} models (updated {lb['updated']}); " + f"#{top1['rank']} {top1['model']} {top1['score']}; board=text") +except Exception as e: + print(f"FAIL: {type(e).__name__}: {e}") + traceback.print_exc() + # не валимся молча — код 1, чтобы крон пометил run как error + sys.exit(1) diff --git a/arena_leaderboard.py b/arena_leaderboard.py new file mode 100644 index 0000000..021c7a5 --- /dev/null +++ b/arena_leaderboard.py @@ -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']*title="([^"]+)"', model_cell) + or re.search(r"]*>(.*?)", model_cell, re.S)) + if not mm: + return "", "" + name = mm.group(1) + if "<" in name: + name = _strip_tags(name) + pm = re.search(r']*class="[^"]*text-text-secondary[^"]*"[^>]*>(.*?)', + 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("") + if start == -1: + raise ValueError("Таблица text не найдена (нет )") + + tr_re = re.compile(r"]*>(.*?)", re.S) + td_re = re.compile(r"]*>(.*?)", 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("") + if start == -1: + raise ValueError("Таблица agent не найдена (нет )") + + tr_re = re.compile(r"]*>(.*?)", re.S) + td_re = re.compile(r"]*>(.*?)", 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("
")[0].strip() + + provider = "" + if "
" in raw_model: + p_part = raw_model.split("
")[-1].strip() + provider = re.sub(r"<[^>]+>", "", p_part).strip() + if "[" in provider: + provider = raw_model.split("
")[0].strip() + + score_cell = parts[3] + sm = re.search(r">?\s*(\d{3,4})\s* 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 "
" in raw_model: + provider = re.sub(r"<[^>]+>", "", raw_model.split("
")[-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() diff --git a/omniroute_arena_report.py b/omniroute_arena_report.py new file mode 100644 index 0000000..f907243 --- /dev/null +++ b/omniroute_arena_report.py @@ -0,0 +1,218 @@ +#!/usr/bin/env python3 +""" +omniroute_arena_report.py — прогон всех LLM на OmniRoute-гейтвее + арена-рейтинги. + +Проверяет каждую openrouter/zai/nvidia/... модель гейтвея мини-чатом (2+2), +затем дополняет ответ рейтингом из arena.ai (через arena_leaderboard). +Отчёт: markdown в stdout. + +Запуск: python3 omniroute_arena_report.py [--provider openrouter] [--timeout 60] +""" + +import argparse +import json +import re +import sys +import time +import urllib.error +import urllib.request + +sys.path.insert(0, "/opt/data/skills/research/arena-leaderboard-parser") +from arena_leaderboard import get_leaderboard, search_models + +GW = "http://192.168.1.101:20128" + + +def gateway_models() -> dict: + """Все модели гейтвея по префиксам провайдеров.""" + headers = {} + env_path = "/opt/data/.env" + try: + with open(env_path) as f: + for line in f: + if line.startswith("OMNIROUTE_API_KEY="): + key = line.strip().split("=", 1)[1] + if key: + headers["Authorization"] = f"Bearer {key}" + break + except Exception: + pass + + req = urllib.request.Request(f"{GW}/v1/models", headers=headers) + data = json.loads(urllib.request.urlopen(req, timeout=15).read()) + out = {} + for m in data["data"]: + mid = m["id"] if isinstance(m, dict) else str(m) + prov = mid.split("/", 1)[0] + out.setdefault(prov, []).append(mid) + return out + + +def chat(model: str, prompt="What is 2+2? Answer with just the number.", + timeout: int = 60) -> tuple: + """Мини-вызов. Возвращает (статус, ответ, сек). stream=True всегда (NIM-quirk).""" + body = json.dumps({"model": model, + "messages": [{"role": "user", "content": prompt}], + "stream": True, "max_tokens": 150}).encode() + headers = {"Content-Type": "application/json"} + try: + with open("/opt/data/.env") as f: + for line in f: + if line.startswith("OMNIROUTE_API_KEY="): + key = line.strip().split("=", 1)[1] + if key: + headers["Authorization"] = f"Bearer {key}" + break + except Exception: + pass + + req = urllib.request.Request(f"{GW}/v1/chat/completions", data=body, + headers=headers) + t0 = time.time() + try: + raw = urllib.request.urlopen(req, timeout=timeout).read().decode("utf-8", errors="replace") + dt = time.time() - t0 + content = "" + if raw.lstrip().startswith("data:"): + for line in raw.splitlines(): + if line.startswith("data:") and "[DONE]" not in line: + try: + j = json.loads(line[5:].strip()) + content += j.get("choices", [{}])[0].get("delta", {}).get("content") or "" + except Exception: + pass + else: + j = json.loads(raw) + content = j.get("choices", [{}])[0].get("message", {}).get("content") or "" + return ("ok" if content.strip() else "empty", content.strip()[:70], round(dt, 1)) + except urllib.error.HTTPError as e: + return (f"HTTP {e.code}", e.read().decode(errors="replace")[:60], round(time.time() - t0, 1)) + except Exception as e: + return ("fail", type(e).__name__, round(time.time() - t0, 1)) + + +def test_with_retry(model: str, attempts: int = 2, timeout: int = 60) -> tuple: + """1 ретрай для transient (429/5xx/empty/fail), не для вечного 403/404.""" + st, ans, dt = chat(model, timeout=timeout) + transient = st in ("empty", "fail") or st in ("HTTP 429", "HTTP 502", "HTTP 503") + if st != "ok" and transient and attempts > 1: + time.sleep(2) + st2, ans2, dt2 = chat(model, timeout=timeout) + return st2, ans2, dt2 + return st, ans, dt + + +# ---- сопоставление с ареной ------------------------------------------------ + +def norm(s: str) -> str: + """Норм-имя: последнее слово модели, только [a-z0-9].""" + s = s.lower() + s = s.split("/")[-1] + s = re.sub(r":[^:]*$", "", s) + s = re.sub(r"[^a-z0-9]", "", s) + return s + + +def norm_keep(s: str) -> str: + """Норм-имя с сохранением дефисов (для containment-сопоставления).""" + s = s.lower() + s = s.split("/")[-1] + s = re.sub(r":[^:]*$", "", s) + return s + + +def arena_lookup(gw_name: str, arena_models: list) -> tuple: + """(arena_dict|None, note). Сопоставление гейтвей-имени с ареной.""" + # 0) Снимаем префикс провайдера (agy/..., openrouter/..., etc.) + raw = gw_name.split("/", 1)[-1].lower() + + # Снимаем суффиксы -it, -preview, reasoning суффиксы (-thinking, -tiered, -low, -medium, -high, -disabled, -mini, -ultra, -max, -xhigh) + base = re.sub(r"-(it|preview)(:[^:]*)?$", "", raw) + base = re.sub(r"(:[^:]*)$", "", base) + + n_full = norm(raw) + n = norm(base) + + # 1) точное по норм-имени + for m in arena_models: + if norm(m["model"]) == n_full: + return m, "exact" + + # 2) без суффиксов (-it, -preview) + for m in arena_models: + if norm(m["model"]) == n: + return m, "suffix" + + # 2.5) отсекаем reasoning / variant суффиксы: -thinking, -low, -medium, -high, -tiered, -disabled + cleaned = re.sub(r"-(thinking|tiered|disabled|low|medium|high|xhigh|max|ultra|mini)", "", base) + cleaned = re.sub(r"-(thinking|tiered|disabled|low|medium|high|xhigh|max|ultra|mini)", "", cleaned) + + # 2.6) пробуем восстановить суффиксы для точного матча (например claude-opus-4-6-thinking-high -> Claude Opus 4.6 (High)) + tier = "" + for t in ("high", "medium", "low", "max", "xhigh"): + if raw.endswith("-" + t): + tier = t + break + if tier: + target_name = f"{cleaned}-{tier}" + for m in arena_models: + if norm(m["model"]) == norm(target_name): + return m, "variant-tier" + + n_cleaned = norm(cleaned) + for m in arena_models: + if norm(m["model"]) == n_cleaned: + return m, "variant" + + # 3) containment: гейтвей-имя (с дефисами) — подстрока/суффикс арен-имени. + # Покрывает вендор-префиксы арены: nemotron-3-super → nvidia-nemotron-3-super. + # Защита от ложняков (ling→inkling): минимум 10 символов И есть цифра или дефис. + gk = norm_keep(cleaned) + if len(gk) >= 10 and (re.search(r"\d", gk) or "-" in gk): + for m in arena_models: + am = norm_keep(m["model"]) + if gk == am or am.endswith("-" + gk) or ("-" + gk) in am: + return m, "containment" + return None, "" + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--provider", default="openrouter", + help="префикс провайдера гейтвея (openrouter/zai/nvidia/...)") + ap.add_argument("--timeout", type=int, default=60) + args = ap.parse_args() + + models = gateway_models().get(args.provider, []) + if not models: + print(f"Провайдер '{args.provider}' не найден. Есть: {', '.join(gateway_models())}") + sys.exit(1) + + lb = get_leaderboard("text") + arena_models = lb["models"] + + print(f"# OmniRoute прогон: {args.provider} ({len(models)} моделей)\n") + print(f"Арена-снапшот: {lb['updated']}, {lb['total_models']} моделей\n") + print("| Модель гейтвея | Статус | Время | Arena Rank | Arena Score | Arena Model |") + print("|---|---|---|---|---|---|") + + ok_n = 0 + for m in sorted(models): + if any(x in m for x in ("image", "flux", "embedding")): + st, ans, dt = "skip", "image/embed", "-" + else: + st, ans, dt = test_with_retry(m, timeout=args.timeout) + ok_n += st == "ok" + + a, note = (None, "") + if st == "ok": + a, note = arena_lookup(m, arena_models) + a_cell = f"#{a['rank']} ({a['score']})" if a else "—" + a_name = a["model"] if a else "нет в арене" + print(f"| {m} | {st} | {dt}s | {a_cell} | {a['score'] if a else '—'} | {a_name} |") + + print(f"\n**Итого: {ok_n}/{len(models)} отвечают.**") + + +if __name__ == "__main__": + main()