Files
UHD_Media_Tool/apps/web/app/api/tasks.ts
T

340 lines
12 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import { useQuery, useQueryClient } from "@tanstack/react-query";
import {
type AcquisitionPlan,
cursorPage,
Decision,
type DownloadTask,
PublicOrganizationRecord,
SourceRelease,
Task,
type TaskKind,
type TaskStatus,
} from "@umt/contracts";
import { consolePath } from "~/components/console/nav.ts";
import { recordPath, subjectLink, taskPath } from "~/components/console/record-sheet.tsx";
import { DECISION_STATUS } from "~/components/library/labels.ts";
import { organizationActionLabel, organizationLabel } from "~/lib/labels.ts";
import { humanizeReason, reasonLine } from "~/lib/reasons.ts";
import { useSse } from "~/lib/sse.ts";
import { PAUSE_REASON_LABEL, PLAN_STATUS_LABEL } from "./downloads.ts";
import { apiParsed } from "./parse.ts";
// 业务任务视图:按 trace 汇总的业务时间线(参考 MovieClaw task-activity)
// 关联键全部由服务端给出:plan →(/tasks?subjectId=planId)submitDownload → download_task
// →(/tasks?subjectId=downloadTaskId)verifyAndImport →(/organization-records?downloadTaskId=)整理记录 → decision/agent_run。
const TaskPage = cursorPage(Task);
const RecordPage = cursorPage(PublicOrganizationRecord);
const DecisionPage = cursorPage(Decision);
const RESOLUTION_LABEL = {
unresolved: "未识别",
resolved: "已识别",
ambiguous: "匹配到多部作品,待确认",
unidentifiable: "无法识别",
rejected: "已排除",
} as const;
/** 时间线阶段:observation → plan → download → organization → agent */
export type TimelineStage = "observation" | "plan" | "download" | "organization" | "agent" | "engine";
export type TimelineTone = "success" | "warning" | "destructive" | "neutral";
export type TimelineEvent = {
key: string;
stage: TimelineStage;
at: string;
title: string;
detail?: string | null;
tone: TimelineTone;
href?: string;
/** 进行中的阶段(不是终态) */
ongoing?: boolean;
};
export function useSourceRelease(id: string | null | undefined) {
return useQuery({
queryKey: ["source-release", id ?? ""],
enabled: !!id,
queryFn: ({ signal }) =>
apiParsed(SourceRelease, "/api/v1/source-releases/:id", { pathParams: { id: id ?? "" }, signal }),
});
}
/** 与该下载有关的待处理事项(subject = 下载任务或计划),含 Agent 调查 run */
export function useSubjectDecisions(subjectIds: string[]) {
const ids = subjectIds.filter(Boolean);
const qc = useQueryClient();
useSse(
["decision.created", "decision.updated"],
() => {
void qc.invalidateQueries({ queryKey: ["decisions", "subjects", ids] });
},
ids.length > 0,
);
return useQuery({
queryKey: ["decisions", "subjects", ids],
enabled: ids.length > 0,
queryFn: ({ signal }) => apiParsed(DecisionPage, "/api/v1/decisions", { params: { limit: 200 }, signal }),
select: (d) => d.items.filter((x) => x.subjectId && ids.includes(x.subjectId)),
});
}
/** 业务任务:submitDownload 的 subject 是计划,verifyAndImport 的 subject 是下载任务(服务端写入 task 表) */
export function useTraceTasks(opts: { planId?: string | null; downloadTaskId?: string }) {
const ids = [opts.planId, opts.downloadTaskId].filter((x): x is string => !!x);
return useQuery({
queryKey: ["tasks", "trace", ids],
enabled: ids.length > 0,
queryFn: async ({ signal }): Promise<Task[]> => {
const pages = await Promise.all(
ids.map((subjectId) => apiParsed(TaskPage, "/api/v1/tasks", { params: { subjectId, limit: 50 }, signal })),
);
return pages.flatMap((p) => p.items);
},
refetchInterval: (q) => (q.state.data?.some((t) => ACTIVE_TASK.includes(t.status)) ? 5_000 : false),
});
}
const ACTIVE_TASK: TaskStatus[] = ["pending", "running", "waiting"];
/** 这个下载产生的整理记录(服务端按 download_task_id 关联) */
export function useRecordsForDownload(downloadTaskId: string | undefined) {
return useQuery({
queryKey: ["records", "download", downloadTaskId ?? ""],
enabled: !!downloadTaskId,
queryFn: ({ signal }) =>
apiParsed(RecordPage, "/api/v1/organization-records", {
params: { downloadTaskId: downloadTaskId ?? "", limit: 50 },
signal,
}),
});
}
const TASK_STATUS_TEXT: Record<string, string> = {
pending: "排队中",
running: "运行中",
waiting: "等待中",
succeeded: "成功",
failed: "失败",
cancelled: "已取消",
dead: "放弃",
};
const TASK_KIND_LABEL: Partial<Record<TaskKind, string>> = {
submitDownload: "提交下载",
verifyAndImport: "核验整理",
evaluateAcquisition: "评估资源",
cleanupDownloadCache: "清理缓存",
resolveDecision: "处理需要处理的事项",
};
const TASK_STATUS_TONE: Record<TaskStatus, TimelineTone> = {
pending: "neutral",
running: "neutral",
waiting: "warning",
succeeded: "success",
failed: "destructive",
cancelled: "neutral",
dead: "destructive",
};
const RECORD_TONE: Record<PublicOrganizationRecord["status"], TimelineTone> = {
queued: "neutral",
running: "neutral",
succeeded: "success",
partial: "warning",
failed: "destructive",
cancelled: "neutral",
};
/** 汇总一个下载 trace 的业务事件,按时间升序 */
export function buildTimeline(input: {
task: DownloadTask;
release?: SourceRelease | null;
plan?: AcquisitionPlan | null;
records?: PublicOrganizationRecord[];
decisions?: Decision[];
tasks?: Task[];
}): TimelineEvent[] {
const { task, release, plan, records = [], decisions = [], tasks = [] } = input;
const ev: TimelineEvent[] = [];
if (release) {
ev.push({
key: "obs",
stage: "observation",
at: release.firstSeenAt,
title: "首次发现资源",
detail: release.title,
tone: "neutral",
});
if (release.resolutionStatus !== "unresolved")
ev.push({
key: "resolve",
stage: "observation",
// 合同里没有识别完成时间,挂在首次发现之后,只表达结果
at: release.firstSeenAt,
title:
release.resolutionStatus === "resolved"
? "识别结果:已关联本地作品"
: `识别结果:${RESOLUTION_LABEL[release.resolutionStatus]}`,
tone: release.resolutionStatus === "resolved" ? "success" : "warning",
});
}
if (plan) {
ev.push({
key: "plan",
stage: "plan",
at: plan.createdAt,
title: `准备下载(${plan.purpose === "fill" ? "补缺" : plan.purpose === "upgrade" ? "升级" : "换源"})`,
detail: humanizeReason(plan.statusReason),
tone: "neutral",
});
if (plan.updatedAt !== plan.createdAt)
ev.push({
key: "plan-status",
stage: "plan",
at: plan.updatedAt,
title: `下载准备:${PLAN_STATUS_LABEL[plan.status]}`,
detail: humanizeReason(plan.statusReason),
tone: plan.status === "failed" ? "destructive" : plan.status === "expired" ? "warning" : "neutral",
});
}
ev.push({
key: "dl-submit",
stage: "download",
at: task.createdAt,
title: task.origin === "adopted" ? "开始管理已有下载" : "提交下载",
detail: task.name,
tone: "neutral",
});
if (task.pauseReason)
ev.push({
key: "dl-pause",
stage: "download",
at: task.updatedAt,
title: task.pauseReason === "threshold" ? "下载盘超过暂停线,已暂停" : PAUSE_REASON_LABEL[task.pauseReason],
tone: "warning",
ongoing: task.status === "paused",
});
if (task.noProgressSince)
ev.push({
key: "dl-stall",
stage: "download",
at: task.noProgressSince,
title: "无进展,不自动换源",
tone: "warning",
ongoing: task.status === "stalled",
});
if (task.completedAt)
ev.push({ key: "dl-done", stage: "download", at: task.completedAt, title: "下载完成", tone: "success" });
else if (task.status === "downloading")
ev.push({
key: "dl-progress",
stage: "download",
at: task.lastProgressAt ?? task.updatedAt,
title: `下载中 ${(task.progress * 100).toFixed(1)}%`,
tone: "neutral",
ongoing: true,
});
if (task.lastError)
ev.push({
key: "dl-err",
stage: "download",
at: task.updatedAt,
title: "下载器报错",
detail: humanizeReason(String((task.lastError as { message?: unknown }).message ?? "")) || null,
tone: "destructive",
});
for (const r of records) {
const c = r.counts;
ev.push({
key: `rec-${r.id}`,
stage: "organization",
at: r.startedAt ?? r.createdAt,
title: `整理 ${organizationActionLabel(r.action)}:${organizationLabel(r.status).label}`,
detail:
[
c.total ? `${c.succeeded}/${c.total} 个文件成功` : null,
c.failed ? `${c.failed} 个失败` : null,
c.skipped ? `${c.skipped} 个跳过` : null,
r.reason ? reasonLine(r.reason) : null,
]
.filter(Boolean)
.join(" · ") || null,
tone: RECORD_TONE[r.status],
href: recordPath("organize", r.id),
ongoing: r.status === "running" || r.status === "queued",
});
}
for (const d of decisions) {
const state = DECISION_STATUS[d.status];
const needsUser = d.status === "open" || d.status === "awaiting_user";
const recovery = d.evidence.autoRecover as { suggestion?: unknown } | undefined;
const suggestion =
d.status === "awaiting_user" && typeof recovery?.suggestion === "string" ? recovery.suggestion : null;
ev.push({
key: `dec-${d.id}`,
stage: "agent",
at: d.createdAt,
title: `${state.label}:${d.title}`,
detail: suggestion ?? d.summary,
tone: needsUser && d.severity === "critical" ? "destructive" : state.tone,
href: subjectLink(d) ?? (d.agentRunId ? consolePath("agent", d.agentRunId) : consolePath("agent")),
ongoing: d.status === "investigating",
});
if (d.agentRunId)
ev.push({
key: `agent-${d.id}`,
stage: "agent",
at: d.updatedAt,
title: "智能体调查记录",
tone: "neutral",
href: consolePath("agent", d.agentRunId),
});
}
for (const t of tasks) {
const err = t.lastError
? String((t.lastError as { message?: unknown }).message ?? JSON.stringify(t.lastError))
: null;
ev.push({
key: `task-${t.id}`,
stage: "engine",
at: t.startedAt ?? t.createdAt,
title: `任务 ${TASK_KIND_LABEL[t.kind] ?? t.kind}:${TASK_STATUS_TEXT[t.status] ?? t.status}`,
detail:
[t.attempts > 1 ? `第 ${t.attempts} 次尝试` : null, humanizeReason(t.waitReason), humanizeReason(err)]
.filter(Boolean)
.join(" · ") || null,
tone: TASK_STATUS_TONE[t.status],
href: taskPath(t.id),
ongoing: ACTIVE_TASK.includes(t.status),
});
}
return ev.sort((a, b) => a.at.localeCompare(b.at));
}
/** 一站式取 trace 数据 */
export function useDownloadTrace(task: DownloadTask | undefined, plan: AcquisitionPlan | null | undefined) {
const release = useSourceRelease(task?.sourceReleaseId ?? plan?.sourceReleaseId);
const records = useRecordsForDownload(task?.id);
const decisions = useSubjectDecisions([task?.id ?? "", task?.acquisitionPlanId ?? ""]);
const tasks = useTraceTasks({ planId: task?.acquisitionPlanId, downloadTaskId: task?.id });
return {
release,
records,
decisions,
tasks,
events: task
? buildTimeline({
task,
release: release.data,
plan,
records: records.data?.items ?? [],
decisions: decisions.data,
tasks: tasks.data,
})
: [],
loading: release.isLoading || records.isLoading || decisions.isLoading || tasks.isLoading,
};
}