""" Arena Dashboard — FastAPI backend Таблица моделей OmniRoute с Arena Score (text + agent) и liveness-статусом. """ from __future__ import annotations import json import os import sqlite3 import time import urllib.request from contextlib import asynccontextmanager from datetime import datetime, timezone from typing import Optional from fastapi import FastAPI, Query from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles from usage_utils import extract_usage DB_PATH = os.environ.get("DB_PATH", "/data/arena.db") OMNIROUTE_URL = os.environ.get("OMNIROUTE_URL", "http://192.168.1.101:20128") OMNIROUTE_KEY = os.environ.get("OMNIROUTE_API_KEY", "") ARENA_CACHE_TEXT = os.environ.get("ARENA_CACHE_TEXT", "/data/arena_text.json") ARENA_CACHE_AGENT = os.environ.get("ARENA_CACHE_AGENT", "/data/arena_agent.json") # Сколько последних проверок показывать сегментами в колонке Uptime. HISTORY_LIMIT = int(os.environ.get("HISTORY_LIMIT", "40")) # Статика — относительно этого файла: работает и в контейнере (/app/app), # и при локальном запуске/тестах из чекаута репозитория. STATIC_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static") # ─── DB setup ─────────────────────────────────────────────────────────────── def get_db() -> sqlite3.Connection: db = sqlite3.connect(DB_PATH, check_same_thread=False) db.row_factory = sqlite3.Row return db def init_db(): db = get_db() # Автомиграция таблицы models если создана со старой схемой try: cols = [r[1] for r in db.execute("PRAGMA table_info(models)").fetchall()] if cols and "raw_base" not in cols: db.execute("DROP TABLE models") except Exception: pass db.executescript(""" CREATE TABLE IF NOT EXISTS arena_text_leaderboard ( rank INTEGER, model_name TEXT PRIMARY KEY, norm_name TEXT NOT NULL, clean_name TEXT NOT NULL, provider TEXT, score TEXT, preliminary INTEGER, rank_spread TEXT, votes INTEGER, price TEXT, context TEXT, fetched_at TEXT ); CREATE TABLE IF NOT EXISTS arena_agent_leaderboard ( rank INTEGER, model_name TEXT PRIMARY KEY, norm_name TEXT NOT NULL, clean_name TEXT NOT NULL, provider TEXT, net_improvement TEXT, confirmed_success TEXT, praise_vs_complaint TEXT, steerability TEXT, bash_recovery TEXT, tool_hallucination TEXT, sessions INTEGER, cost_per_task TEXT, output_tokens_per_task TEXT, price TEXT, fetched_at TEXT ); CREATE TABLE IF NOT EXISTS models ( id TEXT PRIMARY KEY, provider TEXT NOT NULL, name TEXT NOT NULL, raw_base TEXT NOT NULL, norm_base TEXT NOT NULL, clean_base TEXT NOT NULL, text_model_match TEXT, agent_model_match TEXT, updated_at TEXT, FOREIGN KEY(text_model_match) REFERENCES arena_text_leaderboard(model_name), FOREIGN KEY(agent_model_match) REFERENCES arena_agent_leaderboard(model_name) ); CREATE TABLE IF NOT EXISTS checks ( id INTEGER PRIMARY KEY AUTOINCREMENT, model_id TEXT NOT NULL, checked_at TEXT NOT NULL, ok INTEGER NOT NULL, latency_ms INTEGER, input_tokens INTEGER, output_tokens INTEGER, total_tokens INTEGER, usage_json TEXT ); CREATE INDEX IF NOT EXISTS idx_checks_model ON checks(model_id); CREATE INDEX IF NOT EXISTS idx_checks_time ON checks(checked_at); CREATE INDEX IF NOT EXISTS idx_at_norm ON arena_text_leaderboard(norm_name); CREATE INDEX IF NOT EXISTS idx_ag_norm ON arena_agent_leaderboard(norm_name); """) # Миграция: текст ошибки нужен для деталей в тултипах сегментов. cols = [r[1] for r in db.execute("PRAGMA table_info(checks)").fetchall()] if "error" not in cols: db.execute("ALTER TABLE checks ADD COLUMN error TEXT") # Идемпотентная миграция: учёт token usage. Старые строки остаются NULL # (неизвестное потребление не подменяется нулями), повторный запуск — no-op. for col, decl in ( ("input_tokens", "INTEGER"), ("output_tokens", "INTEGER"), ("total_tokens", "INTEGER"), ("usage_json", "TEXT"), ): if col not in cols: db.execute(f"ALTER TABLE checks ADD COLUMN {col} {decl}") db.commit() db.close() # ─── Arena data ingestion into DB ─────────────────────────────────────────── def ingest_arena_tables(db: sqlite3.Connection): """Импортирует сырые JSON кеши арены прямо в реляционные таблицы SQLite.""" now = datetime.now(timezone.utc).isoformat() if os.path.exists(ARENA_CACHE_TEXT): try: with open(ARENA_CACHE_TEXT) as f: tdata = json.load(f) for m in tdata.get("models", []): raw_m = m["model"] m_only = raw_m.split(" · ")[0] m_only = re.sub(r"\s+(Google|Anthropic|OpenAI|Meta|Nvidia|Mistral|DeepSeek|Alibaba)$", "", m_only, flags=re.I) clean = re.sub(r"\s*\([^)]+\)", "", m_only) db.execute(""" INSERT INTO arena_text_leaderboard ( rank, model_name, norm_name, clean_name, provider, score, preliminary, rank_spread, votes, price, context, fetched_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(model_name) DO UPDATE SET rank = excluded.rank, norm_name = excluded.norm_name, clean_name = excluded.clean_name, provider = excluded.provider, score = excluded.score, preliminary = excluded.preliminary, rank_spread = excluded.rank_spread, votes = excluded.votes, price = excluded.price, context = excluded.context, fetched_at = excluded.fetched_at """, ( m.get("rank"), raw_m, _norm(m_only), _norm(clean), m.get("provider"), m.get("score"), int(bool(m.get("preliminary"))), m.get("rank_spread"), m.get("votes"), m.get("price"), m.get("context"), now )) except Exception as e: print(f"[ingest] text leaderboard error: {e}") if os.path.exists(ARENA_CACHE_AGENT): try: with open(ARENA_CACHE_AGENT) as f: adata = json.load(f) for m in adata.get("models", []): raw_m = m["model"] m_only = raw_m.split(" · ")[0] m_only = re.sub(r"\s+(Google|Anthropic|OpenAI|Meta|Nvidia|Mistral|DeepSeek|Alibaba)$", "", m_only, flags=re.I) clean = re.sub(r"\s*\([^)]+\)", "", m_only) db.execute(""" INSERT INTO arena_agent_leaderboard ( rank, model_name, norm_name, clean_name, provider, net_improvement, confirmed_success, praise_vs_complaint, steerability, bash_recovery, tool_hallucination, sessions, cost_per_task, output_tokens_per_task, price, fetched_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(model_name) DO UPDATE SET rank = excluded.rank, norm_name = excluded.norm_name, clean_name = excluded.clean_name, provider = excluded.provider, net_improvement = excluded.net_improvement, confirmed_success = excluded.confirmed_success, praise_vs_complaint = excluded.praise_vs_complaint, steerability = excluded.steerability, bash_recovery = excluded.bash_recovery, tool_hallucination = excluded.tool_hallucination, sessions = excluded.sessions, cost_per_task = excluded.cost_per_task, output_tokens_per_task = excluded.output_tokens_per_task, price = excluded.price, fetched_at = excluded.fetched_at """, ( m.get("rank"), raw_m, _norm(m_only), _norm(clean), m.get("provider"), m.get("net_improvement"), m.get("confirmed_success"), m.get("praise_vs_complaint"), m.get("steerability"), m.get("bash_recovery"), m.get("tool_hallucination"), m.get("sessions"), m.get("cost_per_task"), m.get("output_tokens_per_task"), m.get("price"), now )) except Exception as e: print(f"[ingest] agent leaderboard error: {e}") db.commit() # ─── OmniRoute helpers ──────────────────────────────────────────────────────── def fetch_omni_models() -> list[dict]: req = urllib.request.Request( f"{OMNIROUTE_URL}/v1/models", headers={"Authorization": f"Bearer {OMNIROUTE_KEY}"}, ) with urllib.request.urlopen(req, timeout=15) as r: data = json.loads(r.read()) return data.get("data", []) def is_skip(model_id: str) -> bool: return any(x in model_id.lower() for x in [ "image", "embed", "flux", "tts", "stt", "audio", "whisper", "vision", "guard", "safety", "moderat", ]) # ─── Arena matching (simple) ───────────────────────────────────────────────── import re def _norm(s: str) -> str: return re.sub(r"[^a-z0-9]", "", s.lower()) def _norm_keep(s: str) -> str: return re.sub(r"[^a-z0-9\-]", "", s.lower()) def arena_match(model_id: str, arena: dict[str, dict]) -> Optional[dict]: """Сопоставить модель OmniRoute с записью арены.""" raw = model_id.split("/")[-1].lower() base = re.sub(r"-(it|preview)(:[^:]*)?$", "", raw) base = re.sub(r"(:[^:]*)$", "", base) # 1) точное совпадение нормализованного имени bn = _norm(base) for k, v in arena.items(): # убираем вендор-хвост " Anthropic · Proprietary" из названия модели арены model_name_only = v.get("model", k).split(" · ")[0] model_name_only = re.sub(r"\s+(Google|Anthropic|OpenAI|Meta|Nvidia|Mistral|DeepSeek|Alibaba)$", "", model_name_only, flags=re.I) an = _norm(model_name_only) if an == bn: return v # 2) без суффиксов тиров 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) cn = _norm(cleaned) for k, v in arena.items(): model_name_only = v.get("model", k).split(" · ")[0] model_name_only = re.sub(r"\s+(Google|Anthropic|OpenAI|Meta|Nvidia|Mistral|DeepSeek|Alibaba)$", "", model_name_only, flags=re.I) # убираем скобки тиров типа (Max), (High) m_cleaned = re.sub(r"\s*\([^)]+\)", "", model_name_only) if _norm(m_cleaned) == cn: return v # 3) containment for candidate in (base, cleaned): gk = _norm_keep(candidate) if len(gk) >= 10 and (re.search(r"\d", gk) or "-" in gk): for k, v in arena.items(): model_name_only = v.get("model", k).split(" · ")[0] am = _norm_keep(model_name_only) if gk == am or am.endswith("-" + gk) or ("-" + gk) in am or (gk in am and len(gk) >= 12): return v return None # ─── model sync ────────────────────────────────────────────────────────────── def sync_models(): """Обновить список моделей в DB из OmniRoute + найти связи с таблицами арены.""" db = get_db() ingest_arena_tables(db) try: omni = fetch_omni_models() except Exception as e: print(f"[sync] OmniRoute error: {e}") db.close() return # Загружаем нормализованные имена из таблиц арены для сопоставления text_rows = db.execute("SELECT model_name, norm_name, clean_name FROM arena_text_leaderboard").fetchall() agent_rows = db.execute("SELECT model_name, norm_name, clean_name FROM arena_agent_leaderboard").fetchall() def find_match(base: str, cleaned: str, rows) -> Optional[str]: bn = _norm(base) for r in rows: if r["norm_name"] == bn or r["clean_name"] == bn: return r["model_name"] cn = _norm(cleaned) for r in rows: if r["clean_name"] == cn: return r["model_name"] for candidate in (base, cleaned): gk = _norm_keep(candidate) if len(gk) >= 10 and (re.search(r"\d", gk) or "-" in gk): for r in rows: am = _norm_keep(r["model_name"]) if gk == am or am.endswith("-" + gk) or ("-" + gk) in am or (gk in am and len(gk) >= 12): return r["model_name"] return None now = datetime.now(timezone.utc).isoformat() for m in omni: mid = m["id"] if is_skip(mid): continue # openrouter — только :free if mid.startswith("openrouter/") and not mid.endswith(":free"): continue provider = mid.split("/")[0] raw = mid.split("/")[-1].lower() base = re.sub(r"-(it|preview)(:[^:]*)?$", "", raw) base = re.sub(r"(:[^:]*)$", "", base) 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) text_match = find_match(base, cleaned, text_rows) agent_match = find_match(base, cleaned, agent_rows) db.execute(""" INSERT INTO models (id, provider, name, raw_base, norm_base, clean_base, text_model_match, agent_model_match, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET text_model_match = excluded.text_model_match, agent_model_match = excluded.agent_model_match, updated_at = excluded.updated_at """, ( mid, provider, mid, base, _norm(base), _norm(cleaned), text_match, agent_match, now )) db.commit() db.close() print(f"[sync] synced {len(omni)} models at {now} with SQL foreign keys") # ─── Stats helpers ──────────────────────────────────────────────────────────── def model_stats(db: sqlite3.Connection, model_id: str, hours: int = 24) -> dict: """Последний статус и % успехов за N часов.""" cutoff = datetime.fromtimestamp(time.time() - hours * 3600, timezone.utc).isoformat() last = db.execute( "SELECT ok, latency_ms, checked_at FROM checks WHERE model_id=? ORDER BY checked_at DESC LIMIT 1", (model_id,) ).fetchone() rows = db.execute( "SELECT ok, latency_ms, checked_at, error, input_tokens, output_tokens, total_tokens FROM checks " "WHERE model_id=? AND checked_at > ? ORDER BY checked_at ASC", (model_id, cutoff) ).fetchall() total = len(rows) success = sum(1 for r in rows if r["ok"]) pct = round(100 * success / total) if total else None # История последних проверок (старые → новые) для сегментной витрины. # Каждый элемент — детали проверки: ok, latency, время, текст ошибки, токены. history = [ { "ok": bool(r["ok"]), "latency_ms": r["latency_ms"], "at": r["checked_at"], "error": r["error"], "input_tokens": r["input_tokens"], "output_tokens": r["output_tokens"], "total_tokens": r["total_tokens"], } for r in rows[-HISTORY_LIMIT:] ] return { "last_ok": bool(last["ok"]) if last else None, "last_latency_ms": last["latency_ms"] if last else None, "last_checked": last["checked_at"] if last else None, "uptime_pct": pct, "checks_count": total, "history": history, } # ─── Token usage aggregates ─────────────────────────────────────────────────── def _token_agg_rows(db: sqlite3.Connection, where: str, params: list) -> list[dict]: """GROUP BY провайдер по checks с суммами токенов и покрытием known/unknown.""" # Провайдер — из таблицы models; если модели уже нет, выводим из model_id. prov_expr = ( "COALESCE(m.provider, " "CASE WHEN instr(c.model_id, '/') > 0 " "THEN substr(c.model_id, 1, instr(c.model_id, '/') - 1) " "ELSE 'unknown' END)" ) return [ { "provider": r["provider"], "checks": r["checks"], "known": r["known"], "unknown": r["unknown"], "coverage_pct": round(100 * r["known"] / r["checks"]) if r["checks"] else None, "input_tokens": r["input_tokens"], "output_tokens": r["output_tokens"], "total_tokens": r["total_tokens"], } for r in db.execute( f""" SELECT {prov_expr} AS provider, COUNT(*) AS checks, SUM(CASE WHEN c.total_tokens IS NOT NULL THEN 1 ELSE 0 END) AS known, SUM(CASE WHEN c.total_tokens IS NULL THEN 1 ELSE 0 END) AS unknown, SUM(c.input_tokens) AS input_tokens, SUM(c.output_tokens) AS output_tokens, SUM(c.total_tokens) AS total_tokens FROM checks c LEFT JOIN models m ON c.model_id = m.id WHERE {where} GROUP BY provider ORDER BY total_tokens DESC """, params, ).fetchall() ] def _merge_aggs(rows: list[dict]) -> dict: """Свернуть группы провайдеров в одну сводку.""" checks = sum(r["checks"] for r in rows) known = sum(r["known"] for r in rows) return { "checks": checks, "known": known, "unknown": checks - known, "coverage_pct": round(100 * known / checks) if checks else None, "input_tokens": sum(r["input_tokens"] or 0 for r in rows), "output_tokens": sum(r["output_tokens"] or 0 for r in rows), "total_tokens": sum(r["total_tokens"] or 0 for r in rows), } def token_stats(db: sqlite3.Connection) -> dict: """Сводка token usage: скользящие 24h и всё время + разбивка по провайдерам.""" cutoff24 = datetime.fromtimestamp(time.time() - 24 * 3600, timezone.utc).isoformat() now = datetime.now(timezone.utc).isoformat() prov24 = _token_agg_rows(db, "c.checked_at > ?", [cutoff24]) prov_all = _token_agg_rows(db, "1=1", []) all_map = {r["provider"]: r for r in prov_all} return { "generated_at": now, "window_24h": _merge_aggs(prov24), "all_time": _merge_aggs(prov_all), "providers": [ { "provider": r["provider"], "window_24h": r, "all_time": all_map.get(r["provider"], _merge_aggs([])), } for r in prov24 ] + [ { "provider": r["provider"], "window_24h": _merge_aggs([]), "all_time": r, } for r in prov_all if r["provider"] not in {p["provider"] for p in prov24} ], } # ─── Lifespan ──────────────────────────────────────────────────────────────── @asynccontextmanager async def lifespan(app: FastAPI): init_db() sync_models() yield app = FastAPI(title="Arena Dashboard", lifespan=lifespan) # ─── API routes ─────────────────────────────────────────────────────────────── @app.get("/api/models") def api_models( provider: Optional[str] = Query(None), arena_only: bool = Query(False), sort: str = Query("arena_text_rank"), ): db = get_db() # SQL JOIN таблиц арены с моделями OmniRoute! q = """ SELECT m.id, m.provider, m.name, t.score as arena_text_score, t.rank as arena_text_rank, a.net_improvement as arena_agent_score, a.rank as arena_agent_rank, t.votes as arena_text_votes, t.price as arena_text_price FROM models m LEFT JOIN arena_text_leaderboard t ON m.text_model_match = t.model_name LEFT JOIN arena_agent_leaderboard a ON m.agent_model_match = a.model_name WHERE 1=1 """ params: list = [] if provider: q += " AND m.provider = ?" params.append(provider) if arena_only: q += " AND t.score IS NOT NULL" rows = db.execute(q, params).fetchall() result = [] for r in rows: mid = r["id"] stats = model_stats(db, mid) result.append({ "id": mid, "provider": r["provider"], "name": r["name"], "arena_text_score": r["arena_text_score"], "arena_text_rank": r["arena_text_rank"], "arena_agent_score": r["arena_agent_score"], "arena_agent_rank": r["arena_agent_rank"], "arena_text_votes": r["arena_text_votes"], "arena_text_price": r["arena_text_price"], **stats, }) db.close() # сортировка def sort_key(x): if sort == "arena_text_rank": return (x["arena_text_rank"] is None, x["arena_text_rank"] or 9999) if sort == "uptime_pct": return (x["uptime_pct"] is None, -(x["uptime_pct"] or 0)) if sort == "provider": return x["provider"] return x.get(sort) or "" result.sort(key=sort_key) return JSONResponse(result) @app.get("/api/providers") def api_providers(): db = get_db() rows = db.execute("SELECT DISTINCT provider FROM models ORDER BY provider").fetchall() db.close() return [r["provider"] for r in rows] @app.post("/api/sync") def api_sync(): sync_models() return {"ok": True} @app.get("/api/check/{model_id:path}") def api_check(model_id: str): """Мгновенная проверка одной модели.""" import time as _t t0 = _t.time() ok = False latency = None error = None usage = {"input_tokens": None, "output_tokens": None, "total_tokens": None, "usage_json": None} try: body = json.dumps({ "model": model_id, "messages": [{"role": "user", "content": "What is 2+2? Reply with just the number."}], "max_tokens": 5, }).encode() req = urllib.request.Request( f"{OMNIROUTE_URL}/v1/chat/completions", data=body, headers={ "Authorization": f"Bearer {OMNIROUTE_KEY}", "Content-Type": "application/json", }, ) with urllib.request.urlopen(req, timeout=30) as r: resp = json.loads(r.read()) # Usage фиксируем ДО валидации контента: даже если парсинг choices # упадёт, реально потраченные токены сохраняются в checks. usage = extract_usage(resp) content = resp["choices"][0]["message"]["content"] ok = bool(content.strip()) except Exception as e: error = f"{type(e).__name__}: {e}" print(f"[check] {model_id}: {error}") latency = int((_t.time() - t0) * 1000) now = datetime.now(timezone.utc).isoformat() db = get_db() db.execute( "INSERT INTO checks (model_id, checked_at, ok, latency_ms, error, " "input_tokens, output_tokens, total_tokens, usage_json) " "VALUES (?,?,?,?,?,?,?,?,?)", (model_id, now, int(ok), latency, error, usage["input_tokens"], usage["output_tokens"], usage["total_tokens"], usage["usage_json"]), ) db.commit() stats = model_stats(db, model_id) db.close() return { "model_id": model_id, "ok": ok, "latency_ms": latency, "checked_at": now, "input_tokens": usage["input_tokens"], "output_tokens": usage["output_tokens"], "total_tokens": usage["total_tokens"], "history": stats["history"], "uptime_pct": stats["uptime_pct"], } @app.get("/api/checks/{model_id:path}") def api_checks(model_id: str): """Все проверки модели за период — для дропдауна с деталями ошибок.""" limit = int(os.environ.get("CHECKS_LIMIT", "100")) db = get_db() rows = db.execute( "SELECT ok, latency_ms, checked_at, error, input_tokens, output_tokens, total_tokens, usage_json " "FROM checks WHERE model_id=? ORDER BY checked_at DESC LIMIT ?", (model_id, limit), ).fetchall() db.close() return { "model_id": model_id, "checks": [ { "ok": bool(r["ok"]), "latency_ms": r["latency_ms"], "checked_at": r["checked_at"], "error": r["error"], "input_tokens": r["input_tokens"], "output_tokens": r["output_tokens"], "total_tokens": r["total_tokens"], "usage_json": r["usage_json"], } for r in rows ], } @app.get("/api/token_stats") def api_token_stats(): """Агрегат token usage: скользящие 24h и всё время, покрытие known/unknown, разбивка по провайдерам.""" db = get_db() result = token_stats(db) db.close() return JSONResponse(result) @app.get("/", response_class=HTMLResponse) def index(): with open(os.path.join(STATIC_DIR, "index.html")) as f: return f.read() app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")