Merge branch 'dl3'

Co-Authored-By: Claude Fable 5.1 <[email protected]>
This commit is contained in:
rootandClaude Fable 5.1 committed 2026-09-30 01:15:40 +08:00
commit 85aa9d47d0
50 files changed
+1535 -348

No files matched your search

@@ -0,0 +1,314 @@
import { sql } from "@umt/db";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { createHarness, type Harness, pgAvailable, seedEntity, seedRelease, seedSlot } from "./harness.ts";
// DL-3 下载计划并入下载面板:等批准 / 等磁盘空间的计划作为「进行中」行与下载任务按时间一起排、计入计数;
// 深链按计划 id 取行;「为什么下这个」;取消还没提交的下载;没提交就结束的进作品下载历史;
// 审批被拒 / 过期时草案跟着结束。
const hasPg = await pgAvailable();
const h40 = (n: number) => n.toString(16).padStart(40, "c");
type Row = { kind: "task" | "plan"; id: string; name: string; wait?: string; waitReason?: string | null };
describe.skipIf(!hasPg)("DL-3 下载计划并入下载面板(真实 PG)", () => {
let h: Harness;
let entity: string;
let siteId: string;
const ids = {} as Record<"reserved" | "draft" | "orphan" | "submitted" | "failed" | "agentDraft", string>;
const slots = {} as Record<string, string>;
let taskId: string;
let instId: string;
const at = (min: number) => new Date(Date.UTC(2026, 8, 30, 8, min));
async function plan(
key: keyof typeof ids,
p: { status: string; min: number; purpose?: string; evaluation?: Record<string, unknown>; reason?: string },
) {
const slot = await seedSlot(h.db, { entityId: entity, name: key === "submitted" ? "4K" : `slot-${key}` });
slots[key] = slot;
const rel = await seedRelease(h.db, {
siteId,
entityId: entity,
title: `Severance.S02.${key}.2160p`,
size: 10_000_000_000,
hash: h40(Object.keys(ids).length + 1),
});
await h.db
.updateTable("source_release")
.set({ promotion: JSON.stringify({ kind: "free" }), seeders: 41 })
.where("id", "=", rel)
.execute();
const row = await h.db
.insertInto("acquisition_plan")
.values({
slot_id: slot,
source_release_id: rel,
purpose: (p.purpose ?? "fill") as "fill",
status: p.status as "draft",
evaluation: JSON.stringify(p.evaluation ?? { reasons: ["hard_conditions_met", "pri:90"], instanceId: instId }),
reserved_bytes: 10_000_000_000,
payload_hash: `hash-${key}`,
idempotency_key: `test:${key}`,
status_reason: p.reason ?? null,
created_at: at(p.min),
})
.returning("id")
.executeTakeFirstOrThrow();
ids[key] = row.id;
if (p.status === "reserved")
await h.db
.updateTable("acquisition_slot")
.set({ status: "reserved", reserved_plan_id: row.id })
.where("id", "=", slot)
.execute();
return row.id;
}
async function approval(planId: string, status = "pending", scope: Record<string, unknown> = {}) {
return (
await h.db
.insertInto("approval")
.values({
plan_type: "acquisition_plan",
plan_id: planId,
payload_hash: "hash-x",
expected_revision: 1,
scope: JSON.stringify(scope),
status: status as "pending",
requested_by: "agent:test",
expires_at: new Date(Date.now() + 3600_000),
})
.returning("id")
.executeTakeFirstOrThrow()
).id;
}
beforeAll(async () => {
h = await createHarness("dl3plans");
entity = await seedEntity(h.db, { kind: "series", title: "人生切割术", year: 2022 });
siteId = (
await h.db
.insertInto("site")
.values({ key: "dl3site", name: "测试站", adapter: "nexusphp", base_url: "https://pt.example.test" })
.returning("id")
.executeTakeFirstOrThrow()
).id;
// 下载器停用:面板不去连 qB,只看数据库
instId = (
await h.db
.insertInto("downloader_instance")
.values({ name: "qb", base_url: "http://127.0.0.1:9", username: "a", enabled: false })
.returning("id")
.executeTakeFirstOrThrow()
).id;
// 时间线(新 → 旧):已预留等空间(50)> 下载中的任务(40)> 等批准(30)> 智能体草案(25)> 孤立草案(20)> 失败(10)
await plan("failed", { status: "failed", min: 10, reason: "rejected: torrent is invalid" });
await plan("orphan", { status: "draft", min: 20 });
await plan("agentDraft", { status: "draft", min: 25 });
await plan("draft", { status: "draft", min: 30 });
await plan("submitted", { status: "submitted", min: 35 });
await plan("reserved", { status: "reserved", min: 50 });
await approval(ids.draft);
await approval(ids.agentDraft, "pending", { agentRunId: "x" });
taskId = (
await h.db
.insertInto("download_task")
.values({
instance_id: instId,
torrent_hash: h40(99),
acquisition_plan_id: ids.submitted,
name: "Severance.S02.submitted.2160p",
status: "downloading",
managed: true,
media_entity_id: entity,
created_at: at(40),
})
.returning("id")
.executeTakeFirstOrThrow()
).id;
// 已预留的提交任务被下载盘阈值延后
await h.db
.insertInto("task")
.values({
kind: "submitDownload",
idempotency_key: `plan:${ids.reserved}`,
status: "waiting",
subject_type: "acquisition_plan",
subject_id: ids.reserved,
output: JSON.stringify({ status: "deferred", reason: "cache:projected_over_threshold" }),
})
.execute();
// 智能体会话准备的草案
const run = await h.db
.insertInto("agent_run")
.values({ trigger: "user", goal: "补齐人生切割术第二季的 4K 版本" })
.returning("id")
.executeTakeFirstOrThrow();
await h.db
.insertInto("capability_invocation")
.values({
idempotency_key: "inv-1",
capability_id: "download.plan",
capability_version: 1,
agent_run_id: run.id,
actor_id: "agent",
source: "agent",
status: "succeeded",
input: JSON.stringify({}),
output: JSON.stringify({ plan: { id: ids.agentDraft } }),
})
.execute();
}, 60_000);
afterAll(async () => {
await h?.close();
});
it("进行中:等批准、等磁盘空间的计划与下载任务按加入时间一起排,计数包含它们;孤立草案不出现", async () => {
const r = await h.api("GET", "/download-tasks/board");
expect(r.status).toBe(200);
const b = r.body.data;
const rows = b.items as Row[];
expect(rows.map((x) => [x.kind, x.id])).toEqual([
["plan", ids.reserved],
["task", taskId],
["plan", ids.draft],
["plan", ids.agentDraft],
]);
expect(b.counts).toEqual({ active: 4, completed: 0 });
const reserved = b.items[0];
expect(reserved).toMatchObject({
wait: "space",
waitReason: "cache:projected_over_threshold",
status: "reserved",
purpose: "fill",
name: "Severance.S02.reserved.2160p",
sizeBytes: 10_000_000_000,
media: { id: entity, title: "人生切割术", year: 2022 },
approval: null,
});
const draft = b.items[2];
expect(draft).toMatchObject({ wait: "approval", status: "draft", approval: { payloadHash: "hash-x" } });
// 已完成分组不含计划
const done = (await h.api("GET", "/download-tasks/board?state=completed")).body.data;
expect(done.items).toEqual([]);
});
it("翻页:两路按同一游标归并,不重不漏", async () => {
const p1 = (await h.api("GET", "/download-tasks/board?limit=3")).body.data;
expect(p1.items.map((x: Row) => x.id)).toEqual([ids.reserved, taskId, ids.draft]);
expect(p1.nextCursor).toBeTruthy();
const p2 = (await h.api("GET", `/download-tasks/board?limit=3&cursor=${encodeURIComponent(p1.nextCursor)}`)).body
.data;
expect(p2.items.map((x: Row) => x.id)).toEqual([ids.agentDraft]);
expect(p2.nextCursor).toBeNull();
});
it("深链 planId:等待中的计划给计划行,已提交的给它的下载任务,结束的没有行", async () => {
const a = (await h.api("GET", `/download-tasks/board?planId=${ids.reserved}`)).body.data;
expect(a.items.map((x: Row) => [x.kind, x.id])).toEqual([["plan", ids.reserved]]);
const b = (await h.api("GET", `/download-tasks/board?planId=${ids.submitted}&state=completed`)).body.data;
expect(b.items.map((x: Row) => [x.kind, x.id])).toEqual([["task", taskId]]);
const c = (await h.api("GET", `/download-tasks/board?planId=${ids.failed}`)).body.data;
expect(c.items).toEqual([]);
});
it("为什么下这个:订阅 / 智能体会话来源、目的、选中理由原始要点", async () => {
const sub = (await h.api("GET", `/acquisition-plans/${ids.submitted}/why`)).body.data;
expect(sub).toMatchObject({
planId: ids.submitted,
source: { kind: "subscription", name: "人生切割术" },
purpose: "fill",
slotName: "4K",
replaces: null,
reasons: ["hard_conditions_met", "pri:90"],
promotion: "free",
seeders: 41,
siteName: "测试站",
});
expect(sub.evaluation.reasons).toEqual(["hard_conditions_met", "pri:90"]);
const agent = (await h.api("GET", `/acquisition-plans/${ids.agentDraft}/why`)).body.data;
expect(agent.source).toMatchObject({ kind: "agent", name: "补齐人生切割术第二季的 4K 版本" });
expect((await h.api("GET", `/acquisition-plans/${crypto.randomUUID()}/why`)).status).toBe(404);
});
it("升级:被替换的版本取评估时记下的文件规格", async () => {
const asset = await h.db
.insertInto("media_asset")
.values({
media_entity_id: entity,
kind: "video",
delivery: JSON.stringify({ resolution: "1080p", source: "BluRay", hdr: [], releaseGroup: "CMCT" }),
})
.returning("id")
.executeTakeFirstOrThrow();
await h.db
.updateTable("acquisition_plan")
.set({
purpose: "upgrade",
evaluation: JSON.stringify({ reasons: ["source_tier_improved"], replacesAssetId: asset.id }),
})
.where("id", "=", ids.submitted)
.execute();
const w = (await h.api("GET", `/acquisition-plans/${ids.submitted}/why`)).body.data;
expect(w.purpose).toBe("upgrade");
expect(w.replaces).toEqual({ assetId: asset.id, label: "1080p 蓝光压制 · CMCT" });
});
it("取消:已预留的释放收录版本并进下载历史;已提交的 409", async () => {
const r = await h.api("POST", `/acquisition-plans/${ids.reserved}/cancel`);
expect(r.status).toBe(200);
expect(r.body.data).toMatchObject({ status: "cancelled", statusReason: "cancelled_by_user" });
const slot = await h.db
.selectFrom("acquisition_slot")
.select(["status", "reserved_plan_id"])
.where("id", "=", slots.reserved!)
.executeTakeFirstOrThrow();
expect(slot.status).not.toBe("reserved");
expect((await h.api("POST", `/acquisition-plans/${ids.submitted}/cancel`)).status).toBe(409);
expect((await h.api("POST", `/acquisition-plans/${ids.reserved}/cancel`)).status).toBe(409);
const b = (await h.api("GET", "/download-tasks/board")).body.data;
expect(b.counts.active).toBe(3);
expect(b.items.some((x: Row) => x.id === ids.reserved)).toBe(false);
});
it("作品下载历史:没提交就结束的下载带原因码;孤立草案与进行中的不在里面", async () => {
const d = (await h.api("GET", `/library/${entity}`)).body.data;
const dropped = d.droppedPlans.map((x: { plan: { id: string; statusReason: string } }) => [
x.plan.id,
x.plan.statusReason,
]);
expect(dropped).toEqual(
expect.arrayContaining([
[ids.failed, "rejected: torrent is invalid"],
[ids.reserved, "cancelled_by_user"],
]),
);
expect(dropped).toHaveLength(2);
expect(d.droppedPlans.find((x: { plan: { id: string } }) => x.plan.id === ids.failed).releaseTitle).toBe(
"Severance.S02.failed.2160p",
);
});
it("审批被拒 / 过期:草案跟着结束,不再算进行中", async () => {
const { closeDraftPlan } = await import("../../agent/interaction.ts");
await closeDraftPlan(h.db, { plan_type: "acquisition_plan", plan_id: ids.draft }, "rejected");
await closeDraftPlan(h.db, { plan_type: "acquisition_plan", plan_id: ids.agentDraft }, "expired");
// 不是草案的不动
await closeDraftPlan(h.db, { plan_type: "acquisition_plan", plan_id: ids.submitted }, "rejected");
const rows = await sql<{ id: string; status: string; status_reason: string | null }>`
select id, status, status_reason from acquisition_plan where id in (${sql.join([ids.draft, ids.agentDraft, ids.submitted])})`.execute(
h.db,
);
const by = new Map(rows.rows.map((r) => [r.id, [r.status, r.status_reason]]));
expect(by.get(ids.draft)).toEqual(["cancelled", "approval_rejected"]);
expect(by.get(ids.agentDraft)).toEqual(["expired", "approval_expired"]);
expect(by.get(ids.submitted)?.[0]).toBe("submitted");
const b = (await h.api("GET", "/download-tasks/board")).body.data;
expect(b.items.map((x: Row) => x.kind)).toEqual(["task"]);
expect(b.counts.active).toBe(1);
});
});
@@ -363,7 +363,7 @@ export function registerAoCapabilities(registry: CapabilityRegistry, deps: () =>
const d = deps();
const db = d.ctx.db;
const plan = await db.selectFrom("acquisition_plan").selectAll().where("id", "=", input.planId).executeTakeFirst();
if (!plan) throw new CapabilityError("not_found", "下载计划不存在");
if (!plan) throw new CapabilityError("not_found", "这个 planId 对应的下载不存在");
if (plan.status === "draft") {
// 草案 → reserved:Slot 必须与生成计划时一致(revision 未变、没有其它预留)
const ok = await db
@@ -4,11 +4,12 @@ import { ApiException } from "../../../http/errors.ts";
import type { ModuleApp } from "../../types.ts";
import { type AoDeps, enqueueTracked, timeSlot } from "../services/common.ts";
import { downloadBoard } from "../services/download-board.ts";
import { userPause, userResume } from "../services/downloads.ts";
import { actorOf, notFound, toApi, toApiList } from "../util.ts";
import { applySlotEvent, userPause, userResume } from "../services/downloads.ts";
import { planWhys } from "../services/plan-why.ts";
import { actorOf, conflict, notFound, toApi, toApiList } from "../util.ts";
import { mount, paginate } from "./helpers.ts";
// 下载计划 / 下载任务(含用户暂停与恢复)/ 手动触发缓存清理与资产对账。
// 计划(界面统称"下载")/ 下载任务(含用户暂停与恢复)/ 手动触发缓存清理与资产对账。
const TAG = "downloads";
@@ -32,7 +33,62 @@ export function registerDownloadRoutes(app: ModuleApp, d: AoDeps) {
Api.acquisitionPlans.get,
async (_c, { params }) => {
const r = await db().selectFrom("acquisition_plan").selectAll().where("id", "=", params.id).executeTakeFirst();
return toApi<AcquisitionPlan>(r ?? notFound("下载计划"));
return toApi<AcquisitionPlan>(r ?? notFound("下载"));
},
{ tag: TAG },
);
mount(
app,
Api.acquisitionPlans.why,
async (_c, { params }) => (await planWhys(db(), [params.id])).get(params.id) ?? notFound("下载"),
{ tag: TAG },
);
// DL-3:取消还没提交的计划(已预留 / 草案);已有下载任务的不在这里取消(到下载器里删种)
mount(
app,
Api.acquisitionPlans.cancel,
async (c, { params }) => {
const row = await db()
.transaction()
.execute(async (trx) => {
const p = await trx
.selectFrom("acquisition_plan")
.selectAll()
.where("id", "=", params.id)
.forUpdate()
.executeTakeFirst();
if (!p) return notFound("下载");
if (p.status !== "draft" && p.status !== "reserved") conflict("这个下载已经结束或已提交,不能取消");
const dt = await trx
.selectFrom("download_task")
.select("id")
.where("acquisition_plan_id", "=", p.id)
.executeTakeFirst();
if (dt) conflict("已经提交到下载器,到下载任务里处理");
const out = await trx
.updateTable("acquisition_plan")
.set({ status: "cancelled", status_reason: "cancelled_by_user" })
.where("id", "=", p.id)
.returningAll()
.executeTakeFirstOrThrow();
// 草案没有占用收录版本;已预留的释放预留,收录版本到期重评
if (p.status === "reserved")
await applySlotEvent(d, trx, p.slot_id, { type: "release" }, { reason: "cancelled_by_user", planId: p.id });
await trx
.insertInto("audit_event")
.values({
actor_type: actorOf(c).startsWith("api_key:") ? "api_key" : "user",
actor_id: actorOf(c),
action: "acquisition_plan.cancel",
target_type: "acquisition_plan",
target_id: p.id,
payload: JSON.stringify({ from: p.status }),
})
.execute();
return out;
});
return toApi<AcquisitionPlan>(row);
},
{ tag: TAG },
);
@@ -336,6 +336,26 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) {
.orderBy("created_at")
.execute()
: [];
// DL-3:没提交到下载器就结束的计划(失败 / 取消 / 过期、没有下载任务)也进下载历史
const dropped = slotIds.length
? await db()
.selectFrom("acquisition_plan as p")
.leftJoin("source_release as sr", "sr.id", "p.source_release_id")
.selectAll("p")
.select("sr.title as release_title")
.where("p.slot_id", "in", slotIds)
.where("p.status", "in", ["failed", "cancelled", "expired"])
.where((eb) =>
eb.not(
eb.exists(
eb.selectFrom("download_task as x").select("x.id").whereRef("x.acquisition_plan_id", "=", "p.id"),
),
),
)
.orderBy("p.updated_at", "desc")
.limit(50)
.execute()
: [];
const rootPath = new Map(roots.map((r) => [r.id, r.local_path]));
const detail: LibraryDetail = {
entity: toApi<MediaEntity>(e!),
@@ -367,6 +387,10 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) {
lastOrganizationStatus: mine[mine.length - 1]?.status ?? null,
};
}),
droppedPlans: dropped.map(({ release_title, ...p }) => ({
plan: toApi<AcquisitionPlan>(p),
releaseTitle: release_title,
})),
organizationRecords: entityRecs.map((r) => ({
id: r.id,
status: r.status,
@@ -1000,16 +1000,18 @@ async function reserve(
mode: p.mode,
unitKeys: p.unitKeys,
};
// 升级 = 替换已有资产;电影 Slot 缺口(无资产)的 mode 虽记 upgrade,用途仍是补缺
const purpose = p.mode === "upgrade" && p.slot.current_asset_id ? ("upgrade" as const) : ("fill" as const);
const evaluation = jsonb({
...toContractSlotEvaluation(p.ev),
mode: p.mode,
instanceId: p.instanceId,
volumeKey: p.volumeKey,
savePath: p.savePath,
// DL-3「为什么下这个」:升级时记下被替换的文件(之后 Slot 的当前文件会变)
...(purpose === "upgrade" ? { replacesAssetId: p.slot.current_asset_id } : {}),
...(p.evaluationExtra ?? {}),
});
// 升级 = 替换已有资产;电影 Slot 缺口(无资产)的 mode 虽记 upgrade,用途仍是补缺
const purpose = p.mode === "upgrade" && p.slot.current_asset_id ? ("upgrade" as const) : ("fill" as const);
if (c.draft) {
// 草案:同一 Slot+发布复用未提交的草案;Slot 状态与预算不动,expected_revision 记下当前 Slot 版本 +1
const key = `draft:${p.slot.id}:${p.rel.id}`;
@@ -4,15 +4,18 @@ import {
DOWNLOAD_BOARD_STATUSES,
type DownloadBoard,
type DownloadBoardItem,
type DownloadBoardPlanItem,
type DownloadBoardQuery,
type DownloadBoardRow,
type DownloadLive,
type DownloadTask,
type OrganizationStatus,
type PlanWait,
} from "@umt/contracts";
import { type DownloaderInstanceRow, sql } from "@umt/db";
import { paginate } from "../routes/helpers.ts";
import { errMessage, toApi } from "../util.ts";
import type { AoDeps } from "./common.ts";
import { encodeCursor, errMessage, toApi } from "../util.ts";
import type { AoDeps, Exec } from "./common.ts";
// 下载面板(DL-2):数据库里的下载任务 + qB 实时数据。
// 实时数据按实例缓存 liveTtlMs(缺省 2 s)并合并并发请求:多个标签页 / 多人每 3–5 s 轮询,
@@ -119,28 +122,160 @@ export function instanceLive(d: AoDeps, inst: DownloaderInstanceRow): Promise<In
return entry.p;
}
/**
* 还没提交到下载器、在等条件的计划(DL-3):草案且有待批准的审批(等批准);已预留但还没有下载任务
* (等下载盘空间 / 自动化开启 / 提交任务执行)。没有审批的草案(智能体只生成没提交)不算在等。
*/
export function waitingPlansQuery(db: Exec, instanceId?: string) {
return db
.selectFrom("acquisition_plan as p")
.selectAll("p")
.where((eb) =>
eb.or([
eb.and([
eb("p.status", "=", "reserved"),
eb.not(
eb.exists(
eb.selectFrom("download_task as dt").select("dt.id").whereRef("dt.acquisition_plan_id", "=", "p.id"),
),
),
]),
eb.and([
eb("p.status", "=", "draft"),
eb.exists(
eb
.selectFrom("approval as a")
.select("a.id")
.where("a.plan_type", "=", "acquisition_plan")
.whereRef("a.plan_id", "=", sql.ref("p.id"))
.where("a.status", "=", "pending")
.where((x) => x.or([x("a.expires_at", "is", null), x("a.expires_at", ">", sql<Date>`now()`)])),
),
]),
]),
)
.$if(!!instanceId, (qb) => qb.where(sql<string>`p.evaluation->>'instanceId'`, "=", instanceId!));
}
/** 提交任务的延后原因 → 在等什么 */
export function planWaitOf(status: string, deferred: string | null): PlanWait {
if (status === "draft") return "approval";
if (!deferred) return "queued";
if (deferred.startsWith("cache:") || deferred === "volume_state_unknown") return "space";
if (deferred === "automation_disabled" || deferred === "read_only_reconcile") return "automation_off";
return "queued";
}
/** 等待中的计划 → 面板行 */
async function planItems(db: Exec, rows: { id: string; slot_id: string; source_release_id: string }[]) {
if (!rows.length)
return new Map<string, Omit<DownloadBoardPlanItem, "kind" | "id" | "createdAt" | "purpose" | "status">>();
const ids = rows.map((r) => r.id);
const [extra, approvals, tasks] = await Promise.all([
db
.selectFrom("acquisition_plan as p")
.innerJoin("acquisition_slot as s", "s.id", "p.slot_id")
.leftJoin("media_entity as e", "e.id", "s.media_entity_id")
.leftJoin("source_release as r", "r.id", "p.source_release_id")
.select([
"p.id",
"p.reserved_bytes",
"r.title as release_title",
"r.size_bytes",
"e.id as entity_id",
"e.title",
"e.year",
])
.where("p.id", "in", ids)
.execute(),
db
.selectFrom("approval")
.select(["id", "plan_id", "payload_hash", "decision_id"])
.where("plan_type", "=", "acquisition_plan")
.where("plan_id", "in", ids)
.where("status", "=", "pending")
.orderBy("created_at", "desc")
.execute(),
// 提交任务最近一次的结果:延后(deferred)时 output.reason 是原因码
sql<{ subject_id: string; status: string; reason: string | null }>`
select distinct on (subject_id) subject_id, status,
case when output->>'status' = 'deferred' then output->>'reason' end as reason
from task
where kind = 'submitDownload' and subject_type = 'acquisition_plan' and subject_id in (${sql.join(ids)})
order by subject_id, updated_at desc`
.execute(db)
.then((r) => r.rows),
]);
const out = new Map<string, Omit<DownloadBoardPlanItem, "kind" | "id" | "createdAt" | "purpose" | "status">>();
for (const r of rows) {
const x = extra.find((e) => e.id === r.id);
const a = approvals.find((ap) => ap.plan_id === r.id);
const t = tasks.find((tk) => tk.subject_id === r.id);
out.set(r.id, {
name: x?.release_title ?? "未知资源",
sizeBytes: x?.size_bytes ?? (x?.reserved_bytes ? x.reserved_bytes : null),
media: x?.entity_id && x.title ? { id: x.entity_id, title: x.title, year: x.year ?? null } : null,
wait: "queued",
waitReason: t?.reason ?? null,
approval: a ? { id: a.id, payloadHash: a.payload_hash, decisionId: a.decision_id } : null,
});
}
return out;
}
export async function downloadBoard(d: AoDeps, query: DownloadBoardQuery): Promise<DownloadBoard> {
const db = d.ctx.db;
let q = db
.selectFrom("download_task")
.selectAll()
.where("removed_at", "is", null)
.where("status", "in", [...DOWNLOAD_BOARD_STATUSES[query.state]]);
.where("status", "in", [
...(query.planId
? [...DOWNLOAD_BOARD_STATUSES.active, ...DOWNLOAD_BOARD_STATUSES.completed]
: DOWNLOAD_BOARD_STATUSES[query.state]),
]);
if (query.instanceId) q = q.where("instance_id", "=", query.instanceId);
if (query.planId) q = q.where("acquisition_plan_id", "=", query.planId);
let insts = db.selectFrom("downloader_instance").selectAll().where("enabled", "=", true);
if (query.instanceId) insts = insts.where("id", "=", query.instanceId);
const [page, counts, instances] = await Promise.all([
const withPlans = query.state === "active" || !!query.planId;
const planQ = query.planId
? waitingPlansQuery(db, query.instanceId).where("p.id", "=", query.planId)
: waitingPlansQuery(db, query.instanceId);
const [page, planPage, counts, waitingCount, instances] = await Promise.all([
paginate(q, query),
withPlans ? paginate(planQ, query, "p.") : Promise.resolve({ rows: [], nextCursor: null }),
sql<{ active: string; completed: string }>`
select count(*) filter (where status in (${sql.join([...DOWNLOAD_BOARD_STATUSES.active])})) as active,
count(*) filter (where status in (${sql.join([...DOWNLOAD_BOARD_STATUSES.completed])})) as completed
from download_task
where removed_at is null ${query.instanceId ? sql`and instance_id = ${query.instanceId}` : sql``}`.execute(db),
waitingPlansQuery(db, query.instanceId)
.clearSelect()
.select((eb) => eb.fn.countAll<string>().as("n"))
.executeTakeFirst(),
insts.orderBy("name").execute(),
]);
const rows = page.rows;
// 两路按 (created_at, id) 倒序归并,取前 limit 条;游标取最后一条,两路下一页用同一游标
type Merged =
| { kind: "task"; row: (typeof page.rows)[number] }
| { kind: "plan"; row: (typeof planPage.rows)[number] };
const merged: Merged[] = [
...page.rows.map((row) => ({ kind: "task" as const, row })),
...planPage.rows.map((row) => ({ kind: "plan" as const, row })),
].sort((a, b) => {
const t = b.row.created_at.getTime() - a.row.created_at.getTime();
return t !== 0 ? t : b.row.id < a.row.id ? -1 : b.row.id > a.row.id ? 1 : 0;
});
const more = merged.length > query.limit || page.nextCursor !== null || planPage.nextCursor !== null;
const taken = merged.slice(0, query.limit);
const lastTaken = taken[taken.length - 1];
const nextCursor = more && lastTaken ? encodeCursor(lastTaken.row.created_at, lastTaken.row.id) : null;
const rows = taken.flatMap((m) => (m.kind === "task" ? [m.row] : []));
const planRows = taken.flatMap((m) => (m.kind === "plan" ? [m.row] : []));
const entityIds = [...new Set(rows.map((r) => r.media_entity_id).filter((x): x is string => !!x))];
const taskIds = rows.map((r) => r.id);
@@ -182,11 +317,12 @@ export async function downloadBoard(d: AoDeps, query: DownloadBoardQuery): Promi
const recordByTask = new Map(records.map((r) => [r.download_task_id, r]));
const liveByInst = new Map(lives.map((l) => [l.instanceId, l]));
const items: DownloadBoardItem[] = rows.map((r) => {
const taskItems: DownloadBoardItem[] = rows.map((r) => {
const e = r.media_entity_id ? entityById.get(r.media_entity_id) : undefined;
const rec = recordByTask.get(r.id);
return {
...toApi<DownloadTask>(r),
kind: "task" as const,
live: liveByInst.get(r.instance_id)?.torrents.get(r.torrent_hash) ?? null,
media: e ? { id: e.id, title: e.title, year: e.year } : null,
organize: rec
@@ -199,13 +335,32 @@ export async function downloadBoard(d: AoDeps, query: DownloadBoardQuery): Promi
: null,
};
});
const planExtra = await planItems(db, planRows);
const taskById = new Map(taskItems.map((t) => [t.id, t]));
const items: DownloadBoardRow[] = taken.map((m): DownloadBoardRow => {
if (m.kind === "task") return taskById.get(m.row.id)!;
const p = m.row;
const x = planExtra.get(p.id)!;
return {
kind: "plan",
id: p.id,
createdAt: p.created_at.toISOString(),
purpose: p.purpose,
status: p.status,
...x,
wait: planWaitOf(p.status, x.waitReason),
};
});
const c = counts.rows[0];
const reachable = lives.filter((l) => l.reachable);
const oldest = reachable.length ? Math.min(...reachable.map((l) => l.at.getTime())) : null;
return {
items,
nextCursor: page.nextCursor,
counts: { active: Number(c?.active ?? 0), completed: Number(c?.completed ?? 0) },
nextCursor,
counts: {
active: Number(c?.active ?? 0) + Number(waitingCount?.n ?? 0),
completed: Number(c?.completed ?? 0),
},
transfer: {
dlSpeed: reachable.reduce((a, l) => a + l.dlSpeed, 0),
upSpeed: reachable.reduce((a, l) => a + l.upSpeed, 0),
@@ -0,0 +1,131 @@
import type { DownloadWhy } from "@umt/contracts";
import { sql } from "@umt/db";
import type { Exec } from "./common.ts";
// DL-3「为什么下这个」:计划的来源(订阅 / 智能体会话 / 手动)、目的(补缺 / 升级 / 换源)、被替换的版本、
// 选中理由的原始要点(evaluation.reasons + 资源促销 / 做种数)。文案由前端按这些事实拼,原始评估进技术信息。
type Delivery = {
resolution?: string | null;
source?: string | null;
hdr?: string[] | null;
releaseGroup?: string | null;
};
const SOURCE_NAME: Record<string, string> = { REMUX: "Remux", BluRay: "蓝光压制", "BluRay-Disc": "原盘" };
/** 现有版本的规格短句:"1080p 蓝光压制 HDR10 · CMCT";认不出规格时退回资源标题 */
export function assetLabel(delivery: unknown, fallback: string | null): string {
const d = (delivery ?? {}) as Delivery;
const spec = [d.resolution, d.source ? (SOURCE_NAME[d.source] ?? d.source) : null, ...(d.hdr ?? [])]
.filter((x): x is string => !!x)
.join(" ");
const label = [spec, d.releaseGroup].filter(Boolean).join(" · ");
return label || fallback || "现有版本";
}
export async function planWhys(db: Exec, planIds: readonly string[]): Promise<Map<string, DownloadWhy>> {
const out = new Map<string, DownloadWhy>();
const ids = [...new Set(planIds)];
if (!ids.length) return out;
const plans = await db
.selectFrom("acquisition_plan as p")
.innerJoin("acquisition_slot as s", "s.id", "p.slot_id")
.leftJoin("source_release as r", "r.id", "p.source_release_id")
.leftJoin("site", "site.id", "r.site_id")
.select([
"p.id",
"p.purpose",
"p.status",
"p.evaluation",
"s.name as slot_name",
"s.subscription_id",
"s.current_asset_id",
"r.promotion",
"r.seeders",
"site.name as site_name",
])
.where("p.id", "in", ids)
.execute();
if (!plans.length) return out;
// 智能体 / 手动:download.plan 的输出或 download.submit 的输入里有这个计划
const invocations = await sql<{ plan_id: string; agent_run_id: string | null; goal: string | null }>`
select coalesce(ci.input->>'planId', ci.output->'plan'->>'id') as plan_id, ci.agent_run_id, ar.goal
from capability_invocation ci
left join agent_run ar on ar.id = ci.agent_run_id
where ci.capability_id in ('download.plan', 'download.submit')
and coalesce(ci.input->>'planId', ci.output->'plan'->>'id') in (${sql.join(ids)})`.execute(db);
const subIds = [...new Set(plans.map((p) => p.subscription_id).filter((x): x is string => !!x))];
const subs = subIds.length
? await db
.selectFrom("subscription as sub")
.leftJoin("media_entity as e", "e.id", "sub.media_entity_id")
.leftJoin("content_unit as cu", "cu.id", "sub.season_unit_id")
.select(["sub.id", "sub.title_query", "e.title", "cu.season_number"])
.where("sub.id", "in", subIds)
.execute()
: [];
const evOf = (p: (typeof plans)[number]) => (p.evaluation ?? {}) as Record<string, unknown>;
// 被替换的版本:评估时记下的 replacesAssetId(DL-3 起);更早的计划在未结束前按收录版本当前的文件
const replacedIds = new Map<string, string>();
for (const p of plans) {
if (p.purpose === "fill") continue;
const recorded = evOf(p).replacesAssetId;
const id =
typeof recorded === "string"
? recorded
: ["draft", "reserved", "submitted"].includes(p.status)
? p.current_asset_id
: null;
if (id) replacedIds.set(p.id, id);
}
const assetIds = [...new Set(replacedIds.values())];
const assets = assetIds.length
? await db
.selectFrom("media_asset as a")
.leftJoin("source_release as r", "r.id", "a.source_release_id")
.select(["a.id", "a.delivery", "r.title"])
.where("a.id", "in", assetIds)
.execute()
: [];
for (const p of plans) {
const inv = invocations.rows.filter((r) => r.plan_id === p.id);
const agent = inv.find((r) => r.agent_run_id);
const sub = p.subscription_id ? subs.find((s) => s.id === p.subscription_id) : undefined;
let source: DownloadWhy["source"];
if (agent?.agent_run_id) {
const goal = (agent.goal ?? "").trim().replace(/\s+/g, " ");
source = { kind: "agent", runId: agent.agent_run_id, name: goal.length > 40 ? `${goal.slice(0, 40)}…` : goal };
} else if (inv.length) source = { kind: "manual" };
else if (sub) {
const title = sub.title ?? sub.title_query ?? "";
source = {
kind: "subscription",
id: sub.id,
name:
sub.season_number !== null && sub.season_number !== undefined ? `${title} 第 ${sub.season_number} 季` : title,
};
} else source = { kind: "auto" };
const ev = evOf(p);
const aId = replacedIds.get(p.id);
const a = aId ? assets.find((x) => x.id === aId) : undefined;
const promo = (p.promotion ?? {}) as { kind?: unknown };
out.set(p.id, {
planId: p.id,
source,
purpose: p.purpose,
slotName: p.slot_name,
replaces: a ? { assetId: a.id, label: assetLabel(a.delivery, a.title) } : null,
reasons: Array.isArray(ev.reasons) ? ev.reasons.filter((x): x is string => typeof x === "string") : [],
promotion: typeof promo.kind === "string" && promo.kind ? promo.kind : null,
seeders: p.seeders !== null && p.seeders !== undefined && p.seeders >= 0 ? p.seeders : null,
siteName: p.site_name ?? null,
evaluation: ev,
});
}
return out;
}
@@ -70,7 +70,7 @@ export async function taskTimeline(ctx: ServerContext, t: TaskRow) {
out.push({
at: p.updated_at.toISOString(),
type: "plan.status",
summary: `下载计划 ${p.status}${p.status_reason ? `(${p.status_reason})` : ""}`,
summary: `下载准备 ${p.status}${p.status_reason ? `(${p.status_reason})` : ""}`,
data: { planId: t.subject_id, status: p.status },
});
const dts = await db
+37 -3
View File
@@ -648,14 +648,37 @@ export async function inlineApproval(
return { allowed: false, reason: "已在步骤处发起审批,等待管理员批准" };
}
/**
* DL-3:下载的审批没通过(拒绝 / 撤销 / 过期)时,还是草案的计划跟着结束,进作品的下载历史;
* 否则草案一直挂着,下载面板与历史里都看不到。
*/
export async function closeDraftPlan(
db: Kysely<DB>,
a: { plan_type?: string | null; plan_id?: string | null },
outcome: "rejected" | "revoked" | "expired",
): Promise<void> {
if (a.plan_type !== "acquisition_plan" || !a.plan_id) return;
await db
.updateTable("acquisition_plan")
.set(
outcome === "expired"
? { status: "expired", status_reason: "approval_expired" }
: { status: "cancelled", status_reason: outcome === "rejected" ? "approval_rejected" : "approval_revoked" },
)
.where("id", "=", a.plan_id)
.where("status", "=", "draft")
.execute();
}
async function expireApproval(ctx: ServerContext, approvalId: string) {
const a = await ctx.db
.updateTable("approval")
.set({ status: "expired" })
.where("id", "=", approvalId)
.where("status", "=", "pending")
.returning("decision_id")
.returning(["decision_id", "plan_type", "plan_id"])
.executeTakeFirst();
if (a) await closeDraftPlan(ctx.db, a, "expired");
if (a?.decision_id) await closeDecision(ctx, a.decision_id, "expired", { approval: "expired" }, "system:timeout");
}
@@ -664,9 +687,18 @@ async function expireApproval(ctx: ServerContext, approvalId: string) {
*/
export async function afterApprovalDecided(
ctx: ServerContext,
a: { id: string; status: string; scope: unknown; decision_id: string | null; reason: string | null },
a: {
id: string;
status: string;
scope: unknown;
decision_id: string | null;
reason: string | null;
plan_type?: string | null;
plan_id?: string | null;
},
by: string,
): Promise<void> {
if (a.status === "rejected" || a.status === "revoked") await closeDraftPlan(ctx.db, a, a.status);
const scope = (a.scope ?? {}) as Partial<ApprovalScope>;
if (!scope.agentRunId) return;
if (a.decision_id) {
@@ -697,12 +729,14 @@ export async function dismissPendingInteractions(ctx: ServerContext, runId: stri
.where((eb) => eb.or([eb("dedupe_key", "like", `${OPTIONS_KEY}:%`), eb("dedupe_key", "like", `${APPROVAL_KEY}:%`)]))
.execute();
for (const d of open) await closeDecision(ctx, d.id, "dismissed", { reason: "会话已停止" }, by);
await ctx.db
const revoked = await ctx.db
.updateTable("approval")
.set({ status: "revoked", decided_by: by, decided_at: new Date(), reason: "会话已停止" })
.where(sql<boolean>`scope->>'agentRunId' = ${runId}`)
.where("status", "=", "pending")
.returning(["plan_type", "plan_id"])
.execute();
for (const a of revoked) await closeDraftPlan(ctx.db, a, "revoked");
// 选定即批准的审批还没用掉:会话停了就撤销(只绑本会话,不留给别处)
await ctx.db
.updateTable("approval")
@@ -206,7 +206,7 @@ export function registerBrowserCapabilities(registry: CapabilityRegistry, svc: (
sourceReleaseId: v.sourceReleaseId,
duplicate: !!v.duplicate,
note: v.duplicate
? "同一会话已捕获过这个文件;先查已有下载计划,不要重复提交"
? "同一会话已捕获过这个文件;先查已有的下载,不要重复提交"
: !v.verification.valid
? "不是有效的种子文件"
: v.sourceReleaseId
+1 -1
View File
@@ -262,7 +262,7 @@ export function registerBrowserRoutes(app: ModuleApp, svc: () => BrowserService)
...body.submit,
sourceReleaseId: v.sourceReleaseId,
});
if (!submission.submitted) note = "提交未生成下载计划,见 rejected";
if (!submission.submitted) note = "没有下载,原因见 rejected";
} catch (e) {
// 与 POST /acquisition-plans 相同的拒绝(未识别、无授权等):产物已暂存,结果照常返回
if (!(e instanceof ApiException) || e.code === "internal") throw e;
+3 -3
View File
@@ -144,7 +144,7 @@ export const MCP_PROMPTS: readonly McpPromptDef[] = [
{
name: "acquire_title",
title: "收录一部作品",
description: "从查作品到生成下载计划;提交下载要有授权。",
description: "从查作品到准备下载;提交下载要有授权。",
arguments: [
{ name: "title", description: "作品标题", required: true },
{ name: "year", description: "年份(可选)", required: false },
@@ -154,8 +154,8 @@ export const MCP_PROMPTS: readonly McpPromptDef[] = [
`目标:收录「${a.title ?? ""}」${a.year ? `(${a.year})` : ""}。`,
"1. catalog__lookup 确认作品,拿不准时列出候选,不要猜;",
"2. release__search 找资源,release__inspect 看细节;",
"3. download__plan 生成下载计划并说明选择理由;",
"4. 有持续授权时再调用 download__submit;返回 unauthorized 就停下,给出计划 ID 等审批。",
"3. download__plan 准备下载(返回 planId)并说明选择理由;",
"4. 有持续授权时再调用 download__submit;返回 unauthorized 就停下,给出 planId 等审批。",
].join("\n"),
},
{
@@ -29,7 +29,6 @@ export const SEARCH_ACTIONS: readonly SearchAction[] = [
page("subscriptions", "获取", "订阅", "/console/subscriptions", ["subscription", "追剧"]),
page("rules", "获取", "规则", "/console/rules", ["rules", "收录策略", "policy"]),
page("downloads", "整理", "下载面板", "/console/downloads", ["download", "qb", "下载任务"]),
page("download-plans", "整理", "下载计划", "/console/downloads/plans", ["plan", "计划"]),
page("organize", "整理", "整理记录", "/console/organize", ["organize", "入库", "历史"]),
page("files", "整理", "文件管理", "/console/files", ["files", "文件"]),
page("trash", "整理", "回收站", "/console/files/trash", ["trash", "删除"]),