fix: run leaderboard stats in background

This commit is contained in:
吉鑫升 committed 2026-06-05 21:15:20 +08:00
1 parent f4ccef8c70
commit db1b32612b
4 files changed
+269 -1

No files matched your search

+3
View File
@@ -104,6 +104,9 @@ export interface AdminLeaderboardResponse {
range_days: number; range_days: number;
since: string; since: string;
until: string; until: string;
status?: "ready" | "running" | "refreshing";
generated_at?: string;
message?: string;
overview: AdminLeaderboardOverview; overview: AdminLeaderboardOverview;
users_by_activity: AdminLeaderboardUser[]; users_by_activity: AdminLeaderboardUser[];
users_by_questions: AdminLeaderboardUser[]; users_by_questions: AdminLeaderboardUser[];
+4
View File
@@ -33,6 +33,10 @@ export function useUserLeaderboard(
limit, limit,
], ],
queryFn: () => getUserLeaderboard(range, limit), queryFn: () => getUserLeaderboard(range, limit),
refetchInterval: (query) => {
const status = query.state.data?.status;
return status === "running" || status === "refreshing" ? 3000 : false;
},
}); });
} }
@@ -1087,6 +1087,9 @@ export default function AdminUserLeaderboardPage() {
}, [data, query, tab]); }, [data, query, tab]);
const overview = data?.overview; 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 topUsers = rankedUsers.slice(0, 3);
const userPageCount = Math.max(1, Math.ceil(rankedUsers.length / LEADERBOARD_PAGE_SIZE)); const userPageCount = Math.max(1, Math.ceil(rankedUsers.length / LEADERBOARD_PAGE_SIZE));
const visibleUsers = useMemo(() => { const visibleUsers = useMemo(() => {
@@ -1446,8 +1449,29 @@ export default function AdminUserLeaderboardPage() {
</div> </div>
)} )}
{!isLoading && data && overview && ( {!isLoading && !error && leaderboardPending && !leaderboardHasRows && (
<div className="flex flex-col items-center justify-center gap-3 rounded-lg border border-[#F1D6D6] bg-white px-6 py-16 text-center text-sm text-[#5F6368] shadow-sm dark:border-[#462123] dark:bg-[#230D0F]">
<div className="flex items-center gap-2 font-medium text-[#202124] dark:text-white">
<Loader2 className="size-4 animate-spin text-[#C1121F]" />
<span>排行榜正在后台统计</span>
</div>
<div className="max-w-lg text-xs leading-5 text-[#7A5A5A] dark:text-[#D8B8B8]">
{data?.message || "统计任务已放到后台执行,页面会自动刷新结果。这样排行榜慢慢算,不会长时间占住当前接口请求。"}
</div>
</div>
)}
{!isLoading && data && overview && (!leaderboardPending || leaderboardHasRows) && (
<> <>
{leaderboardPending && leaderboardHasRows && (
<div className="flex flex-wrap items-center justify-between gap-3 rounded-lg border border-[#FDE68A] bg-[#FFFBEB] px-4 py-3 text-sm text-[#92400E] dark:border-amber-200/25 dark:bg-amber-300/10 dark:text-amber-100">
<div className="flex items-center gap-2">
<Loader2 className="size-4 animate-spin" />
<span>{data.message || "正在后台刷新排行榜,当前展示缓存结果。"}</span>
</div>
{data.generated_at && <span className="text-xs">缓存生成于 {fmtTime(data.generated_at)}</span>}
</div>
)}
<div className="grid gap-3 md:grid-cols-2 xl:grid-cols-4"> <div className="grid gap-3 md:grid-cols-2 xl:grid-cols-4">
<StatCard icon={Users} accent="crimson" label="总用户" value={fmtNumber(overview.total_users)} note={`${fmtNumber(overview.active_users)} 位在当前时间段活跃`} /> <StatCard icon={Users} accent="crimson" label="总用户" value={fmtNumber(overview.total_users)} note={`${fmtNumber(overview.active_users)} 位在当前时间段活跃`} />
<StatCard icon={MousePointerClick} accent="teal" label="使用次数" value={fmtNumber(overview.total_runs)} note={`${fmtNumber(overview.total_threads)} 个会话被创建`} /> <StatCard icon={MousePointerClick} accent="teal" label="使用次数" value={fmtNumber(overview.total_runs)} note={`${fmtNumber(overview.total_threads)} 个会话被创建`} />
@@ -18,6 +18,7 @@ import json
import logging import logging
import os import os
import re import re
import time
from collections.abc import AsyncIterator from collections.abc import AsyncIterator
from difflib import SequenceMatcher from difflib import SequenceMatcher
from hashlib import sha256 from hashlib import sha256
@@ -67,7 +68,14 @@ _LEADERBOARD_ACQUIRE_TIMEOUT_SECONDS = max(
0.1, 0.1,
float(os.getenv("ADMIN_LEADERBOARD_ACQUIRE_TIMEOUT_SECONDS", "0.5")), 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_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: async def _require_admin(request: Request) -> None:
@@ -154,6 +162,9 @@ class AdminLeaderboardResponse(BaseModel):
range_days: int range_days: int
since: str since: str
until: str until: str
status: str = "ready"
generated_at: str = ""
message: str = ""
overview: AdminLeaderboardOverview overview: AdminLeaderboardOverview
users_by_activity: list[AdminLeaderboardUser] = Field(default_factory=list) users_by_activity: list[AdminLeaderboardUser] = Field(default_factory=list)
users_by_questions: 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) @router.get("/leaderboard", response_model=AdminLeaderboardResponse)
async def get_user_leaderboard( async def get_user_leaderboard(
request: Request, request: Request,
@@ -999,6 +1173,69 @@ async def get_user_leaderboard(
if session_factory is None: if session_factory is None:
raise HTTPException(status_code=503, detail="Analytics requires database persistence") 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: try:
await asyncio.wait_for( await asyncio.wait_for(
_leaderboard_semaphore.acquire(), _leaderboard_semaphore.acquire(),