From db1b32612b8de48bbb4decc112064cb3877e47ec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=89=E9=91=AB=E5=8D=87?= Date: Fri, 5 Jun 2026 21:15:20 +0800 Subject: [PATCH] fix: run leaderboard stats in background --- frontend-web/src/core/admin/api.ts | 3 + frontend-web/src/core/admin/hooks.ts | 4 + .../src/pages/AdminUserLeaderboardPage.tsx | 26 +- .../app/gateway/routers/admin_users.py | 237 ++++++++++++++++++ 4 files changed, 269 insertions(+), 1 deletion(-) diff --git a/frontend-web/src/core/admin/api.ts b/frontend-web/src/core/admin/api.ts index d56004d9..e54b494b 100644 --- a/frontend-web/src/core/admin/api.ts +++ b/frontend-web/src/core/admin/api.ts @@ -104,6 +104,9 @@ export interface AdminLeaderboardResponse { range_days: number; since: string; until: string; + status?: "ready" | "running" | "refreshing"; + generated_at?: string; + message?: string; overview: AdminLeaderboardOverview; users_by_activity: AdminLeaderboardUser[]; users_by_questions: AdminLeaderboardUser[]; diff --git a/frontend-web/src/core/admin/hooks.ts b/frontend-web/src/core/admin/hooks.ts index 9d19457b..243c06a5 100644 --- a/frontend-web/src/core/admin/hooks.ts +++ b/frontend-web/src/core/admin/hooks.ts @@ -33,6 +33,10 @@ export function useUserLeaderboard( limit, ], queryFn: () => getUserLeaderboard(range, limit), + refetchInterval: (query) => { + const status = query.state.data?.status; + return status === "running" || status === "refreshing" ? 3000 : false; + }, }); } diff --git a/frontend-web/src/pages/AdminUserLeaderboardPage.tsx b/frontend-web/src/pages/AdminUserLeaderboardPage.tsx index 412f0b3c..0122e7b2 100644 --- a/frontend-web/src/pages/AdminUserLeaderboardPage.tsx +++ b/frontend-web/src/pages/AdminUserLeaderboardPage.tsx @@ -1087,6 +1087,9 @@ export default function AdminUserLeaderboardPage() { }, [data, query, tab]); const overview = data?.overview; + const leaderboardStatus = data?.status; + const leaderboardPending = leaderboardStatus === "running" || leaderboardStatus === "refreshing"; + const leaderboardHasRows = rankedUsers.length > 0 || (overview?.total_runs ?? 0) > 0 || (overview?.total_users ?? 0) > 0; const topUsers = rankedUsers.slice(0, 3); const userPageCount = Math.max(1, Math.ceil(rankedUsers.length / LEADERBOARD_PAGE_SIZE)); const visibleUsers = useMemo(() => { @@ -1446,8 +1449,29 @@ export default function AdminUserLeaderboardPage() { )} - {!isLoading && data && overview && ( + {!isLoading && !error && leaderboardPending && !leaderboardHasRows && ( +
+
+ + 排行榜正在后台统计 +
+
+ {data?.message || "统计任务已放到后台执行，页面会自动刷新结果。这样排行榜慢慢算，不会长时间占住当前接口请求。"} +
+
+ )} + + {!isLoading && data && overview && (!leaderboardPending || leaderboardHasRows) && ( <> + {leaderboardPending && leaderboardHasRows && ( +
+
+ + {data.message || "正在后台刷新排行榜，当前展示缓存结果。"} +
+ {data.generated_at && 缓存生成于 {fmtTime(data.generated_at)}} +
+ )}
diff --git a/offline-backend-20260512/backend/app/gateway/routers/admin_users.py b/offline-backend-20260512/backend/app/gateway/routers/admin_users.py index bdb020d2..829361d2 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/admin_users.py +++ b/offline-backend-20260512/backend/app/gateway/routers/admin_users.py @@ -18,6 +18,7 @@ import json import logging import os import re +import time from collections.abc import AsyncIterator from difflib import SequenceMatcher from hashlib import sha256 @@ -67,7 +68,14 @@ _LEADERBOARD_ACQUIRE_TIMEOUT_SECONDS = max( 0.1, float(os.getenv("ADMIN_LEADERBOARD_ACQUIRE_TIMEOUT_SECONDS", "0.5")), ) +_LEADERBOARD_CACHE_TTL_SECONDS = max( + 5.0, + float(os.getenv("ADMIN_LEADERBOARD_CACHE_TTL_SECONDS", "300")), +) _leaderboard_semaphore = asyncio.Semaphore(_LEADERBOARD_MAX_CONCURRENT) +_leaderboard_jobs: dict[str, asyncio.Task] = {} +_leaderboard_cache: dict[str, dict] = {} +_leaderboard_jobs_lock = asyncio.Lock() async def _require_admin(request: Request) -> None: @@ -154,6 +162,9 @@ class AdminLeaderboardResponse(BaseModel): range_days: int since: str until: str + status: str = "ready" + generated_at: str = "" + message: str = "" overview: AdminLeaderboardOverview users_by_activity: list[AdminLeaderboardUser] = Field(default_factory=list) users_by_questions: list[AdminLeaderboardUser] = Field(default_factory=list) @@ -981,6 +992,169 @@ async def _query_overview( ) +def _empty_leaderboard_response( + since_dt: datetime, + until_dt: datetime, + range_days: int, + *, + status: str, + message: str, +) -> AdminLeaderboardResponse: + return AdminLeaderboardResponse( + range_days=range_days, + since=_iso(since_dt), + until=_iso(until_dt), + status=status, + message=message, + overview=AdminLeaderboardOverview(), + ) + + +def _leaderboard_cache_key( + *, + since_dt: datetime, + until_dt: datetime, + limit: int, + excluded: set[str], + include_scheduled: bool, + include_admins: bool, + include_failed: bool, + cleanup_system_questions: bool, +) -> str: + payload = { + "version": 2, + "since": _iso(since_dt), + "until": _iso(until_dt), + "limit": limit, + "excluded": sorted(excluded), + "include_scheduled": include_scheduled, + "include_admins": include_admins, + "include_failed": include_failed, + "cleanup_system_questions": cleanup_system_questions, + } + return sha256(json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest() + + +async def _build_user_leaderboard_response( + session_factory, + settings, + since_dt: datetime, + until_dt: datetime, + range_days: int, + limit: int, + excluded: set[str], + include_scheduled: bool, + include_admins: bool, + include_failed: bool, +) -> AdminLeaderboardResponse: + async with session_factory() as session: + common = { + "exclude_user_ids": excluded, + "include_scheduled": include_scheduled, + "include_admins": include_admins, + "include_failed": include_failed, + } + tool_common = { + "exclude_user_ids": excluded, + "include_scheduled": include_scheduled, + "include_admins": include_admins, + } + question_common = { + **common, + "cleanup_system_questions": settings.cleanup_system_questions, + } + overview = await _query_overview(session, since_dt, until_dt, **common) + users = await _query_user_leaderboard(session, since_dt, until_dt, **common) + skills = await _query_call_rankings(session, since_dt, until_dt, user_id=None, field="skill", limit=limit, **tool_common) + tools = await _query_call_rankings(session, since_dt, until_dt, user_id=None, field="tool", limit=limit, **tool_common) + models = await _query_model_rankings(session, since_dt, until_dt, user_id=None, limit=limit, **common) + if not models: + models = await _query_run_dimensions(session, since_dt, until_dt, user_id=None, field="model", limit=limit, **common) + agents = await _query_run_dimensions(session, since_dt, until_dt, user_id=None, field="agent", limit=limit, **common) + run_statuses = await _query_run_dimensions(session, since_dt, until_dt, user_id=None, field="status", limit=limit, **common) + recent_questions = await _query_recent_questions(session, since_dt, until_dt, user_id=None, limit=limit, **question_common) + frequent_questions = await _query_frequent_questions(session, since_dt, until_dt, limit=limit, **question_common) + trends = await _query_trends(session, since_dt, until_dt, **common) + + return AdminLeaderboardResponse( + range_days=range_days, + since=_iso(since_dt), + until=_iso(until_dt), + status="ready", + generated_at=_iso(datetime.utcnow()), + overview=overview, + users_by_activity=sorted(users, key=lambda u: (u.run_count + u.tool_call_count, u.total_tokens), reverse=True)[:limit], + users_by_questions=sorted(users, key=lambda u: (u.qa_count, u.run_count), reverse=True)[:limit], + users_by_tokens=sorted(users, key=lambda u: u.total_tokens, reverse=True)[:limit], + skills=skills, + tools=tools, + models=models, + agents=agents, + run_statuses=run_statuses, + recent_questions=recent_questions, + frequent_questions=frequent_questions, + trends=trends, + ) + + +async def _refresh_leaderboard_cache( + cache_key: str, + *, + session_factory, + settings, + since_dt: datetime, + until_dt: datetime, + range_days: int, + limit: int, + excluded: set[str], + include_scheduled: bool, + include_admins: bool, + include_failed: bool, +) -> None: + acquired = False + try: + await _leaderboard_semaphore.acquire() + acquired = True + response = await asyncio.wait_for( + _build_user_leaderboard_response( + session_factory, + settings, + since_dt, + until_dt, + range_days, + limit, + excluded, + include_scheduled, + include_admins, + include_failed, + ), + timeout=_LEADERBOARD_TIMEOUT_SECONDS, + ) + _leaderboard_cache[cache_key] = { + "created_at": time.monotonic(), + "response": response, + } + except asyncio.TimeoutError: + logger.warning("Leaderboard refresh timed out for key=%s", cache_key) + except Exception: + logger.exception("Leaderboard refresh failed for key=%s", cache_key) + finally: + if acquired: + _leaderboard_semaphore.release() + async with _leaderboard_jobs_lock: + _leaderboard_jobs.pop(cache_key, None) + + +async def _ensure_leaderboard_refresh(cache_key: str, **kwargs) -> None: + async with _leaderboard_jobs_lock: + task = _leaderboard_jobs.get(cache_key) + if task and not task.done(): + return + _leaderboard_jobs[cache_key] = asyncio.create_task( + _refresh_leaderboard_cache(cache_key, **kwargs) + ) + + @router.get("/leaderboard", response_model=AdminLeaderboardResponse) async def get_user_leaderboard( request: Request, @@ -999,6 +1173,69 @@ async def get_user_leaderboard( if session_factory is None: raise HTTPException(status_code=503, detail="Analytics requires database persistence") + settings = load_system_settings().leaderboard + since_dt, until_dt, range_days = _resolve_range(days, since, until) + excluded = {uid for uid in settings.exclude_user_ids if uid} + excluded.update({uid for uid in (exclude_user_ids or []) if uid}) + resolved_include_scheduled = settings.include_scheduled if include_scheduled is None else include_scheduled + resolved_include_admins = settings.include_admins if include_admins is None else include_admins + resolved_include_failed = settings.include_failed if include_failed is None else include_failed + cache_key = _leaderboard_cache_key( + since_dt=since_dt, + until_dt=until_dt, + limit=limit, + excluded=excluded, + include_scheduled=resolved_include_scheduled, + include_admins=resolved_include_admins, + include_failed=resolved_include_failed, + cleanup_system_questions=settings.cleanup_system_questions, + ) + + cached = _leaderboard_cache.get(cache_key) + if cached: + response = cached["response"].model_copy() + age = time.monotonic() - float(cached["created_at"]) + if age <= _LEADERBOARD_CACHE_TTL_SECONDS: + response.status = "ready" + return response + response.status = "refreshing" + response.message = "排行榜缓存已返回，后台正在刷新最新统计。" + await _ensure_leaderboard_refresh( + cache_key, + session_factory=session_factory, + settings=settings, + since_dt=since_dt, + until_dt=until_dt, + range_days=range_days, + limit=limit, + excluded=excluded, + include_scheduled=resolved_include_scheduled, + include_admins=resolved_include_admins, + include_failed=resolved_include_failed, + ) + return response + + await _ensure_leaderboard_refresh( + cache_key, + session_factory=session_factory, + settings=settings, + since_dt=since_dt, + until_dt=until_dt, + range_days=range_days, + limit=limit, + excluded=excluded, + include_scheduled=resolved_include_scheduled, + include_admins=resolved_include_admins, + include_failed=resolved_include_failed, + ) + return _empty_leaderboard_response( + since_dt, + until_dt, + range_days, + status="running", + message="排行榜统计已在后台运行。线上用户较多时会慢一些，页面会自动刷新结果。", + ) + try: await asyncio.wait_for( _leaderboard_semaphore.acquire(),