feat(admin): 用户排行榜每日快照——手动刷新当天 + 凌晨自动定稿前一天

- 新增 POST /api/admin/users/leaderboard/refresh-today:同步强制重建当天快照
  并返回合并区间;前端 useRefreshTodayLeaderboard + AdminUserLeaderboardPage
  「刷新当天」按钮
- Gateway 调度循环在每日 ADMIN_LEADERBOARD_FINALIZE_HOUR(默认凌晨1点)对前一天
  强制重算一次,避免临近午夜落库的 runs/tool-calls 漏统计;判定逻辑下沉为纯函数
  deerflow.persistence.admin_stats.finalize.should_finalize_previous_day
- 新增 scripts/sql/column_additions_mysql.sql:既有 MySQL 库手动补列的幂等脚本
  (Alembic 自动迁移已禁用,create_all 不改既有表)
- CLAUDE.md 补充排行榜快照、Schema 迁移约定,并补齐技能/知识库选模型的文档

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
吉鑫升andClaude Opus 4.8 committed 2026-06-10 11:24:21 +08:00
1 parent 96e0d695c0
commit d61aa853f1
8 files changed
+356 -3

No files matched your search

+26
View File
@@ -247,6 +247,32 @@ export async function getUserLeaderboard(
); );
} }
/**
* Force a synchronous rebuild of *today's* leaderboard snapshot so the latest
* real-time records are included, then return the leaderboard for `range`.
*/
export async function refreshTodayLeaderboard(
range: {
since: string;
until: string;
days?: number;
},
limit = 20,
): Promise<AdminLeaderboardResponse> {
const params = new URLSearchParams({
days: String(range.days ?? 30),
since: range.since,
until: range.until,
limit: String(limit),
});
return parse(
await apiFetch(`${base()}/leaderboard/refresh-today?${params}`, {
method: "POST",
}),
"Failed to refresh today's leaderboard",
);
}
export async function getUserAnalytics( export async function getUserAnalytics(
userId: string, userId: string,
range: { since: string; until: string; days?: number }, range: { since: string; until: string; days?: number },
+31
View File
@@ -9,6 +9,7 @@ import {
getUserMemory, getUserMemory,
getUserAnalytics, getUserAnalytics,
getUserLeaderboard, getUserLeaderboard,
refreshTodayLeaderboard,
listUserQa, listUserQa,
listUserThreadMessages, listUserThreadMessages,
listUserThreads, listUserThreads,
@@ -41,6 +42,36 @@ export function useUserLeaderboard(
}); });
} }
/**
* Force-rebuild today's snapshot, then prime + invalidate the matching
* leaderboard query so the page shows the freshly recomputed numbers.
*/
export function useRefreshTodayLeaderboard() {
const qc = useQueryClient();
return useMutation({
mutationFn: ({
range,
limit = 20,
}: {
range: { since: string; until: string; days?: number };
limit?: number;
}) => refreshTodayLeaderboard(range, limit),
onSuccess: (data, variables) => {
qc.setQueryData(
[
"admin-activity",
"leaderboard",
variables.range.since,
variables.range.until,
variables.limit ?? 20,
],
data,
);
qc.invalidateQueries({ queryKey: ["admin-activity", "leaderboard"] });
},
});
}
export function useUserAnalytics( export function useUserAnalytics(
userId: string | null, userId: string | null,
range: { since: string; until: string; days?: number }, range: { since: string; until: string; days?: number },
@@ -18,6 +18,7 @@ import {
Medal, Medal,
MessageSquareText, MessageSquareText,
MousePointerClick, MousePointerClick,
RefreshCw,
Search, Search,
Settings2, Settings2,
Sparkles, Sparkles,
@@ -55,7 +56,7 @@ import type {
AdminLeaderboardUser, AdminLeaderboardUser,
} from "@/core/admin/api"; } from "@/core/admin/api";
import { streamUserLeaderboardReport } from "@/core/admin/api"; import { streamUserLeaderboardReport } from "@/core/admin/api";
import { useUserAnalytics, useUserLeaderboard } from "@/core/admin/hooks"; import { useRefreshTodayLeaderboard, useUserAnalytics, useUserLeaderboard } from "@/core/admin/hooks";
import { useModels } from "@/core/models/hooks"; import { useModels } from "@/core/models/hooks";
import { downloadMarkdownAsDocx } from "@/core/scheduled-tasks/markdown-docx"; import { downloadMarkdownAsDocx } from "@/core/scheduled-tasks/markdown-docx";
import { cn } from "@/lib/utils"; import { cn } from "@/lib/utils";
@@ -1052,6 +1053,7 @@ export default function AdminUserLeaderboardPage() {
[range], [range],
); );
const { data, isLoading, error } = useUserLeaderboard(apiRange, 100); const { data, isLoading, error } = useUserLeaderboard(apiRange, 100);
const refreshToday = useRefreshTodayLeaderboard();
const [reportOpen, setReportOpen] = useState(false); const [reportOpen, setReportOpen] = useState(false);
const { models: reportModels, isLoading: reportModelsLoading } = useModels({ enabled: reportOpen }); const { models: reportModels, isLoading: reportModelsLoading } = useModels({ enabled: reportOpen });
const [reportModelName, setReportModelName] = useState(DEFAULT_REPORT_MODEL); const [reportModelName, setReportModelName] = useState(DEFAULT_REPORT_MODEL);
@@ -1269,6 +1271,30 @@ export default function AdminUserLeaderboardPage() {
> >
<Settings2 className="size-4" /> <Settings2 className="size-4" />
</Button> </Button>
<Button
variant="outline"
size="sm"
disabled={refreshToday.isPending}
onClick={() =>
refreshToday.mutate(
{ range: apiRange, limit: 100 },
{
onSuccess: () => toast.success("已刷新当天最新统计数据"),
onError: (err) =>
toast.error(err instanceof Error ? err.message : "刷新当天数据失败"),
},
)
}
className="h-9"
title="强制重算今天的统计,把最新的实时记录纳入榜单"
>
{refreshToday.isPending ? (
<Loader2 className="mr-1 size-4 animate-spin" />
) : (
<RefreshCw className="mr-1 size-4" />
)}
{refreshToday.isPending ? "刷新中" : "刷新当天"}
</Button>
<Button <Button
variant="outline" variant="outline"
size="sm" size="sm"
+20 -2
View File
@@ -250,10 +250,18 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA
| **Scheduled Tasks** (`/api/scheduled-tasks`) | Per-user cron-like agent tasks. `GET/POST /` - list/create owned tasks; `PATCH/DELETE /{id}` - update/delete (owner); `POST /{id}/{pause,resume,trigger}`; `GET /{id}/runs`, `GET /runs/{rid}`, run-file preview/edit/download; `GET /runs/{rid}/messages` - agent messages for the run's underlying agent run (used to inspect a stuck/running run; authorized via `_load_run`); `GET /runs/{rid}/html` - generated HTML document for an `html_page` run; `DELETE /runs/{rid}` - owner-scoped, with an admin fallback so admins can delete any user's run; publish/subscribe under `/{id}/{publish,unpublish,subscribe}` and `GET /published`. **Admin (cross-user):** `GET /admin/all` - every task; `PATCH/DELETE /admin/{id}` - edit/delete any task; `POST /admin/{id}/trigger`; `GET /admin/{id}/runs`. Admin routes require `system_role == "admin"`. | | **Scheduled Tasks** (`/api/scheduled-tasks`) | Per-user cron-like agent tasks. `GET/POST /` - list/create owned tasks; `PATCH/DELETE /{id}` - update/delete (owner); `POST /{id}/{pause,resume,trigger}`; `GET /{id}/runs`, `GET /runs/{rid}`, run-file preview/edit/download; `GET /runs/{rid}/messages` - agent messages for the run's underlying agent run (used to inspect a stuck/running run; authorized via `_load_run`); `GET /runs/{rid}/html` - generated HTML document for an `html_page` run; `DELETE /runs/{rid}` - owner-scoped, with an admin fallback so admins can delete any user's run; publish/subscribe under `/{id}/{publish,unpublish,subscribe}` and `GET /published`. **Admin (cross-user):** `GET /admin/all` - every task; `PATCH/DELETE /admin/{id}` - edit/delete any task; `POST /admin/{id}/trigger`; `GET /admin/{id}/runs`. Admin routes require `system_role == "admin"`. |
| **HTML Page Favorites** (`/api/html-page-favorites`) | Bookmarked generated HTML pages, owner-scoped. `GET /` - list; `POST /` `{page_id, title?, description?, tags?}` - bookmark a `scheduled_html_pages` row (extracts a reusable style/layout summary via GLM5 at favorite time); a page can only be favorited once per user (`DuplicateFavoriteError` → HTTP 409). `GET/DELETE /{id}` - read/delete (the create-task reference picker deletes favorites here). A favorite can be selected as a style reference when creating a new `html_page` task. | | **HTML Page Favorites** (`/api/html-page-favorites`) | Bookmarked generated HTML pages, owner-scoped. `GET /` - list; `POST /` `{page_id, title?, description?, tags?}` - bookmark a `scheduled_html_pages` row (extracts a reusable style/layout summary via GLM5 at favorite time); a page can only be favorited once per user (`DuplicateFavoriteError` → HTTP 409). `GET/DELETE /{id}` - read/delete (the create-task reference picker deletes favorites here). A favorite can be selected as a style reference when creating a new `html_page` task. |
| **Public Scheduled Tasks** (`/api/public/scheduled-tasks`) | **Unauthenticated** mirror for embedding. Adds `GET /runs/{rid}/html` and `GET /{task_id}/latest-html` for the `html_page` preview; the run result keeps `result.html` (only file paths are redacted). | | **Public Scheduled Tasks** (`/api/public/scheduled-tasks`) | **Unauthenticated** mirror for embedding. Adds `GET /runs/{rid}/html` and `GET /{task_id}/latest-html` for the `html_page` preview; the run result keeps `result.html` (only file paths are redacted). |
| **Knowledge** (`/api/knowledge`) | Product knowledge base, **global/shared** (no user isolation; writes record `created_by`/`updated_by` for audit). `POST /ingest/thread/{thread_id}` - sediment one thread into a note (reads messages from the checkpointer, `threads:read` owner-checked); `POST /ingest/threads/batch` - batch-sediment the caller's recent threads (`skip_existing` via `.manifest.json` dedup); `POST /ingest/search` - sediment search/tool results as a reference note; `POST /notes` - manual note; `GET /notes` (keyword/tag/source_type/status + pagination); `GET/PUT/DELETE /notes/{id}` (DELETE is soft = archive, `?hard=true` removes the vault file); `POST /search` - `mode=keyword\|vector\|hybrid` (vector→keyword fallback when no embedding endpoint); `GET /notes/{id}/versions` - edit history; `GET /duplicates` + `POST /merge` - dedup; `POST /reindex` - rebuild vector index; `GET /graph` - product graph data (notes + entities); `POST /export/graph` + `GET /export/files/{graph.json\|graph.html\|graph.graphml\|cypher.txt}` - obsidian-wiki `wiki-export` artifacts (self-contained `graph.html` for iframe); `POST /resolve` `{targets:[…]}` + `GET /notes/{id}/wikilinks` - resolve `[[wikilink]]` targets to a note id / vault page / missing (powers clickable wikilinks); `GET /vault-page?path=` - read a raw vault meta/hub Markdown page (no DB note). Backed by the `deerflow.knowledge` module (see below). | | **Knowledge** (`/api/knowledge`) | Product knowledge base, **global/shared** (no user isolation; writes record `created_by`/`updated_by` for audit). `POST /ingest/thread/{thread_id}` - sediment one thread into a note (reads messages from the checkpointer, `threads:read` owner-checked); `POST /ingest/threads/batch` - batch-sediment the caller's recent threads (`skip_existing` via `.manifest.json` dedup). **Optional model selection**: both ingest endpoints accept `model_name` (overrides `knowledge.extract_model_name` for that capture; threaded through `KnowledgeService.capture_thread` → `extract_thread_llm_multi`). **Model-error surfacing**: when the chosen model produces nothing usable (timeout / API error / non-JSON) capture still **falls back to rule-based extraction and lands a draft** (`llm_fallback:true`); the single-thread (non-background) response and the batch response (`llm_fallback` count) carry this so the UI tells the user "已改用规则抽取". The "保存到知识库" chat button stays background/fire-and-forget (passes `model_name`, no per-save error toast); the batch-import dialog is synchronous and shows the fallback count. `POST /ingest/search` - sediment search/tool results as a reference note; `POST /notes` - manual note; `GET /notes` (keyword/tag/source_type/status + pagination); `GET/PUT/DELETE /notes/{id}` (DELETE is soft = archive, `?hard=true` removes the vault file); `POST /search` - `mode=keyword\|vector\|hybrid` (vector→keyword fallback when no embedding endpoint); `GET /notes/{id}/versions` - edit history; `GET /duplicates` + `POST /merge` - dedup; `POST /reindex` - rebuild vector index; `GET /graph` - product graph data (notes + entities); `POST /export/graph` + `GET /export/files/{graph.json\|graph.html\|graph.graphml\|cypher.txt}` - obsidian-wiki `wiki-export` artifacts (self-contained `graph.html` for iframe); `POST /resolve` `{targets:[…]}` + `GET /notes/{id}/wikilinks` - resolve `[[wikilink]]` targets to a note id / vault page / missing (powers clickable wikilinks); `GET /vault-page?path=` - read a raw vault meta/hub Markdown page (no DB note). Backed by the `deerflow.knowledge` module (see below). |
Proxied through nginx: `/api/langgraph/*` → LangGraph, all other `/api/*` → Gateway. Proxied through nginx: `/api/langgraph/*` → LangGraph, all other `/api/*` → Gateway.
### Admin Leaderboard Snapshots (`/api/admin/users/leaderboard*`)
The admin 用户排行榜 is served from **precomputed per-day snapshots** in `admin_leaderboard_daily_stats` (keyed by `(stat_date, settings_hash)`), not by aggregating raw rows on every request. A background scheduler inside the Gateway (`_leaderboard_snapshot_scheduler_loop` in `app/gateway/app.py`, singleton file-locked) ticks every `ADMIN_LEADERBOARD_BACKGROUND_INTERVAL_SECONDS` (default 300s): it prewarms recent days, then builds any `queued`/`error`/stale-`running` day. `GET /leaderboard` merges the per-day snapshots for the requested range. **Today** is rebuilt whenever its snapshot is older than `ADMIN_LEADERBOARD_TODAY_TTL_SECONDS` (default 300s); a past day, once `ready`, is treated as fresh forever.
Two on-demand/automatic refresh paths layer on top:
- **Manual "刷新当天"** — `POST /leaderboard/refresh-today` force-rebuilds *today's* snapshot **synchronously** (bounded by the same semaphore/timeout the scheduler uses), then returns the merged range. Lets an admin pull the latest real-time records into the stats without waiting for the TTL. Frontend: a "刷新当天" button on `AdminUserLeaderboardPage` → `useRefreshTodayLeaderboard` (`core/admin/hooks.ts`), which primes + invalidates the `["admin-activity","leaderboard"]` query.
- **Daily finalize of the previous day** — once per day, on/after `ADMIN_LEADERBOARD_FINALIZE_HOUR` (local Beijing, default `1` = 凌晨 1 点), the scheduler tick force-recomputes *yesterday* a single time so runs/tool-calls that only landed near midnight are fully captured. The decision is the pure `deerflow.persistence.admin_stats.finalize.should_finalize_previous_day` (status/generated_at/now/hour) — the rebuild stamps `generated_at` to today, which self-debounces it for the rest of the day (no extra state file). Tests: `tests/test_leaderboard_finalize.py`.
### Scheduled Task Runtime ### Scheduled Task Runtime
`ScheduledTaskService` (`deerflow/runtime/scheduler/service.py`) polls due tasks every `poll_interval_seconds` and executes each in its own throwaway thread, serialized by a shared `_agent_run_lock`. `ScheduledTaskService` (`deerflow/runtime/scheduler/service.py`) polls due tasks every `poll_interval_seconds` and executes each in its own throwaway thread, serialized by a shared `_agent_run_lock`.
@@ -311,6 +319,16 @@ Two `TypeDecorator`s keep ORM columns correct across the `sqlite` / `postgres` /
- **`PortableLongText`** — a large text column: maps to MySQL `LONGTEXT` (4 GB) and plain `TEXT` on SQLite/Postgres (already unbounded). Use for big JSON/markdown blobs that would overflow MySQL's 64 KB `TEXT` cap — e.g. `ai_writing_sessions.transcript`. - **`PortableLongText`** — a large text column: maps to MySQL `LONGTEXT` (4 GB) and plain `TEXT` on SQLite/Postgres (already unbounded). Use for big JSON/markdown blobs that would overflow MySQL's 64 KB `TEXT` cap — e.g. `ai_writing_sessions.transcript`.
- **`BeijingDateTime`** — stores timestamps as Beijing (UTC+8) wall-clock time. The app records time with `datetime.now(UTC)`, but MySQL `DATETIME` / SQLite have no timezone and persist whatever wall clock the driver serializes — so a raw `DateTime` column would store UTC and read 8 hours behind Beijing. `BeijingDateTime` converts UTC→Beijing on bind and re-attaches `+08:00` on result, so the database holds Beijing time while Python still gets timezone-aware datetimes (instant comparisons stay correct, including in WHERE clauses). PostgreSQL `timestamptz` stores the instant unambiguously, so values pass through unchanged there. Uses a fixed UTC+8 offset (China has no DST), so no `tzdata` dependency. - **`BeijingDateTime`** — stores timestamps as Beijing (UTC+8) wall-clock time. The app records time with `datetime.now(UTC)`, but MySQL `DATETIME` / SQLite have no timezone and persist whatever wall clock the driver serializes — so a raw `DateTime` column would store UTC and read 8 hours behind Beijing. `BeijingDateTime` converts UTC→Beijing on bind and re-attaches `+08:00` on result, so the database holds Beijing time while Python still gets timezone-aware datetimes (instant comparisons stay correct, including in WHERE clauses). PostgreSQL `timestamptz` stores the instant unambiguously, so values pass through unchanged there. Uses a fixed UTC+8 offset (China has no DST), so no `tzdata` dependency.
### Schema Migrations / Manual Column-Add SQL (`scripts/sql/`)
Alembic auto-migration is **disabled** — `_run_alembic_upgrade(url)` is commented out in `persistence/engine.py`. On startup only `Base.metadata.create_all` runs, which creates **new tables** but never patches **existing** ones (no new columns, no new indexes). So a column added to an ORM model appears automatically on a brand-new deployment, but is **missing on any existing MySQL database** — and code reading/writing that column will error there until the column is added by hand.
**Convention (required):** whenever code adds a field to an existing table (a new `mapped_column` on an ORM model), you must, in the same change:
1. Keep the matching alembic version file under `persistence/migrations/versions/` (guarded with `_has_table`/`_has_column`), and
2. Append the idempotent `ADD COLUMN` statement to **`scripts/sql/column_additions_mysql.sql`**, with a definition matching the ORM column exactly (type, length, nullability).
Operators upgrading an existing DB run the whole file (`USE <db>;` then execute) — it is idempotent (existing columns are skipped via `information_schema.columns`), so it is safe to re-run. Index-only changes go in the sibling `scripts/sql/leaderboard_indexes_mysql.sql` (same idempotent pattern).
### Sandbox System (`packages/harness/deerflow/sandbox/`) ### Sandbox System (`packages/harness/deerflow/sandbox/`)
**Interface**: Abstract `Sandbox` with `execute_command`, `read_file`, `write_file`, `list_dir` **Interface**: Abstract `Sandbox` with `execute_command`, `read_file`, `write_file`, `list_dir`
@@ -380,7 +398,7 @@ Two `TypeDecorator`s keep ORM columns correct across the `sqlite` / `postgres` /
- **Loading**: `load_skills()` recursively scans `skills/{public,custom}` for `SKILL.md`, parses metadata, and reads enabled state from extensions_config.json - **Loading**: `load_skills()` recursively scans `skills/{public,custom}` for `SKILL.md`, parses metadata, and reads enabled state from extensions_config.json
- **Injection**: Enabled skills listed in agent system prompt with container paths - **Injection**: Enabled skills listed in agent system prompt with container paths
- **Installation**: `POST /api/skills/install` extracts .skill ZIP archive to custom/ directory - **Installation**: `POST /api/skills/install` extracts .skill ZIP archive to custom/ directory
- **Chinese name (`name_zh`)**: a Chinese display name stored in the **`skills` DB table** (`SkillRow.name_zh`, `String(128)`, migration `20260609_01`) — never written to the on-disk SKILL.md, so it works for built-in skills too. Threaded through `SkillStore` create/ensure_legacy/update/update_admin exactly like `detail`; surfaced on `SkillResponse.name_zh` from the ownership record in `_skill_to_response`. Authorization mirrors `detail`: custom skill → owner or admin; built-in/public → admin only (row materialized on demand via `ensure_legacy`). Endpoints (in `app/gateway/routers/skills.py`): `PUT /api/skills/{name}/name-zh` (set/clear), `POST /api/skills/{name}/name-zh/generate` (LLM generates one — reuses the curator model setting, falls back to the main model), `POST /api/skills/name-zh/generate-missing` (一键补充: bulk-generate for every editable skill that lacks one, bounded-concurrency `asyncio.gather`, skips skills the caller can't edit). The frontend (`skill-settings-page.tsx` list + `skill-detail-dialog.tsx` editor, plus the agent skill pickers/badges in `NewAgentPage.tsx` + `agent-card.tsx`) shows `name_zh || name` and offers per-skill "AI 生成" plus a header "一键补充中文名" button. Tests: `tests/test_skill_name_zh.py`. - **Chinese name (`name_zh`)**: a Chinese display name stored in the **`skills` DB table** (`SkillRow.name_zh`, `String(128)`, migration `20260609_01`) — never written to the on-disk SKILL.md, so it works for built-in skills too. Threaded through `SkillStore` create/ensure_legacy/update/update_admin exactly like `detail`; surfaced on `SkillResponse.name_zh` from the ownership record in `_skill_to_response`. Authorization mirrors `detail`: custom skill → owner or admin; built-in/public → admin only (row materialized on demand via `ensure_legacy`). Endpoints (in `app/gateway/routers/skills.py`): `PUT /api/skills/{name}/name-zh` (set/clear), `POST /api/skills/{name}/name-zh/generate` (LLM generates one), `POST /api/skills/name-zh/generate-missing` (一键补充: **admin-only** bulk-generate for every skill that lacks one, bounded-concurrency `asyncio.gather`). **Optional model selection**: both generate endpoints accept an optional JSON body `{model_name}` — an explicit name wins, else the curator model setting, else the main model (`_resolve_skill_name_zh_model_name`). **Model-error surfacing**: `_generate_skill_name_zh` raises `SkillNameZhModelError` when the model itself fails (build/timeout/API error) — distinct from the model returning an unusable name (→ `None`). The single endpoint maps that to `502 大模型报错,无法补全技能中文名:…`; the bulk endpoint fails fast on a model-build error (502) and otherwise carries a representative `model_error` string in `SkillNameZhBulkResponse` so the UI can say "N 个因大模型报错失败". The frontend (`skill-settings-page.tsx` list + `skill-detail-dialog.tsx` editor, plus the agent skill pickers/badges in `NewAgentPage.tsx` + `agent-card.tsx`) shows `name_zh || name`, offers per-skill "AI 生成" plus an **admin-only** header "一键补充中文名" button, and a compact model dropdown (`components/workspace/model-name-select.tsx`, default = configured model) beside each action.
#### Skill Curator (`packages/harness/deerflow/skills/curator.py`) #### Skill Curator (`packages/harness/deerflow/skills/curator.py`)
@@ -474,6 +474,13 @@ _LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS = _env_float(
"ADMIN_LEADERBOARD_BACKGROUND_STARTUP_DELAY_SECONDS", "ADMIN_LEADERBOARD_BACKGROUND_STARTUP_DELAY_SECONDS",
60.0, 60.0,
) )
# Hour (local Beijing time, 0-23) after which the scheduler force-recomputes the
# *previous* day's snapshot once, so records that landed near midnight are fully
# captured. Default 1 → "每天凌晨 1 点重新统计前一天".
_LEADERBOARD_FINALIZE_HOUR = min(
23,
_env_int("ADMIN_LEADERBOARD_FINALIZE_HOUR", 1, minimum=0),
)
def _leaderboard_scheduler_sql_for_log(stmt) -> str: def _leaderboard_scheduler_sql_for_log(stmt) -> str:
@@ -653,6 +660,58 @@ async def _claim_leaderboard_snapshot(
return True return True
async def _maybe_finalize_previous_leaderboard_day(session_factory, settings_hash: str) -> bool:
"""Once per day after ``_LEADERBOARD_FINALIZE_HOUR``, force-recompute yesterday.
A ``ready`` past-day snapshot is otherwise treated as fresh forever and never
rebuilt. But the previous day's snapshot is usually generated before midnight,
so it can miss runs/tool-calls that only land in the DB around the day
boundary. This re-queues yesterday (``force=True``) a single time; the rebuild
stamps ``generated_at`` to today, which stops it from re-firing for the rest of
the day.
"""
from datetime import timedelta
from app.gateway.routers.admin_users import (
_as_beijing,
_queue_daily_leaderboard_snapshots,
)
from deerflow.persistence.admin_stats.finalize import should_finalize_previous_day
from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow
from deerflow.persistence.types import BEIJING_TZ
now = datetime.now(BEIJING_TZ)
yesterday = (now.date() - timedelta(days=1)).isoformat()
async with session_factory() as session:
row = await session.get(
AdminLeaderboardDailyStatRow,
{"stat_date": yesterday, "settings_hash": settings_hash},
)
generated_at = _as_beijing(row.generated_at) if (row is not None and row.generated_at is not None) else None
if not should_finalize_previous_day(
status=(row.status if row is not None else None),
generated_at=generated_at,
now=now,
finalize_hour=_LEADERBOARD_FINALIZE_HOUR,
):
return False
queued = await _queue_daily_leaderboard_snapshots(
session_factory,
[yesterday],
settings_hash,
force=True,
message="每日凌晨重新统计前一天完整数据,确保全部记录纳入统计。",
)
if queued:
logger.info(
"Leaderboard snapshot finalize: queued previous day %s for full recompute (hash=%s)",
yesterday,
settings_hash[:10],
)
return True
return False
async def _run_leaderboard_snapshot_tick() -> tuple[int, int]: async def _run_leaderboard_snapshot_tick() -> tuple[int, int]:
from app.gateway.routers.admin_users import ( from app.gateway.routers.admin_users import (
_daily_job_key, _daily_job_key,
@@ -677,6 +736,12 @@ async def _run_leaderboard_snapshot_tick() -> tuple[int, int]:
message="排行榜后台统计任务定期预热/补算队列。", message="排行榜后台统计任务定期预热/补算队列。",
) )
# 每天凌晨(默认 1 点)后强制重算"前一天"的完整快照一次,覆盖跨午夜才落库的记录。
try:
await _maybe_finalize_previous_leaderboard_day(session_factory, settings_hash)
except Exception:
logger.exception("Leaderboard snapshot finalize check failed (non-fatal)")
due_dates = await _leaderboard_snapshot_due_dates( due_dates = await _leaderboard_snapshot_due_dates(
session_factory, session_factory,
settings_hash=settings_hash, settings_hash=settings_hash,
@@ -2178,6 +2178,72 @@ async def queue_leaderboard_snapshot_backfill(
) )
@router.post("/leaderboard/refresh-today", response_model=AdminLeaderboardResponse)
async def refresh_today_leaderboard(
request: Request,
days: int = Query(default=30, ge=1, le=365),
since: str | None = Query(default=None, description="Inclusive range start, ISO datetime"),
until: str | None = Query(default=None, description="Inclusive range end, ISO datetime"),
exclude_user_ids: list[str] | None = Query(default=None, description="User ids excluded from this leaderboard aggregation"),
include_scheduled: bool | None = Query(default=None, description="Include runs created by scheduled tasks"),
include_admins: bool | None = Query(default=None, description="Include admin accounts in leaderboard aggregation"),
include_failed: bool | None = Query(default=None, description="Include failed runs and LLM calls"),
limit: int = Query(default=20, ge=5, le=100),
) -> AdminLeaderboardResponse:
"""Force a synchronous rebuild of *today's* snapshot, then return the range.
Today's snapshot otherwise refreshes only every ``_LEADERBOARD_TODAY_TTL_SECONDS``
(~5 min) via the background scheduler. This endpoint lets an admin pull the
latest real-time records into the stats on demand: it rebuilds today's daily
snapshot immediately (bounded by the same semaphore/timeout the scheduler uses)
and then returns the merged leaderboard for the requested range.
"""
await _require_admin(request)
session_factory = get_session_factory()
if session_factory is None:
raise HTTPException(status_code=503, detail="Analytics requires database persistence")
settings = load_system_settings().leaderboard
(
excluded,
resolved_include_scheduled,
resolved_include_admins,
resolved_include_failed,
settings_hash,
) = _resolve_leaderboard_snapshot_scope(
settings,
exclude_user_ids=exclude_user_ids,
include_scheduled=include_scheduled,
include_admins=include_admins,
include_failed=include_failed,
)
today = _today_stat_date()
await _refresh_daily_leaderboard_snapshot(
_daily_job_key(today, settings_hash),
session_factory=session_factory,
settings=settings,
stat_date=today,
settings_hash=settings_hash,
excluded=excluded,
include_scheduled=resolved_include_scheduled,
include_admins=resolved_include_admins,
include_failed=resolved_include_failed,
)
since_dt, until_dt, _range_days = _resolve_range(days, since, until)
return await _load_or_schedule_daily_leaderboard(
session_factory,
settings,
since_dt,
until_dt,
limit=limit,
exclude_user_ids=exclude_user_ids,
include_scheduled=include_scheduled,
include_admins=include_admins,
include_failed=include_failed,
)
async def _build_leaderboard_report_prompt( async def _build_leaderboard_report_prompt(
session_factory, session_factory,
settings, settings,
@@ -0,0 +1,47 @@
"""Pure scheduling helpers for the admin leaderboard daily snapshots.
Kept import-light on purpose (only stdlib ``datetime``) so the decision logic
is unit-testable without standing up the FastAPI app, the DB engine, or the
ORM. The background scheduler in ``app/gateway/app.py`` wires this to the real
snapshot rows.
"""
from __future__ import annotations
from datetime import datetime
def should_finalize_previous_day(
*,
status: str | None,
generated_at: datetime | None,
now: datetime,
finalize_hour: int,
) -> bool:
"""Decide whether *yesterday's* snapshot should be force-recomputed once.
The previous day's snapshot is normally built before midnight, so it can
miss runs / tool calls whose rows only land in the database shortly before
or after the day boundary. Once per day, on/after ``finalize_hour`` (local
Beijing time), we recompute it a single time so the full day is captured.
``now`` and ``generated_at`` are expected to be timezone-aware Beijing
datetimes (``generated_at`` may be ``None`` for legacy rows).
Returns ``True`` only when all hold:
- it is already on/after ``finalize_hour`` today;
- a snapshot row exists (``status`` is not ``None`` — a missing day is
left to the normal prewarm/queue path);
- that snapshot is not already ``queued``/``running`` (a finalize already
in flight, or normal processing);
- it was generated before today (i.e. not yet finalized for this day).
"""
if status is None:
return False
if now.hour < finalize_hour:
return False
if status in ("queued", "running"):
return False
if generated_at is not None and generated_at.date() >= now.date():
return False
return True
@@ -0,0 +1,74 @@
-- ============================================================================
-- 给【已存在的表】补列(ADD COLUMN)—— MySQL 手动执行版
-- ============================================================================
--
-- 背景:
-- 本项目的 alembic 自动迁移是【关闭】的(engine.py 中
-- `# await _run_alembic_upgrade(url)` 被注释)。`Base.metadata.create_all`
-- 只会建【新表】,对【已存在的表】既不会补建索引、也不会补加新列。
-- 因此当我们在某个 ORM model 上新增了一列时,【全新部署】会随建表一并带上,
-- 但【已有数据的 MySQL 部署】上这一列并不会自动出现,代码读写该列会报错。
--
-- 约定(重要):
-- 今后只要有代码给某张【已存在的表】新增字段(在 ORM model 上加列),就必须
-- 把对应的 `ADD COLUMN` 语句以下面的【幂等】写法补进本文件,并在迁移目录里
-- 保留对应的 alembic 版本文件,二者保持一致。运维在升级老库时整文件执行本脚本
-- 即可补齐所有缺失的列。
--
-- 用法:
-- 1) 先选中目标库(与 database.mysql_url 里的库名一致):
-- USE your_deerflow_db;
-- 2) 直接整文件执行即可。本脚本是【幂等】的:已存在的列会跳过,不会报错,
-- 可重复执行。
--
-- 注意:
-- * 列定义必须与 ORM model / alembic migration 完全一致(类型、长度、可空性)。
-- * 新增列一律建议 NULL 或带 DEFAULT,避免在大表上回填导致长时间锁表。
-- * MySQL 8.0 在线 DDL 默认 ALGORITHM=INPLACE 一般不阻塞写入,但仍建议低峰执行。
-- ============================================================================
-- ---- 通用幂等封装:仅当列不存在时才新增 --------------------------------------
DELIMITER //
DROP PROCEDURE IF EXISTS _add_column_if_missing //
CREATE PROCEDURE _add_column_if_missing(
IN p_table VARCHAR(64),
IN p_column VARCHAR(64),
IN p_definition VARCHAR(255) -- 列定义,如 'VARCHAR(128) NULL DEFAULT NULL'
)
BEGIN
DECLARE v_exists INT DEFAULT 0;
SELECT COUNT(*) INTO v_exists
FROM information_schema.columns
WHERE table_schema = DATABASE()
AND table_name = p_table
AND column_name = p_column;
IF v_exists = 0 THEN
SET @ddl = CONCAT('ALTER TABLE `', p_table, '` ADD COLUMN `', p_column, '` ', p_definition);
PREPARE stmt FROM @ddl;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;
SELECT CONCAT('ADDED ', p_table, '.', p_column) AS result;
ELSE
SELECT CONCAT('SKIPPED ', p_table, '.', p_column, ' (already exists)') AS result;
END IF;
END //
DELIMITER ;
-- ---- skills ------------------------------------------------------------------
-- name_zh: 技能的中文显示名(用户配置或 LLM 生成),对应 migration 20260609_01。
-- ORM: SkillRow.name_zh = String(128), nullable=True。
CALL _add_column_if_missing('skills', 'name_zh', 'VARCHAR(128) NULL DEFAULT NULL');
-- detail: 技能的长文介绍,对应 migration 20260518_01。
-- ORM: SkillRow.detail = Text, nullable=True。
CALL _add_column_if_missing('skills', 'detail', 'TEXT NULL DEFAULT NULL');
-- ---- 清理临时存储过程 --------------------------------------------------------
DROP PROCEDURE IF EXISTS _add_column_if_missing;
-- ---- 执行后自检:确认本批次涉及的列都已存在 ---------------------------------
SELECT table_name, column_name
FROM information_schema.columns
WHERE table_schema = DATABASE()
AND (table_name = 'skills' AND column_name IN ('name_zh', 'detail'))
ORDER BY table_name, column_name;