feat(server,contracts,web): dashboard library counts and engine runs triggeredBy/multi-status filters
- GET /dashboard: automation.library {entities, movies, series, seasons,
episodes, assets} and automation.slotSatisfiedRate; home media overview
shows the counts instead of placeholders.
- GET /engine/runs: `status` accepts comma-separated values; new
`triggeredBy` filter (comma-separated) evaluated server-side by scanning
batches and paging with the last inspected row's cursor (scan cap 2000).
- Task stream: source filter and the "active" tab (running,pending) are now
one paginated server query.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
rootandClaude Opus 5.5 committed
1 parent
615917bb45
commit
363f85b1a1
13 files changed
+165
-250
No files matched your search
@@ -445,6 +445,12 @@ describe.skipIf(!hasPg)("acquisition-organize 纵切(真实 PG + mock qB + 临
|
||||
expect(dash.body.data.recentOutput.length).toBeGreaterThan(0);
|
||||
expect(dash.body.data.cache.volumes[0]).toMatchObject({ hardThresholdPercent: 90, resumePercent: 85 });
|
||||
expect(dash.body.data.needsAttention.total).toBeGreaterThan(0);
|
||||
const auto = dash.body.data.automation;
|
||||
expect(auto.library.entities).toBeGreaterThanOrEqual(1);
|
||||
expect(auto.library.movies).toBeGreaterThanOrEqual(1);
|
||||
expect(auto.library.assets).toBeGreaterThanOrEqual(1);
|
||||
const slotTotal = Object.values(auto.slotCounts as Record<string, number>).reduce((a, b) => a + b, 0);
|
||||
expect(auto.slotSatisfiedRate).toBe(slotTotal ? auto.slotCounts.satisfied / slotTotal : null);
|
||||
const libList = await h.api("GET", "/library?q=沙丘");
|
||||
expect(libList.body.data.items[0]).toMatchObject({ entity: { id: movieId }, assetCount: 1 });
|
||||
const det = await h.api("GET", `/library/${movieId}`);
|
||||
|
||||
@@ -63,6 +63,26 @@ export function registerDashboardRoutes(app: ModuleApp, d: AoDeps) {
|
||||
attention: 0,
|
||||
};
|
||||
for (const r of slotRows) slotCounts[r.status] = r.n;
|
||||
const slotTotal = Object.values(slotCounts).reduce((a, b) => a + b, 0);
|
||||
const library = (
|
||||
await sql<{
|
||||
entities: number;
|
||||
movies: number;
|
||||
series: number;
|
||||
seasons: number;
|
||||
episodes: number;
|
||||
assets: number;
|
||||
}>`
|
||||
select
|
||||
(select count(*) from media_entity where status <> 'merged')::int as entities,
|
||||
(select count(*) from media_entity where status <> 'merged' and kind = 'movie')::int as movies,
|
||||
(select count(*) from media_entity where status <> 'merged' and kind in ('series','variety'))::int as series,
|
||||
(select count(*) from content_unit where kind = 'season' and status <> 'retired')::int as seasons,
|
||||
(select count(*) from content_unit where kind = 'episode' and status <> 'retired')::int as episodes,
|
||||
(select count(*) from media_asset where status = 'active' and kind in ('video','bluray_dir'))::int as assets`.execute(
|
||||
db,
|
||||
)
|
||||
).rows[0]!;
|
||||
|
||||
const dlRows = await db
|
||||
.selectFrom("download_task")
|
||||
@@ -203,6 +223,8 @@ export function registerDashboardRoutes(app: ModuleApp, d: AoDeps) {
|
||||
},
|
||||
activeSubscriptions: today.subs,
|
||||
slotCounts,
|
||||
slotSatisfiedRate: slotTotal ? slotCounts.satisfied / slotTotal : null,
|
||||
library,
|
||||
},
|
||||
currentWork: {
|
||||
downloading: count("downloading"),
|
||||
|
||||
@@ -114,7 +114,7 @@ const RUN_COLS = sql.raw(`id, workflow_name, version, status, attempts, worker_i
|
||||
export interface EngineQueries {
|
||||
listRuns(
|
||||
q: z.infer<typeof EngineRunQuery>,
|
||||
): Promise<{ items: ReturnType<typeof runView>[]; nextCursor: string | null }>;
|
||||
): Promise<{ items: ReturnType<typeof runView>[]; nextCursor: string | null; cursors: string[] }>;
|
||||
getRun(id: string): Promise<z.input<typeof EngineRunDetail> | null>;
|
||||
stats(): Promise<z.input<typeof EngineStats>>;
|
||||
}
|
||||
@@ -127,7 +127,7 @@ export function engineQueries(db: Kysely<DB>, schema = "openworkflow"): EngineQu
|
||||
async listRuns(q) {
|
||||
const cur: Cursor | null = decodeCursor(q.cursor);
|
||||
const conds = [sql`namespace_id = ${NAMESPACE}`];
|
||||
if (q.status) conds.push(sql`status = ${q.status}`);
|
||||
if (q.status?.length) conds.push(sql`status = any(${q.status}::text[])`);
|
||||
if (q.workflowName) conds.push(sql`workflow_name = ${q.workflowName}`);
|
||||
if (q.from) conds.push(sql`created_at >= ${new Date(q.from)}`);
|
||||
if (q.to) conds.push(sql`created_at < ${new Date(q.to)}`);
|
||||
@@ -144,6 +144,8 @@ export function engineQueries(db: Kysely<DB>, schema = "openworkflow"): EngineQu
|
||||
items: items.map(runView),
|
||||
nextCursor:
|
||||
rows.rows.length > q.limit && last ? encodeCursor({ us: Number(last.cursor_us), id: last.id }) : null,
|
||||
// 每行的游标(triggeredBy 逐页筛选时在中途截断用)
|
||||
cursors: items.map((r) => encodeCursor({ us: Number(r.cursor_us), id: r.id })),
|
||||
};
|
||||
},
|
||||
|
||||
@@ -242,6 +244,10 @@ export function engineQueries(db: Kysely<DB>, schema = "openworkflow"): EngineQu
|
||||
};
|
||||
}
|
||||
|
||||
/** triggeredBy 筛选:每批扫描行数与单次请求扫描上限 */
|
||||
const TRIGGER_SCAN_BATCH = 200;
|
||||
const TRIGGER_SCAN_CAP = 2000;
|
||||
|
||||
export function registerEngineRoutes(app: ModuleApp, ctx: ServerContext, opts: { schema?: string } = {}) {
|
||||
let q: EngineQueries | null = null;
|
||||
const queries = () => {
|
||||
@@ -251,8 +257,32 @@ export function registerEngineRoutes(app: ModuleApp, ctx: ServerContext, opts: {
|
||||
// W9-I:列表与详情补 business 字段(engine-business.ts;查询失败降级为 null)
|
||||
const schema = opts.schema ?? "openworkflow";
|
||||
endpoint(app, D.runs, async ({ query }) => {
|
||||
const page = await queries().listRuns(query);
|
||||
return { ...page, items: await attachBusiness(ctx.db, schema, page.items) };
|
||||
const triggered = query.triggeredBy?.length ? new Set(query.triggeredBy) : null;
|
||||
if (!triggered) {
|
||||
const page = await queries().listRuns(query);
|
||||
return { items: await attachBusiness(ctx.db, schema, page.items), nextCursor: page.nextCursor };
|
||||
}
|
||||
// 触发来源是逐行推断的(engine-business.ts),不能下推成 SQL:按批扫描、补字段后筛选,
|
||||
// 凑够 limit 或扫满 TRIGGER_SCAN_CAP 行就返回;游标停在最后检查的那一行,可能少于 limit 但仍有下一页
|
||||
const out: Awaited<ReturnType<typeof attachBusiness<ReturnType<typeof runView>>>> = [];
|
||||
let cursor = query.cursor;
|
||||
let scanned = 0;
|
||||
for (;;) {
|
||||
const page = await queries().listRuns({ ...query, cursor, limit: TRIGGER_SCAN_BATCH });
|
||||
const items = await attachBusiness(ctx.db, schema, page.items);
|
||||
for (let i = 0; i < items.length; i++) {
|
||||
scanned += 1;
|
||||
const r = items[i]!;
|
||||
if (r.business?.triggeredBy && triggered.has(r.business.triggeredBy)) out.push(r);
|
||||
if (out.length === query.limit) {
|
||||
const more = i < items.length - 1 || page.nextCursor !== null;
|
||||
return { items: out, nextCursor: more ? page.cursors[i]! : null };
|
||||
}
|
||||
}
|
||||
if (!page.nextCursor) return { items: out, nextCursor: null };
|
||||
cursor = page.nextCursor;
|
||||
if (scanned >= TRIGGER_SCAN_CAP) return { items: out, nextCursor: cursor };
|
||||
}
|
||||
});
|
||||
endpoint(app, D.run, async ({ params }) => {
|
||||
const r = await queries().getRun(params.id);
|
||||
|
||||
@@ -298,6 +298,9 @@ describe.skipIf(!hasPg)("W2-A 管理 API(真实 PG)", () => {
|
||||
expect(p2.data.items.map((r: { id: string }) => r.id)).toEqual(["r1"]);
|
||||
expect(p2.data.nextCursor).toBeNull();
|
||||
expect((await admin("GET", "/engine/runs?status=bogus")).status).toBe(422);
|
||||
const multi = await json(await admin("GET", "/engine/runs?status=running,pending"));
|
||||
expect(multi.data.items.map((r: { id: string }) => r.id)).toEqual(["r4", "r3"]);
|
||||
expect((await admin("GET", "/engine/runs?status=running,bogus")).status).toBe(422);
|
||||
});
|
||||
|
||||
it("run 详情:step attempts + signals;敏感字段脱敏,不返回 config/context", async () => {
|
||||
|
||||
@@ -353,6 +353,28 @@ describe.skipIf(!hasPg)("W9-I 进程与健康(真实 PG)", () => {
|
||||
expect(s.data.items.map((r: { id: string }) => r.id)).toEqual(["b4"]);
|
||||
expect((await admin("GET", "/engine/runs?businessKind=bogus")).status).toBe(422);
|
||||
|
||||
// triggeredBy:服务端逐页推断后筛选;多值逗号分隔;limit 截断时游标接着扫
|
||||
const range = "from=2026-09-10T00:00:00Z&to=2026-09-11T00:00:00Z";
|
||||
const ids = (r: { data: { items: { id: string }[] } }) => r.data.items.map((x) => x.id);
|
||||
expect(ids(await json(await admin("GET", `/engine/runs?${range}&triggeredBy=schedule`)))).toEqual(["b4", "b2"]);
|
||||
expect(ids(await json(await admin("GET", `/engine/runs?${range}&triggeredBy=event,api_key`)))).toEqual([
|
||||
"b3",
|
||||
"b1",
|
||||
]);
|
||||
const t1 = await json(await admin("GET", `/engine/runs?${range}&triggeredBy=schedule&limit=1`));
|
||||
expect(ids(t1)).toEqual(["b4"]);
|
||||
expect(t1.data.nextCursor).toBeTruthy();
|
||||
const t2 = await json(
|
||||
await admin("GET", `/engine/runs?${range}&triggeredBy=schedule&limit=1&cursor=${t1.data.nextCursor}`),
|
||||
);
|
||||
expect(ids(t2)).toEqual(["b2"]);
|
||||
const t3 = await json(
|
||||
await admin("GET", `/engine/runs?${range}&triggeredBy=schedule&limit=1&cursor=${t2.data.nextCursor}`),
|
||||
);
|
||||
expect(ids(t3)).toEqual([]);
|
||||
expect(t3.data.nextCursor).toBeNull();
|
||||
expect((await admin("GET", "/engine/runs?triggeredBy=bogus")).status).toBe(422);
|
||||
|
||||
const detail = (await json(await admin("GET", "/engine/runs/b2"))).data;
|
||||
expect(detail.business).toMatchObject({ kind: "agent", refId: agentId });
|
||||
expect(detail.stepAttempts).toEqual([]);
|
||||
|
||||
@@ -5,7 +5,10 @@ import { qk } from "~/lib/query-keys.ts";
|
||||
import { errorCode } from "./parse.ts";
|
||||
|
||||
export type EngineRunFilters = {
|
||||
/** 单值或逗号分隔多值(running,pending) */
|
||||
status?: string;
|
||||
/** 触发来源,单值或逗号分隔多值(user,api_key) */
|
||||
triggeredBy?: string;
|
||||
workflowName?: string;
|
||||
businessKind?: EngineBusinessKind;
|
||||
businessRefId?: string;
|
||||
@@ -16,9 +19,10 @@ export type EngineRunFilters = {
|
||||
/** 终态:不能再 cancel */
|
||||
export const ENGINE_TERMINAL = new Set(["completed", "succeeded", "failed", "canceled"]);
|
||||
|
||||
export function useEngineRuns(filters: EngineRunFilters, enabled = true) {
|
||||
export function useEngineRuns(filters: EngineRunFilters, enabled = true, refetchMs: number | false = false) {
|
||||
return useInfiniteQuery({
|
||||
enabled,
|
||||
refetchInterval: refetchMs,
|
||||
queryKey: qk.list("engine", filters),
|
||||
initialPageParam: undefined as string | undefined,
|
||||
queryFn: ({ pageParam, signal }) =>
|
||||
|
||||
@@ -25,6 +25,13 @@ export const SOURCE_LABEL: Record<TaskSource, string> = {
|
||||
manual: "手动",
|
||||
agent: "Agent",
|
||||
};
|
||||
/** 来源 → 服务端 triggeredBy 筛选值(逗号分隔多值) */
|
||||
export const SOURCE_TRIGGERED_BY: Record<TaskSource, string> = {
|
||||
schedule: "schedule",
|
||||
event: "event",
|
||||
manual: "user,api_key",
|
||||
agent: "agent",
|
||||
};
|
||||
export function sourceOf(t: EngineTriggeredBy | null | undefined): TaskSource | null {
|
||||
if (!t) return null;
|
||||
if (t === "user" || t === "api_key") return "manual";
|
||||
|
||||
@@ -3,7 +3,7 @@ import type { EngineBusinessKind, EngineRun, ScheduledJob } from "@umt/contracts
|
||||
import { CalendarClock, ExternalLink, History, ListTodo, X } from "lucide-react";
|
||||
import { useMemo, useRef, useState } from "react";
|
||||
import { Link } from "react-router";
|
||||
import { useEngineRuns, useEngineRunsPage } from "~/api/engine.ts";
|
||||
import { useEngineRuns } from "~/api/engine.ts";
|
||||
import { useScheduleBoard } from "~/api/schedules.ts";
|
||||
import { DataTable } from "~/components/console/data-table.tsx";
|
||||
import { FilterBar, FilterSelect } from "~/components/console/filter-bar.tsx";
|
||||
@@ -22,6 +22,7 @@ import {
|
||||
runStatus,
|
||||
runTitle,
|
||||
SOURCE_LABEL,
|
||||
SOURCE_TRIGGERED_BY,
|
||||
sourceOf,
|
||||
type TaskSource,
|
||||
workflowLabel,
|
||||
@@ -132,26 +133,20 @@ export function TaskStream({
|
||||
businessKind: (filters.kind || undefined) as EngineBusinessKind | undefined,
|
||||
businessRefId: filters.ref || undefined,
|
||||
workflowName: filters.workflow || undefined,
|
||||
triggeredBy: SOURCE_TRIGGERED_BY[filters.source as TaskSource] ?? undefined,
|
||||
from,
|
||||
};
|
||||
const active = tab === "active";
|
||||
// 进行中 = 进行中 + 排队(status 多值,服务端分页);来源筛选同样在服务端做
|
||||
const list = useEngineRuns(
|
||||
{
|
||||
...base,
|
||||
status: tab === "failed" ? "failed" : filters.status || undefined,
|
||||
status: active ? "running,pending" : tab === "failed" ? "failed" : filters.status || undefined,
|
||||
},
|
||||
tab !== "active",
|
||||
true,
|
||||
active ? 10_000 : false,
|
||||
);
|
||||
// 进行中 = 进行中 + 排队(两次取第一页合并)
|
||||
const running = useEngineRunsPage({ ...base, status: "running" }, 100, 10_000, tab === "active");
|
||||
const pending = useEngineRunsPage({ ...base, status: "pending" }, 100, 10_000, tab === "active");
|
||||
|
||||
const active = tab === "active";
|
||||
const loaded = active
|
||||
? running.data && pending.data
|
||||
? [...running.data.items, ...pending.data.items]
|
||||
: undefined
|
||||
: list.data?.pages.flatMap((p) => p.items);
|
||||
const rows = filters.source ? loaded?.filter((r) => sourceOf(r.business?.triggeredBy) === filters.source) : loaded;
|
||||
const rows = list.data?.pages.flatMap((p) => p.items);
|
||||
const filtered = Object.values(filters).some(Boolean);
|
||||
|
||||
return (
|
||||
@@ -212,17 +207,12 @@ export function TaskStream({
|
||||
<DataTable<EngineRun>
|
||||
rows={rows}
|
||||
rowKey={(r) => r.id}
|
||||
loading={active ? running.isPending || pending.isPending : list.isPending}
|
||||
error={active ? (running.error ?? pending.error) : list.error}
|
||||
onRetry={() => {
|
||||
if (active) {
|
||||
void running.refetch();
|
||||
void pending.refetch();
|
||||
} else void list.refetch();
|
||||
}}
|
||||
loading={list.isPending}
|
||||
error={list.error}
|
||||
onRetry={() => void list.refetch()}
|
||||
onRowClick={(r) => onOpen(r.id)}
|
||||
selectedKey={selectedId}
|
||||
hasMore={!active && !!list.hasNextPage}
|
||||
hasMore={!!list.hasNextPage}
|
||||
onLoadMore={() => list.fetchNextPage()}
|
||||
loadingMore={list.isFetchingNextPage}
|
||||
empty={{
|
||||
@@ -234,7 +224,7 @@ export function TaskStream({
|
||||
: tab === "failed"
|
||||
? "没有失败的任务"
|
||||
: "还没有任务",
|
||||
description: filters.source && loaded?.length ? "来源筛选只作用于已加载的记录,可加载更多再看。" : undefined,
|
||||
description: filters.source && list.hasNextPage ? "按来源筛选时一页可能较少,可加载更多继续查找。" : undefined,
|
||||
}}
|
||||
columns={[
|
||||
{
|
||||
|
||||
@@ -280,19 +280,18 @@ function NeedsMe({ n }: { n: Dashboard["needsAttention"] }) {
|
||||
);
|
||||
}
|
||||
|
||||
// ─── 媒体概览:版本目标满足率 + 今日产出(作品/季/资产计数待 /dashboard 提供) ───
|
||||
// ─── 媒体概览:作品/季/资产计数 + 版本目标满足率 + 今日产出 ───
|
||||
function MediaOverview({ a }: { a: Dashboard["automation"] }) {
|
||||
const c = a.slotCounts;
|
||||
const total = Object.values(c).reduce((x, y) => x + y, 0);
|
||||
const rate = total ? Math.round((c.satisfied / total) * 100) : null;
|
||||
const lib = (a as { library?: { entities?: number; seasons?: number; assets?: number } }).library;
|
||||
const num = (v: number | undefined) => (v === undefined ? "—" : v);
|
||||
const rate = a.slotSatisfiedRate === null ? null : Math.round(a.slotSatisfiedRate * 100);
|
||||
const lib = a.library;
|
||||
return (
|
||||
<Section title="媒体概览" icon={LibraryIcon} to={consolePath("library")}>
|
||||
<div className="grid grid-cols-3 gap-2">
|
||||
<StatTile plain label="作品" value={num(lib?.entities)} />
|
||||
<StatTile plain label="季" value={num(lib?.seasons)} />
|
||||
<StatTile plain label="资产" value={num(lib?.assets)} />
|
||||
<StatTile plain label="作品" value={lib.entities} hint={`电影 ${lib.movies} · 剧集 ${lib.series}`} />
|
||||
<StatTile plain label="季" value={lib.seasons} hint={`${lib.episodes} 集`} />
|
||||
<StatTile plain label="资产" value={lib.assets} hint="已入库视频" />
|
||||
</div>
|
||||
<div className="mt-3">
|
||||
<div className="flex items-baseline justify-between text-sm">
|
||||
|
||||
Reference in new issue
Block a user