feat(library): download history lists adopted tasks and organize records
- Migration 0053: download_task.media_entity_id (FK, set null on delete),
backfilled from plan slot / source release / latest organize record.
- Submitted and adopted tasks carry the entity; entity merge follows it.
- GET /library/{id}: downloads also matched by dt.media_entity_id (adopted
takeover tasks), plus organizationRecords (incl. manual path organizes);
library list / inventory counts use it too.
- Web detail: "接管" marker and a list of records without a download task.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
1 parent
74f1566985
commit
05d1f76b9f
9 files changed
+159
-11
No files matched your search
@@ -314,7 +314,12 @@ describe.skipIf(!hasPg)("acquisition-organize 纵切(真实 PG + mock qB + 临
|
||||
.selectAll()
|
||||
.where("id", "=", sub.downloadTaskId!)
|
||||
.executeTakeFirstOrThrow();
|
||||
expect(row).toMatchObject({ origin: "adopted", managed: true, acquisition_plan_id: ev.plans[0] });
|
||||
expect(row).toMatchObject({
|
||||
origin: "adopted",
|
||||
managed: true,
|
||||
acquisition_plan_id: ev.plans[0],
|
||||
media_entity_id: e,
|
||||
});
|
||||
expect(qb.callsTo("torrents/add").length).toBe(1);
|
||||
}, 30_000);
|
||||
|
||||
@@ -460,6 +465,14 @@ describe.skipIf(!hasPg)("acquisition-organize 纵切(真实 PG + mock qB + 临
|
||||
);
|
||||
expect(det.body.data.candidates.some((c: { decision: string }) => c.decision === "reject")).toBe(true);
|
||||
expect(det.body.data.downloads[0].organizationRecordIds.length).toBeGreaterThan(0);
|
||||
// QB-1:下载任务直接记作品;整理记录按作品列出
|
||||
const dune = det.body.data.downloads.find((x: { downloadTask: { name: string } }) =>
|
||||
x.downloadTask.name.startsWith("Dune.Part.Two"),
|
||||
);
|
||||
expect(dune.downloadTask.mediaEntityId).toBe(movieId);
|
||||
expect(det.body.data.organizationRecords.map((r: { id: string }) => r.id)).toEqual(
|
||||
expect.arrayContaining(dune.organizationRecordIds),
|
||||
);
|
||||
const inv = await h.api("GET", `/inventory?entityIds=${movieId}`);
|
||||
expect(inv.body.data.items[0].versions[0]).toMatchObject({ resolution: "1080p", available: true });
|
||||
const ms = await h.api("GET", `/inventory/media-servers?mediaEntityId=${movieId}`);
|
||||
|
||||
@@ -147,7 +147,7 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) {
|
||||
.select(
|
||||
sql<number>`(select count(*) from download_task dt left join acquisition_plan p on p.id = dt.acquisition_plan_id
|
||||
left join acquisition_slot s on s.id = p.slot_id left join source_release sr on sr.id = dt.source_release_id
|
||||
where (s.media_entity_id = e.id or sr.media_entity_id = e.id) and dt.removed_at is null
|
||||
where (dt.media_entity_id = e.id or s.media_entity_id = e.id or sr.media_entity_id = e.id) and dt.removed_at is null
|
||||
and dt.status in ('queued','submitting','downloading','stalled','paused'))::int`.as("active_downloads"),
|
||||
)
|
||||
.select(sql<string | null>`e.metadata #>> '{posterPath,value}'`.as("poster_path"))
|
||||
@@ -205,7 +205,7 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) {
|
||||
if (query.filter === "downloading")
|
||||
q = q.where(
|
||||
sql<boolean>`exists (select 1 from download_task dt left join acquisition_plan p on p.id = dt.acquisition_plan_id left join acquisition_slot s on s.id = p.slot_id
|
||||
where s.media_entity_id = e.id and dt.removed_at is null and dt.status in ('queued','submitting','downloading','stalled','paused'))`,
|
||||
where (dt.media_entity_id = e.id or s.media_entity_id = e.id) and dt.removed_at is null and dt.status in ('queued','submitting','downloading','stalled','paused'))`,
|
||||
);
|
||||
const p = await paginate(q, query, "e.");
|
||||
const items: LibraryItem[] = p.rows.map((r) => ({
|
||||
@@ -288,10 +288,36 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) {
|
||||
.leftJoin("acquisition_slot as s", "s.id", "p.slot_id")
|
||||
.leftJoin("source_release as sr", "sr.id", "dt.source_release_id")
|
||||
.selectAll("dt")
|
||||
.where((eb) => eb.or([eb("s.media_entity_id", "=", id), eb("sr.media_entity_id", "=", id)]))
|
||||
// 接管任务(无计划 / 站点资源)按识别 / 指定 / 整理回填的 dt.media_entity_id 归属
|
||||
.where((eb) =>
|
||||
eb.or([
|
||||
eb("dt.media_entity_id", "=", id),
|
||||
eb("s.media_entity_id", "=", id),
|
||||
eb("sr.media_entity_id", "=", id),
|
||||
]),
|
||||
)
|
||||
.orderBy("dt.created_at", "desc")
|
||||
.limit(100)
|
||||
.execute();
|
||||
const entityRecs = await db()
|
||||
.selectFrom("organization_record")
|
||||
.select([
|
||||
"id",
|
||||
"status",
|
||||
"action",
|
||||
"trigger",
|
||||
"download_task_id",
|
||||
"source_path",
|
||||
"target_path",
|
||||
"file_succeeded",
|
||||
"file_total",
|
||||
"created_at",
|
||||
"finished_at",
|
||||
])
|
||||
.where("media_entity_id", "=", id)
|
||||
.orderBy("created_at", "desc")
|
||||
.limit(100)
|
||||
.execute();
|
||||
const planIds = dts.map((t) => t.acquisition_plan_id).filter((x): x is string => !!x);
|
||||
const plans = planIds.length
|
||||
? await db().selectFrom("acquisition_plan").selectAll().where("id", "in", planIds).execute()
|
||||
@@ -339,6 +365,19 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) {
|
||||
lastOrganizationStatus: mine[mine.length - 1]?.status ?? null,
|
||||
};
|
||||
}),
|
||||
organizationRecords: entityRecs.map((r) => ({
|
||||
id: r.id,
|
||||
status: r.status,
|
||||
action: r.action,
|
||||
trigger: r.trigger,
|
||||
downloadTaskId: r.download_task_id,
|
||||
sourcePath: r.source_path,
|
||||
targetPath: r.target_path,
|
||||
fileSucceeded: r.file_succeeded,
|
||||
fileTotal: r.file_total,
|
||||
createdAt: new Date(r.created_at).toISOString(),
|
||||
finishedAt: r.finished_at ? new Date(r.finished_at).toISOString() : null,
|
||||
})),
|
||||
};
|
||||
return detail;
|
||||
},
|
||||
@@ -392,8 +431,14 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) {
|
||||
.leftJoin("acquisition_slot as s", "s.id", "p.slot_id")
|
||||
.leftJoin("source_release as sr", "sr.id", "dt.source_release_id")
|
||||
.select(["dt.id", "dt.name", "dt.status", "dt.progress"])
|
||||
.select(sql<string>`coalesce(s.media_entity_id, sr.media_entity_id)`.as("entity_id"))
|
||||
.where((eb) => eb.or([eb("s.media_entity_id", "in", ids), eb("sr.media_entity_id", "in", ids)]))
|
||||
.select(sql<string>`coalesce(dt.media_entity_id, s.media_entity_id, sr.media_entity_id)`.as("entity_id"))
|
||||
.where((eb) =>
|
||||
eb.or([
|
||||
eb("dt.media_entity_id", "in", ids),
|
||||
eb("s.media_entity_id", "in", ids),
|
||||
eb("sr.media_entity_id", "in", ids),
|
||||
]),
|
||||
)
|
||||
.where("dt.removed_at", "is", null)
|
||||
.where("dt.status", "in", [...ACTIVE_DL])
|
||||
.execute(),
|
||||
|
||||
@@ -313,15 +313,18 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
|
||||
}
|
||||
const r = res as Extract<SubmitResult, { hash: string }>;
|
||||
const ins = await sql<{ id: string; inserted: boolean }>`
|
||||
insert into download_task (instance_id, torrent_hash, infohash_v1, infohash_v2, acquisition_plan_id, source_release_id, name,
|
||||
insert into download_task (instance_id, torrent_hash, infohash_v1, infohash_v2, acquisition_plan_id, source_release_id,
|
||||
media_entity_id, name,
|
||||
save_path, content_path, category, tags, managed, origin, status, size_bytes, next_check_at)
|
||||
values (${topo.inst.id}, ${r.hash}, ${r.hash.length === 40 ? r.hash : null}, ${r.hash.length === 64 ? r.hash : null},
|
||||
${plan.id}, ${rel.id}, ${r.name}, ${r.savePath}, ${r.contentPath}, ${topo.category},
|
||||
${plan.id}, ${rel.id}, (select media_entity_id from acquisition_slot where id = ${plan.slot_id}),
|
||||
${r.name}, ${r.savePath}, ${r.contentPath}, ${topo.category},
|
||||
${r.tags}, true, ${r.kind === "already_exists" ? "adopted" : "submitted"},
|
||||
${r.kind === "accepted_unconfirmed" ? "submitting" : "queued"}, ${rel.size_bytes}, now())
|
||||
on conflict (instance_id, torrent_hash) do update
|
||||
set acquisition_plan_id = coalesce(download_task.acquisition_plan_id, excluded.acquisition_plan_id),
|
||||
source_release_id = coalesce(download_task.source_release_id, excluded.source_release_id),
|
||||
media_entity_id = coalesce(download_task.media_entity_id, excluded.media_entity_id),
|
||||
managed = true, tags = excluded.tags, updated_at = now()
|
||||
returning id, (xmax = 0) as inserted`.execute(trx);
|
||||
dtId = ins.rows[0]!.id;
|
||||
@@ -787,7 +790,7 @@ async function adopt(d: AoDeps, topo: InstanceTopology, t: QbTorrent, now: Date,
|
||||
const db = d.ctx.db;
|
||||
const rel = await db
|
||||
.selectFrom("source_release")
|
||||
.select("id")
|
||||
.select(["id", "media_entity_id"])
|
||||
.where((eb) => eb.or([eb("infohash_v1", "=", t.hash), eb("infohash_v2", "=", t.hash)]))
|
||||
.executeTakeFirst();
|
||||
const obs = mapQbState(
|
||||
@@ -804,6 +807,7 @@ async function adopt(d: AoDeps, topo: InstanceTopology, t: QbTorrent, now: Date,
|
||||
infohash_v1: t.infohashV1,
|
||||
infohash_v2: t.infohashV2,
|
||||
source_release_id: rel?.id ?? null,
|
||||
media_entity_id: rel?.media_entity_id ?? null,
|
||||
name: t.name,
|
||||
save_path: t.savePath,
|
||||
content_path: t.contentPath,
|
||||
|
||||
@@ -440,6 +440,11 @@ export async function mergeEntities(
|
||||
.set({ resolved_media_entity_id: intoResolved })
|
||||
.where("resolved_media_entity_id", "=", fromId)
|
||||
.execute();
|
||||
await trx
|
||||
.updateTable("download_task")
|
||||
.set({ media_entity_id: intoResolved })
|
||||
.where("media_entity_id", "=", fromId)
|
||||
.execute();
|
||||
await trx
|
||||
.updateTable("media_entity")
|
||||
.set({ status: "merged", merged_into_id: intoResolved })
|
||||
|
||||
Reference in new issue
Block a user