diff --git a/apps/server/src/modules/acquisition-organize/__tests__/rules-api.test.ts b/apps/server/src/modules/acquisition-organize/__tests__/rules-api.test.ts index d482de68..bbc68605 100644 --- a/apps/server/src/modules/acquisition-organize/__tests__/rules-api.test.ts +++ b/apps/server/src/modules/acquisition-organize/__tests__/rules-api.test.ts @@ -50,7 +50,9 @@ describe.skipIf(!hasPg)("规则 API(真实 PG)", () => { mode: "continuous", seasonScope: "following_seasons", origin: "user", - status: "active", + // 剧集没有已接受的 TMDB 身份 → 总集数无来源:pending(照常搜索/评估,不能完成) + status: "pending", + statusReason: "metadata_pending", }); const once = await h.api("PATCH", `/subscriptions/${sub.body.data.id}`, { seasonScope: "current_season", diff --git a/apps/server/src/modules/acquisition-organize/__tests__/subscription.test.ts b/apps/server/src/modules/acquisition-organize/__tests__/subscription.test.ts new file mode 100644 index 00000000..e92e6844 --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/__tests__/subscription.test.ts @@ -0,0 +1,317 @@ +import { createQbClient } from "@umt/adapters"; +import { createMockQb, createTempTree, type MockQb, makeTorrent, type TempTree } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { evaluateAcquisition } from "../services/acquisition.ts"; +import { submitDownload, syncDownloads } from "../services/downloads.ts"; +import { executeRecord, planImport } from "../services/organize.ts"; +import { applyEpisodeTotal, subscriptionProgress, sweepDueSlots } from "../services/subscriptions.ts"; +import { createHarness, fakeDisk, type Harness, pgAvailable, seedBase, seedEntity, seedRelease } from "./harness.ts"; + +// W5-C 订阅增强集成:Slot×单元多在途 + 忙时延后重评 + 移除释放;季包按缺集选文件(mock qB files/filePrio); +// 总集数未知不结束(疑似缺集);once 完成写历史;pending 不完成;分集洗版只替换能提升的单元。 + +const hasPg = await pgAvailable(); +const SIZE = 300_000; + +describe.skipIf(!hasPg)("订阅增强(真实 PG + mock qB + 临时目录)", () => { + let h: Harness; + let qb: MockQb; + let lib: TempTree; + let cache: TempTree; + const disk = fakeDisk(10_000_000_000); + const torrents = new Map(); + let base: Awaited>; + + beforeAll(async () => { + qb = await createMockQb({ addVisibleAfterPolls: 0 }); + lib = await createTempTree({}, { sentinel: true, prefix: "umt-lib-" }); + cache = await createTempTree({}, { prefix: "umt-cache-" }); + disk.state.watch = [cache.root]; + disk.setPct(10); + h = await createHarness("subscription", { + diskUsage: disk.fn, + torrentSource: async ({ release }) => ({ kind: "file", bytes: torrents.get(release.id)! }), + }); + base = await seedBase(h.db, { libraryPath: lib.root, cachePath: cache.root, qbUrl: qb.url }); + }, 60_000); + + afterAll(async () => { + await h?.close(); + await qb?.close(); + await lib?.cleanup(); + await cache?.cleanup(); + }); + + const db = () => h.db; + + async function seriesSlot( + title: string, + opts: { mode?: "once" | "continuous"; status?: "active" | "pending"; hard?: object; upgrade?: object } = {}, + ) { + const entityId = await seedEntity(h.db, { kind: "series", title, year: 2024 }); + const sub = await db() + .insertInto("subscription") + .values({ + media_entity_id: entityId, + media_kind: "series", + mode: opts.mode ?? "continuous", + status: opts.status ?? "active", + }) + .returning("id") + .executeTakeFirstOrThrow(); + const slot = await db() + .insertInto("acquisition_slot") + .values({ + media_entity_id: entityId, + subscription_id: sub.id, + name: "1080p", + hard_conditions: JSON.stringify(opts.hard ?? { resolutions: ["1080p"] }), + upgrade_threshold: JSON.stringify(opts.upgrade ?? {}), + }) + .returning("id") + .executeTakeFirstOrThrow(); + return { entityId, subId: sub.id, slotId: slot.id }; + } + + async function release(entityId: string, title: string, files?: { path: string; length: number }[]) { + const t = files + ? makeTorrent({ name: title, files, pieceSeed: Math.floor(Math.random() * 1e6) }) + : makeTorrent({ name: `${title}.mkv`, length: SIZE, pieceSeed: Math.floor(Math.random() * 1e6) }); + const id = await seedRelease(h.db, { siteId: base.siteId, entityId, title, size: SIZE, hash: t.hash }); + torrents.set(id, t.bytes); + return { id, hash: t.hash, name: t.name }; + } + + const slotRow = (id: string) => + db().selectFrom("acquisition_slot").selectAll().where("id", "=", id).executeTakeFirstOrThrow(); + const planRow = (id: string) => + db().selectFrom("acquisition_plan").selectAll().where("id", "=", id).executeTakeFirstOrThrow(); + const subRow = (id: string) => + db().selectFrom("subscription").selectAll().where("id", "=", id).executeTakeFirstOrThrow(); + + async function planFor(relId: string) { + const ev = await evaluateAcquisition(h.d, { sourceReleaseId: relId }); + expect(ev.plans, JSON.stringify(ev)).toHaveLength(1); + return ev.plans[0]!; + } + async function submit(planId: string) { + const s = await submitDownload(h.d, planId); + expect(s.status).toBe("submitted"); + return s.downloadTaskId!; + } + async function completeAndImport(hash: string, dtId: string) { + await qb.complete(hash, { writeFiles: true }); + const sync = await syncDownloads(h.d); + expect(sync.completed).toContain(dtId); + const p = await planImport(h.d, { downloadTaskId: dtId, trigger: "auto", actorType: "system" }); + expect(p.status, JSON.stringify(p)).toBe("running"); + const ex = await executeRecord(h.d, p.recordId!); + expect(ex.status).toBe("succeeded"); + } + + it("Slot×单元多在途:两集同时在途;重叠单元的发布延后;下载被移除即释放预留,重评后补上", async () => { + const { entityId, slotId } = await seriesSlot("熊家餐馆"); + const e1 = await release(entityId, "The.Bear.S01E01.1080p.WEB-DL.x264-GRP"); + const e2 = await release(entityId, "The.Bear.S01E02.1080p.WEB-DL.x264-GRP"); + const p1 = await planFor(e1.id); + const p2 = await planFor(e2.id); // Slot 已 reserved,单元不相交 → 第二个在途计划 + expect((await planRow(p1)).unit_keys).toEqual(["s1e1"]); + expect((await planRow(p2)).unit_keys).toEqual(["s1e2"]); + expect((await slotRow(slotId)).status).toBe("reserved"); + + // 同一集的另一个版本:单元在途 → 延后(deferred),不丢 + const e2b = await release(entityId, "The.Bear.S01E02.1080p.WEB-DL.x265-OTHER"); + const ev = await evaluateAcquisition(h.d, { sourceReleaseId: e2b.id }); + expect(ev.plans).toHaveLength(0); + expect(ev.deferred).toEqual([{ sourceReleaseId: e2b.id, slotId, reason: "units_inflight" }]); + expect( + await db().selectFrom("slot_deferral").select(["unit_keys"]).where("slot_id", "=", slotId).execute(), + ).toEqual([{ unit_keys: ["s1e2"] }]); + + const dt1 = await submit(p1); + const dt2 = await submit(p2); + expect((await slotRow(slotId)).status).toBe("downloading"); + await completeAndImport(e1.hash, dt1); + // E01 入库,但 E02 仍在途:Slot 显示在途阶段而不是静止态;分集账本登记 E01 + const s1 = await slotRow(slotId); + expect(s1.status).toBe("downloading"); + expect(s1.current_asset_id).not.toBeNull(); + expect(s1.next_evaluate_at).not.toBeNull(); + const ledger = await db() + .selectFrom("slot_unit as su") + .innerJoin("content_unit as u", "u.id", "su.content_unit_id") + .select(["u.episode_number", "su.plan_id"]) + .where("su.slot_id", "=", slotId) + .execute(); + expect(ledger).toEqual([{ episode_number: 1, plan_id: p1 }]); + + // 用户在 qB 里删掉 E02:确认移除 → 计划失败释放单元 → Slot 到期重评 → 延后的 E02 另一版本得到计划 + const qc = createQbClient({ instanceId: "t", baseUrl: qb.url, username: "admin", password: "adminadmin" }); + await qc.delete([e2.hash], { deleteFiles: false }); + await syncDownloads(h.d); + expect(await planRow(p2)).toMatchObject({ status: "failed", status_reason: "download_removed" }); + expect((await slotRow(slotId)).status).toBe("attention"); + h.engine.enqueued.length = 0; + expect(await sweepDueSlots(h.d)).toBeGreaterThanOrEqual(1); + expect(h.engine.enqueued).toContainEqual( + expect.objectContaining({ task: "evaluateAcquisition", input: { acquisitionSlotId: slotId } }), + ); + const re = await evaluateAcquisition(h.d, { acquisitionSlotId: slotId }); + expect(re.plans).toHaveLength(1); + expect(await planRow(re.plans[0]!)).toMatchObject({ source_release_id: e2b.id, unit_keys: ["s1e2"] }); + // e2b 的延后已消化;被移除的原 E02 此时单元在途(e2b)→ 转为延后,等 e2b 结束后再判 + const left = await db() + .selectFrom("slot_deferral") + .select("source_release_id") + .where("slot_id", "=", slotId) + .execute(); + expect(left.map((x) => x.source_release_id)).toEqual([e2.id]); + // 候选评估记录带 trace / unitKeys + const art = await db() + .selectFrom("artifact") + .select("content") + .where("subject_id", "=", slotId) + .where("title", "=", "candidate_evaluation") + .execute(); + const byRel = new Map(art.map((a) => [(a.content as { sourceReleaseId: string }).sourceReleaseId, a.content])); + expect(byRel.get(e2b.id)).toMatchObject({ decision: "accept", unitKeys: ["s1e2"], pri: 100 }); + }, 60_000); + + it("季包:总集数已知 → 只预留缺集;提交后读文件清单,已覆盖/在途的集文件优先级置 0", async () => { + const { entityId, slotId } = await seriesSlot("选集剧"); + await db() + .transaction() + .execute((trx) => applyEpisodeTotal(trx, { entityId, season: 1, fresh: 4, now: new Date() })); + const e1 = await release(entityId, "Pick.Show.S01E01.1080p.WEB-DL.x264-GRP"); + await completeAndImport(e1.hash, await submit(await planFor(e1.id))); + const e2 = await release(entityId, "Pick.Show.S01E02.1080p.WEB-DL.x264-GRP"); + const p2 = await planFor(e2.id); // E02 在途(未提交) + + const packTitle = "Pick.Show.S01.1080p.WEB-DL.x264-GRP"; + const files = [1, 2, 3, 4].map((e) => ({ path: `Pick.Show.S01E0${e}.1080p.WEB-DL.x264-GRP.mkv`, length: SIZE })); + const pack = await release(entityId, packTitle, [ + ...files, + { path: "Pick.Show.S01E01.chs.srt", length: 100 }, + { path: "Pick.Show.S01E03.chs.srt", length: 100 }, + ]); + const pp = await planFor(pack.id); + expect((await planRow(pp)).unit_keys).toEqual(["s1e3", "s1e4"]); + const before = qb.callsTo("torrents/filePrio").length; + await submit(pp); + const prio = qb.callsTo("torrents/filePrio").slice(before); + expect(prio).toHaveLength(1); + const t = qb.get(pack.hash)!; + const off = t.files.filter((f) => f.priority === 0).map((f) => f.name.split("/").pop()); + expect(off.sort()).toEqual( + [ + "Pick.Show.S01E01.1080p.WEB-DL.x264-GRP.mkv", + "Pick.Show.S01E01.chs.srt", + "Pick.Show.S01E02.1080p.WEB-DL.x264-GRP.mkv", + ].sort(), + ); + expect((await planRow(pp)).file_selection).toMatchObject({ mode: "selected", selectedUnits: ["s1e3", "s1e4"] }); + // 进度:E01 已覆盖,E02–E04 在途 + const prog = await subscriptionProgress(db(), (await slotRow(slotId)).subscription_id!); + expect(prog!.seasons[0]).toMatchObject({ seasonNumber: 1, episodeTotal: 4, totalKnown: true, missing: [] }); + expect(prog!.seasons[0]!.units.map((u) => u.state)).toEqual(["covered", "inflight", "inflight", "inflight"]); + expect(prog!.slots[0]!.inflightPlans.map((x) => x.planId).sort()).toEqual([p2, pp].sort()); + + // 认不出集号的整包:整包下载并记理由 + const { entityId: e2Id } = await seriesSlot("整包剧"); + const odd = await release(e2Id, "Odd.Show.S01.1080p.WEB-DL.x264-GRP", [ + { path: "disc1.mkv", length: SIZE }, + { path: "disc2.mkv", length: SIZE }, + ]); + const po = await planFor(odd.id); + expect((await planRow(po)).unit_keys).toEqual(["s1*"]); + const n0 = qb.callsTo("torrents/filePrio").length; + await submit(po); + expect(qb.callsTo("torrents/filePrio").length).toBe(n0); + expect((await planRow(po)).file_selection).toMatchObject({ mode: "whole", reason: "unmapped_video_file" }); + }, 60_000); + + it("总集数未知:入库后不结束、记疑似缺集;总集数就绪后补齐 → once 订阅完成并写历史", async () => { + const { entityId, slotId, subId } = await seriesSlot("未知集数剧", { mode: "once" }); + for (const ep of [1, 3]) { + const r = await release(entityId, `Unknown.Total.S01E0${ep}.1080p.WEB-DL.x264-GRP`); + await completeAndImport(r.hash, await submit(await planFor(r.id))); + } + expect((await slotRow(slotId)).status).toBe("empty"); + expect(await subRow(subId)).toMatchObject({ status: "active" }); + const dec = await db() + .selectFrom("decision") + .selectAll() + .where("dedupe_key", "=", `suspected_missing:${slotId}`) + .executeTakeFirstOrThrow(); + expect(dec.evidence).toMatchObject({ suspectedMissing: [{ season: 1, episodes: [2] }] }); + + await db() + .transaction() + .execute((trx) => applyEpisodeTotal(trx, { entityId, season: 1, fresh: 3, now: new Date() })); + const r2 = await release(entityId, "Unknown.Total.S01E02.1080p.WEB-DL.x264-GRP"); + await completeAndImport(r2.hash, await submit(await planFor(r2.id))); + expect((await slotRow(slotId)).status).toBe("satisfied"); + expect(await subRow(subId)).toMatchObject({ status: "completed", status_reason: "slot_satisfied" }); + const hist = await db() + .selectFrom("subscription_history") + .selectAll() + .where("subscription_id", "=", subId) + .execute(); + expect(hist).toHaveLength(1); + expect(hist[0]).toMatchObject({ reason: "slot_satisfied", media_entity_id: entityId, mode: "once" }); + expect((hist[0]!.progress as { seasons: { complete: boolean }[] }).seasons[0]!.complete).toBe(true); + // 历史 API + 重新订阅(改回 active) + const api = await h.api("GET", `/subscription-history?subscriptionId=${subId}`); + expect(api.body.data.items).toHaveLength(1); + const re = await h.api("PATCH", `/subscriptions/${subId}`, { status: "active" }); + expect(re.body.data.status).toBe("active"); + }, 90_000); + + it("pending(TMDB 未就绪)照常下载但不完成;暂停的订阅不评估", async () => { + const { entityId, slotId, subId } = await seriesSlot("待定剧", { mode: "once", status: "pending" }); + await db() + .transaction() + .execute((trx) => applyEpisodeTotal(trx, { entityId, season: 1, fresh: 1, now: new Date() })); + const r = await release(entityId, "Pending.Show.S01E01.1080p.WEB-DL.x264-GRP"); + await completeAndImport(r.hash, await submit(await planFor(r.id))); + expect((await slotRow(slotId)).status).toBe("empty"); + expect((await subRow(subId)).status).toBe("pending"); + + const paused = await seriesSlot("暂停剧", { status: "active" }); + await db().updateTable("subscription").set({ status: "paused" }).where("id", "=", paused.subId).execute(); + const pr = await release(paused.entityId, "Paused.Show.S01E01.1080p.WEB-DL.x264-GRP"); + const ev = await evaluateAcquisition(h.d, { sourceReleaseId: pr.id }); + expect(ev).toMatchObject({ evaluated: 0, plans: [] }); + }, 60_000); + + it("分集洗版:只替换能提升的单元;同档不换(trace 可读)", async () => { + const { entityId, slotId } = await seriesSlot("洗版剧", { + hard: { resolutions: ["720p", "1080p"] }, + upgrade: { enabled: true, allowResolutionUpgrade: true }, + }); + const r = await release(entityId, "Wash.Show.S01E01.720p.WEB-DL.x264-GRP"); + await completeAndImport(r.hash, await submit(await planFor(r.id))); + const same = await release(entityId, "Wash.Show.S01E01.720p.WEB-DL.x264-OTHER"); + const ev = await evaluateAcquisition(h.d, { sourceReleaseId: same.id }); + expect(ev.plans).toHaveLength(0); + const up = await release(entityId, "Wash.Show.S01E01-E02.1080p.WEB-DL.x264-GRP"); + // E02 还缺 → 追更(fill);只有 E01 覆盖时的纯升级场景见下一条 + const fill = await planFor(up.id); + expect(await planRow(fill)).toMatchObject({ purpose: "fill", unit_keys: ["s1e2"] }); + await db().updateTable("acquisition_plan").set({ status: "cancelled" }).where("id", "=", fill).execute(); + await db().updateTable("acquisition_slot").set({ status: "empty" }).where("id", "=", slotId).execute(); + const up1 = await release(entityId, "Wash.Show.S01E01.1080p.WEB-DL.x264-GRP"); + const pu = await planFor(up1.id); + expect(await planRow(pu)).toMatchObject({ purpose: "upgrade", unit_keys: ["s1e1"] }); + const art = await db() + .selectFrom("artifact") + .select("content") + .where("subject_id", "=", slotId) + .where("title", "=", "candidate_evaluation") + .execute(); + const sameEval = art + .map((a) => a.content as { sourceReleaseId: string; trace?: string[] }) + .find((a) => a.sourceReleaseId === same.id); + expect(sameEval?.trace?.[0]).toBe("与当前版本同档,不反复更换"); + }, 60_000); +}); diff --git a/apps/server/src/modules/acquisition-organize/index.ts b/apps/server/src/modules/acquisition-organize/index.ts index 2f5c74d4..f02ae782 100644 --- a/apps/server/src/modules/acquisition-organize/index.ts +++ b/apps/server/src/modules/acquisition-organize/index.ts @@ -12,6 +12,7 @@ import { registerOrganizationRoutes } from "./routes/organization.ts"; import { registerRuleRoutes } from "./routes/rules.ts"; import { type AoDeps, automationBlock, enqueueSweep, enqueueTracked, timeSlot } from "./services/common.ts"; import { SWEEP_SLOT_MS, sweepReservedPlans, syncDownloads } from "./services/downloads.ts"; +import { sweepDueSlots, sweepMetadataRefresh, sweepSubscriptionSearch } from "./services/subscriptions.ts"; import { loadVolumes, usedPercent } from "./services/volumes.ts"; import { errMessage } from "./util.ts"; @@ -55,6 +56,10 @@ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): Se if (over) await enqueueThresholdCleanup(d); await sweepReservedPlans(d); await sweepPendingImports(d); + // W5-C:Slot 到期重评(忙时延后的发布 / 定时)、订阅周期主动搜索、追踪剧集的总集数刷新 + await sweepDueSlots(d); + await sweepSubscriptionSearch(d); + await sweepMetadataRefresh(d); } const every = d.ports.options.reconcileIntervalMs; if (every > 0) { diff --git a/apps/server/src/modules/acquisition-organize/ports.ts b/apps/server/src/modules/acquisition-organize/ports.ts index 445fc693..df59f1bc 100644 --- a/apps/server/src/modules/acquisition-organize/ports.ts +++ b/apps/server/src/modules/acquisition-organize/ports.ts @@ -41,6 +41,10 @@ export interface AoOptions { missingConfirmAfter: number; /** 后台资产对账周期(0 = 不自动跑) */ reconcileIntervalMs: number; + /** 未满足 Slot 的定时重评周期(next_evaluate_at),缺省 6h;0 = 只在计划结束时重评 */ + slotReevaluateIntervalMs?: number; + /** 追踪中剧集的元数据(总集数)刷新周期,缺省 6h;0 = 关 */ + metadataRefreshIntervalMs?: number; } export interface AoPorts { diff --git a/apps/server/src/modules/acquisition-organize/routes/rules.ts b/apps/server/src/modules/acquisition-organize/routes/rules.ts index 4a82589f..bab8900e 100644 --- a/apps/server/src/modules/acquisition-organize/routes/rules.ts +++ b/apps/server/src/modules/acquisition-organize/routes/rules.ts @@ -6,6 +6,7 @@ import { type PreferenceSuppression, type PreferenceTag, type Subscription, + type SubscriptionHistory, } from "@umt/contracts"; import { type DB, sql, type Transaction } from "@umt/db"; import { evaluateCandidate, parseRelease } from "@umt/domain"; @@ -16,6 +17,7 @@ import { ApiException } from "../../../http/errors.ts"; import type { ModuleApp } from "../../types.ts"; import { createSlotsFromTemplates, publishedAcquisitionPolicies } from "../services/acquisition.ts"; import type { AoDeps } from "../services/common.ts"; +import { applyEpisodeTotal, enqueueSubscriptionSearch, subscriptionProgress } from "../services/subscriptions.ts"; import { actorOf, conflict, errMessage, hashJson, jsonb, notFound, toApi, toApiList } from "../util.ts"; import { mount, paginate } from "./helpers.ts"; @@ -137,12 +139,25 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { .executeTakeFirst(); if (!e) throw new ApiException("bad_request", "mediaEntityId 不存在"); } + // pending:TMDB 未就绪(未建档,或剧集没有已接受的 TMDB 身份 → 总集数无来源):照常搜索/评估,不能完成 + const tmdbReady = + !!body.mediaEntityId && + (body.mediaKind === "movie" || + !!(await db() + .selectFrom("external_identity") + .select("id") + .where("media_entity_id", "=", body.mediaEntityId) + .where("provider", "=", "tmdb") + .where("status", "=", "accepted") + .executeTakeFirst())); const row = await db() .transaction() .execute(async (trx) => { const r = await trx .insertInto("subscription") .values({ + status: tmdbReady ? "active" : "pending", + status_reason: tmdbReady ? null : "metadata_pending", media_entity_id: body.mediaEntityId ?? null, season_unit_id: body.seasonUnitId ?? null, title_query: body.titleQuery ?? null, @@ -181,7 +196,21 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { } return r; }); - if (row.media_entity_id) await kickEvaluate(d, { subscriptionId: row.id }); + if (row.media_entity_id) { + await kickEvaluate(d, { subscriptionId: row.id }); + // 新订阅:立即主动搜索一次(之后按周期);剧集先刷新元数据拿季总集数 + await enqueueSubscriptionSearch(d, row.id, "created").catch((err) => + d.ctx.logger.debug({ err: { message: errMessage(err) } }, "订阅搜索入队失败"), + ); + if (row.media_kind !== "movie" && tmdbReady) + await d.ctx.engine + .enqueue( + "refreshMetadata", + { mediaEntityId: row.media_entity_id }, + { idempotencyKey: `refreshMetadata:subscription:${row.id}`, traceId: row.id }, + ) + .catch((err) => d.ctx.logger.debug({ err: { message: errMessage(err) } }, "元数据刷新入队失败")); + } return toApi(row); }, { tag: TAG }, @@ -210,6 +239,94 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { { tag: TAG }, ); + mount( + app, + Api.subscriptions.search, + async (_c, { params }) => { + const sub = await db() + .selectFrom("subscription") + .select(["id", "media_entity_id", "status"]) + .where("id", "=", params.id) + .executeTakeFirst(); + if (!sub) notFound("订阅"); + if (!sub!.media_entity_id) throw new ApiException("bad_request", "订阅尚未关联作品,无法按标题搜索"); + try { + return await enqueueSubscriptionSearch(d, sub!.id, "manual"); + } catch (err) { + throw new ApiException("upstream_unavailable", `搜索任务入队失败:${errMessage(err)}`); + } + }, + { tag: TAG }, + ); + mount( + app, + Api.subscriptions.progress, + async (_c, { params }) => (await subscriptionProgress(db(), params.id)) ?? notFound("订阅"), + { tag: TAG }, + ); + mount( + app, + Api.subscriptions.setEpisodeTotal, + async (_c, { params, body }) => { + const sub = await db() + .selectFrom("subscription") + .select(["id", "media_entity_id", "media_kind"]) + .where("id", "=", params.id) + .executeTakeFirst(); + if (!sub) notFound("订阅"); + if (!sub!.media_entity_id || sub!.media_kind === "movie") + throw new ApiException("bad_request", "只有已关联作品的剧集/综艺订阅有季总集数"); + await db() + .transaction() + .execute(async (trx) => { + const now = d.ports.now(); + if (body.locked) { + await applyEpisodeTotal(trx, { + entityId: sub!.media_entity_id!, + season: body.seasonNumber, + manual: { total: body.total, locked: true }, + now, + }); + } else { + // 未锁定:按注入口径只升 + await applyEpisodeTotal(trx, { + entityId: sub!.media_entity_id!, + season: body.seasonNumber, + injected: body.total, + now, + }); + } + }); + return (await subscriptionProgress(db(), params.id))!; + }, + { tag: TAG }, + ); + mount( + app, + Api.subscriptionHistory.list, + async (_c, { query }) => { + let q = db().selectFrom("subscription_history").selectAll(); + if (query.mediaEntityId) q = q.where("media_entity_id", "=", query.mediaEntityId); + if (query.subscriptionId) q = q.where("subscription_id", "=", query.subscriptionId); + const p = await paginate(q, query); + return { items: toApiList(p.rows), nextCursor: p.nextCursor }; + }, + { tag: TAG }, + ); + mount( + app, + Api.subscriptionHistory.get, + async (_c, { params }) => { + const r = await db() + .selectFrom("subscription_history") + .selectAll() + .where("id", "=", params.id) + .executeTakeFirst(); + return toApi(r ?? notFound("订阅历史")); + }, + { tag: TAG }, + ); + // ── 收录 Slot ── mount( app, diff --git a/apps/server/src/modules/acquisition-organize/services/acquisition.ts b/apps/server/src/modules/acquisition-organize/services/acquisition.ts index fa0b0a51..39043dd6 100644 --- a/apps/server/src/modules/acquisition-organize/services/acquisition.ts +++ b/apps/server/src/modules/acquisition-organize/services/acquisition.ts @@ -10,19 +10,36 @@ import { sql } from "@umt/db"; import { type AcquisitionMode, checkBudget, + computeUnitGaps, determineMode, + type EvaluationContext, + episodeKey, evaluateCandidate, + improvableUnits, type ParsedRelease, parseRelease, priorityOf, type QualityFacts, type SlotEvaluation, type SlotState, + seasonWildcard, transition, + unitsOverlap, } from "@umt/domain"; import { fromParsedColumns, slotSpecFromRow, toContractSlotEvaluation } from "@umt/mapping"; import { hashJson, jsonb, startOfDay, toApi } from "../util.ts"; import { type AoDeps, automationBlock, type Exec, enqueueTracked, loadInstances, recordCandidate } from "./common.ts"; +import { + clearDeferral, + rangeOf, + recordDeferral, + scheduleSlotReevaluate, + seasonLedgers, + slotCoveredUnits, + slotInflightPlans, + subParams, + TRACKING_SUB_STATUSES, +} from "./subscriptions.ts"; import { decideVolume, loadVolumes } from "./volumes.ts"; // evaluateAcquisition:release.resolved → 匹配订阅/Slot → 模式 → 确定性评估 → 预算/缓存卷 → 冻结计划并预留。 @@ -45,6 +62,8 @@ export interface EvaluateOutcome { evaluated: number; plans: string[]; skipped: { sourceReleaseId: string; reason: string }[]; + /** Slot 忙(单元在途冲突 / 电影 Slot 在途)而延后的发布;计划结束后重评 */ + deferred: { sourceReleaseId: string; slotId: string; reason: string }[]; } const PRIORITY: Record, AcquisitionPriority> = { @@ -110,25 +129,6 @@ async function releaseUnits( return [...ids, ...keys]; } -/** Slot 已满足的单元:该 Slot 的计划带来的资产 + 当前资产的覆盖(content_unit id 与合成键) */ -async function satisfiedUnits(exec: Exec, slot: AcquisitionSlotRow): Promise { - const assets = await sql<{ id: string; media_info: unknown }>` - select distinct a.id, a.media_info from media_asset a - left join download_task dt on dt.id = a.download_task_id - left join acquisition_plan p on p.id = dt.acquisition_plan_id - where a.status = 'active' and a.parent_asset_id is null - and (p.slot_id = ${slot.id} or a.id = ${slot.current_asset_id})`.execute(exec); - if (assets.rows.length === 0) return []; - const ids = assets.rows.map((a) => a.id); - const cov = await exec - .selectFrom("asset_unit_coverage") - .select("content_unit_id") - .where("media_asset_id", "in", ids) - .execute(); - const keys = assets.rows.flatMap((a) => (a.media_info as { _umt?: { unitKeys?: string[] } })?._umt?.unitKeys ?? []); - return [...cov.map((c) => c.content_unit_id), ...keys]; -} - async function resolveEntity(exec: Exec, id: string): Promise { let cur = await exec.selectFrom("media_entity").selectAll().where("id", "=", id).executeTakeFirst(); for (let i = 0; cur?.merged_into_id && i < 10; i++) { @@ -229,7 +229,12 @@ async function note(c: Ctx, base: Omit) { } export async function evaluateAcquisition(d: AoDeps, input: EvaluateInput): Promise { - const c: Ctx = { d, now: d.ports.now(), out: { evaluated: 0, plans: [], skipped: [] }, draft: input.draft === true }; + const c: Ctx = { + d, + now: d.ports.now(), + out: { evaluated: 0, plans: [], skipped: [], deferred: [] }, + draft: input.draft === true, + }; const db = d.ctx.db; if (input.sourceReleaseId) { const rel = await db @@ -266,6 +271,22 @@ export async function evaluateAcquisition(d: AoDeps, input: EvaluateInput): Prom )?.media_entity_id ?? null; } if (!entityId) return c.out; + // 先评该 Slot 延后的发布(按集号升序,连续更新的集依次补上),再评最近已识别的发布 + const deferred = input.acquisitionSlotId + ? await db + .selectFrom("slot_deferral as sd") + .innerJoin("source_release as r", "r.id", "sd.source_release_id") + .selectAll("r") + .where("sd.slot_id", "=", input.acquisitionSlotId) + .orderBy("sd.created_at") + .execute() + : []; + const epStart = (r: SourceReleaseRow) => { + const p = parsedOf(r); + return (p.season ?? 1) * 100_000 + (p.episodes?.start ?? 0); + }; + deferred.sort((a, b) => epStart(a) - epStart(b)); + const seen = new Set(deferred.map((r) => r.id)); const rels = await db .selectFrom("source_release") .selectAll() @@ -275,7 +296,8 @@ export async function evaluateAcquisition(d: AoDeps, input: EvaluateInput): Prom .orderBy("first_seen_at", "desc") .limit(50) .execute(); - for (const r of rels) await evaluateRelease(c, r, [], input.acquisitionSlotId); + for (const r of [...deferred, ...rels.filter((x) => !seen.has(x.id))]) + await evaluateRelease(c, r, [], input.acquisitionSlotId); return c.out; } @@ -372,7 +394,11 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon // 订阅停止条件 for (const s of subs) { const stopAfter = (s.params as { stopAfter?: string } | null)?.stopAfter; - if (s.status === "active" && stopAfter && Date.parse(stopAfter) < now.getTime()) { + if ( + (TRACKING_SUB_STATUSES as readonly string[]).includes(s.status) && + stopAfter && + Date.parse(stopAfter) < now.getTime() + ) { await db .updateTable("subscription") .set({ status: "completed", status_reason: "stop_after_reached" }) @@ -381,7 +407,10 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon s.status = "completed"; } } - const activeSubs = new Map(subs.filter((s) => s.status === "active").map((s) => [s.id, s])); + // pending(TMDB 未就绪)照常评估与下载,只是不能完成 + const activeSubs = new Map( + subs.filter((s) => (TRACKING_SUB_STATUSES as readonly string[]).includes(s.status)).map((s) => [s.id, s]), + ); const tracked = activeSubs.size > 0 || slots.some((s) => s.subscription_id === null); const releaseSeason = parsed.season ?? (parsed.episodes ? 1 : undefined); @@ -416,97 +445,293 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon if (su?.season_number !== null && su?.season_number !== undefined && su.season_number !== releaseSeason) continue; } c.out.evaluated++; - const base = { slotId: slot.id, sourceReleaseId: rel.id, title: rel.title, planId: null as string | null }; - if (BUSY.includes(slot.status)) { - await note(c, { ...base, decision: "skipped", mode: null, rejected: [], reasons: [`slot_busy:${slot.status}`] }); - continue; - } - const spec = slotSpecFromRow(toApi(slot)); - const satisfied = await satisfiedUnits(db, slot); - const slotSatisfied = slot.status === "satisfied" || slot.status === "upgrade_pending"; - const gaps = units.filter((u) => !satisfied.includes(u)); - const current = await currentFacts(db, slot.current_asset_id); - const fill = isMovie ? !slotSatisfied : gaps.length > 0 || !current; - const ev: SlotEvaluation = evaluateCandidate(spec, parsed, fill ? null : current); - const flat = toContractSlotEvaluation(ev); - if (!flat.acceptable) { - await note(c, { - ...base, - decision: ev.decision as "reject" | "unknown_field", - mode: null, - rejected: flat.rejected, - reasons: flat.reasons, - }); - continue; - } - const mode: AcquisitionMode | null = - forcedMode ?? - determineMode({ - entity: { tracked, kind: isMovie ? "movie" : "tv" }, - releaseUnits: units, - satisfiedUnits: satisfied, - slotSatisfied, - now: now.toISOString(), - newTitleWindowDays: d.ports.options.newTitleWindowDays, - upgradeCandidate: ev.decision === "upgrade", - }).mode; - if (!mode) { - await note(c, { - ...base, - decision: "skipped", - mode: null, - rejected: [], - reasons: ["mode:none", ...flat.reasons], - }); - continue; - } - if (planned) { - await note(c, { ...base, decision: "skipped", mode, rejected: [], reasons: [`covered_by_plan:${planned}`] }); - continue; - } - const gate = await gates(c, rel, mode, sub); - if (!gate.ok) { - await note(c, { ...base, decision: "skipped", mode, rejected: gate.rejected, reasons: gate.reasons }); - skip(gate.reasons[0] ?? "gated"); - continue; - } - const planId = await reserve(c, { - slot, + const outcome = await evaluateForSlot(c, { rel, - mode, - ev, + entity, + parsed, + units, + slot, sub, - instanceId: gate.instanceId, - volumeKey: gate.volumeKey, - savePath: gate.savePath, + tracked, + forcedMode, + releaseSeason: releaseSeason ?? 1, + slotCount: slots.length, + alreadyPlanned: planned, }); - if (!planId) { - await note(c, { ...base, decision: "skipped", mode, rejected: [], reasons: ["slot_changed_or_reserved"] }); - continue; - } - planned = planId; - c.out.plans.push(planId); + if (outcome.planId) planned = outcome.planId; + if (!outcome.deferred) await clearDeferral(db, slot.id, rel.id); + if (!isMovie && sub) await scheduleSlotReevaluate(d, db, slot.id); + } +} + +/** 站点字段 → 评估上下文(做种数、促销下载系数、副标题/标签、站点 id) */ +function contextOf(rel: SourceReleaseRow): EvaluationContext { + const promo = (rel.promotion ?? {}) as { kind?: string; raw?: { downloadFactor?: number | null } }; + const factor = + typeof promo.raw?.downloadFactor === "number" + ? promo.raw.downloadFactor + : promo.kind === "free" || promo.kind === "free2up" + ? 0 + : promo.kind === "50pctdown" || promo.kind === "50pctdown2up" + ? 0.5 + : promo.kind === "30pctdown" + ? 0.3 + : promo.kind === "none" || promo.kind === "2up" + ? 1 + : undefined; + return { + siteId: rel.site_id, + ...(rel.subtitle ? { subtitle: rel.subtitle } : {}), + labels: rel.tags_raw ?? [], + ...(rel.seeders !== null && rel.seeders !== undefined ? { seeders: rel.seeders } : {}), + ...(factor !== undefined ? { downloadVolumeFactor: factor } : {}), + }; +} + +interface SlotInput { + rel: SourceReleaseRow; + entity: MediaEntityRow; + parsed: ParsedRelease; + units: string[]; + slot: AcquisitionSlotRow; + sub: SubscriptionRow | undefined; + tracked: boolean; + forcedMode: AcquisitionMode | null; + releaseSeason: number; + slotCount: number; + alreadyPlanned: string | null; +} + +/** + * 单个 Slot 的评估(对照 MP 缺集 = 媒体库已有 ∩ 订阅范围 − note): + * - 电影 Slot 有在途计划 → 延后(deferred); + * - 剧集:需要的单元 = 候选单元 − 已覆盖 − 在途;全部在途 → 延后;都已覆盖 → 分集洗版(只取能提升的单元); + * - 季包:总集数已知按缺集展开,未知用整季通配预留(提交后开文件清单再按缺集选文件)。 + */ +async function evaluateForSlot(c: Ctx, p: SlotInput): Promise<{ planId: string | null; deferred: boolean }> { + const { d, now } = c; + const db = d.ctx.db; + const { rel, parsed, slot, sub } = p; + const isMovie = p.entity.kind === "movie"; + const base = { slotId: slot.id, sourceReleaseId: rel.id, title: rel.title, planId: null as string | null }; + const done = { planId: null, deferred: false }; + const params = subParams(sub); + if (params.siteIds?.length && !params.siteIds.includes(rel.site_id)) { + await note(c, { ...base, decision: "skipped", mode: null, rejected: [], reasons: ["site_out_of_scope"] }); + return done; + } + const spec = slotSpecFromRow(toApi(slot)); + const ctx = contextOf(rel); + const inflight = await slotInflightPlans(db, slot.id); + const inflightKeys = inflight.flatMap((x) => x.unitKeys); + const defer = async (reason: string, unitKeys: string[], mode: AcquisitionMode | null = null) => { + await recordDeferral(db, { slotId: slot.id, sourceReleaseId: rel.id, reason, unitKeys }); await note(c, { ...base, - planId, - decision: ev.decision as "accept" | "upgrade", + decision: "deferred", mode, rejected: [], - reasons: flat.reasons, + reasons: [`deferred:${reason}`, ...inflight.map((x) => `inflight:${x.id}`)], + unitKeys, }); - if (c.draft) continue; - await enqueueTracked(d, { - kind: "submitDownload", - idempotencyKey: `plan:${planId}`, - input: { acquisitionPlanId: planId }, - subjectType: "acquisition_plan", - subjectId: planId, - }).then( - ({ taskId }) => db.updateTable("acquisition_plan").set({ task_id: taskId }).where("id", "=", planId).execute(), - (err) => - d.ctx.logger.warn({ err: { message: (err as Error).message }, planId }, "submitDownload 入队失败,留给巡检补投"), - ); + c.out.deferred.push({ sourceReleaseId: rel.id, slotId: slot.id, reason }); + return { planId: null, deferred: true }; + }; + + let ev: SlotEvaluation; + let planUnits: string[] = []; + let releaseKeys: string[] = []; + let covered = new Set(); + const slotSatisfied = slot.status === "satisfied" || slot.status === "upgrade_pending"; + if (isMovie) { + if (inflight.length > 0 || BUSY.includes(slot.status)) return defer("slot_busy", []); + const current = await currentFacts(db, slot.current_asset_id); + ev = evaluateCandidate(spec, parsed, slotSatisfied ? current : null, ctx); + } else { + const cov = await slotCoveredUnits(db, slot, { entityWide: p.slotCount <= 1 }); + covered = cov.keys; + const range = rangeOf(sub); + const isPack = !parsed.episodes && !parsed.airDate; + let wanted: string[]; + let reservedHit: string[]; + if (isPack) { + const ledger = (await seasonLedgers(db, p.entity.id)).get(p.releaseSeason); + if (ledger?.known && ledger.total) { + releaseKeys = Array.from({ length: ledger.total }, (_, i) => episodeKey(p.releaseSeason, i + 1)); + const g = computeUnitGaps({ + units: releaseKeys, + covered: [...covered], + inflight: inflightKeys, + ...(range ? { range } : {}), + }); + wanted = g.missing; + reservedHit = g.reserved; + ctx.unitCount = ledger.total; + } else { + // 总集数未知:整季通配预留;与同季在途单元冲突 → 延后 + releaseKeys = [seasonWildcard(p.releaseSeason)]; + reservedHit = unitsOverlap(releaseKeys, inflightKeys); + wanted = reservedHit.length ? [] : releaseKeys; + } + } else { + releaseKeys = unitKeysOf(parsed); + const g = computeUnitGaps({ + units: releaseKeys, + covered: [...covered], + inflight: inflightKeys, + ...(range ? { range } : {}), + }); + if (releaseKeys.length > 0 && g.outOfRange.length === releaseKeys.length) { + await note(c, { ...base, decision: "skipped", mode: null, rejected: [], reasons: ["episode_out_of_range"] }); + return done; + } + wanted = g.missing; + reservedHit = g.reserved; + } + if (wanted.length > 0) { + ev = evaluateCandidate(spec, parsed, null, ctx); + planUnits = wanted; + } else if (reservedHit.length > 0) { + return defer("units_inflight", reservedHit); + } else { + // 都已覆盖:分集洗版,只取能提升的单元(对照 MP get_best_version_interested_episodes) + const unitCurrent = await unitQuality(db, slot, releaseKeys); + const withCurrent = releaseKeys.filter((k) => unitCurrent[k]); + const firstCurrent = withCurrent.length + ? unitCurrent[withCurrent[0]!]! + : await currentFacts(db, slot.current_asset_id); + if (!spec.upgrade?.enabled || withCurrent.length === 0) { + ev = evaluateCandidate(spec, parsed, firstCurrent ?? {}, ctx); + } else { + const imp = improvableUnits(spec, parsed, withCurrent, unitCurrent, ctx); + if (imp.units.length === 0) { + ev = evaluateCandidate(spec, parsed, firstCurrent, ctx); + } else { + const busyHit = unitsOverlap(imp.units, inflightKeys); + if (busyHit.length) return defer("units_inflight", busyHit, "upgrade"); + ev = evaluateCandidate(spec, parsed, unitCurrent[imp.units[0]!] ?? null, ctx); + planUnits = imp.units; + } + } + } } + const flat = toContractSlotEvaluation(ev); + const evalExtra = { pri: flat.pri, trace: flat.trace }; + if (!flat.acceptable) { + await note(c, { + ...base, + decision: ev.decision as "reject" | "unknown_field", + mode: null, + rejected: flat.rejected, + reasons: flat.reasons, + ...evalExtra, + }); + return done; + } + const mode: AcquisitionMode | null = + p.forcedMode ?? + determineMode({ + entity: { tracked: p.tracked, kind: isMovie ? "movie" : "tv" }, + releaseUnits: isMovie ? p.units : releaseKeys, + satisfiedUnits: isMovie ? [] : releaseKeys.filter((k) => !planUnits.includes(k) || covered.has(k)), + slotSatisfied, + now: now.toISOString(), + newTitleWindowDays: d.ports.options.newTitleWindowDays, + upgradeCandidate: ev.decision === "upgrade", + }).mode; + if (!mode) { + await note(c, { + ...base, + decision: "skipped", + mode: null, + rejected: [], + reasons: ["mode:none", ...flat.reasons], + ...evalExtra, + }); + return done; + } + if (p.alreadyPlanned) { + await note(c, { + ...base, + decision: "skipped", + mode, + rejected: [], + reasons: [`covered_by_plan:${p.alreadyPlanned}`], + }); + return done; + } + const gate = await gates(c, rel, mode, sub); + if (!gate.ok) { + await note(c, { ...base, decision: "skipped", mode, rejected: gate.rejected, reasons: gate.reasons, ...evalExtra }); + c.out.skipped.push({ sourceReleaseId: rel.id, reason: gate.reasons[0] ?? "gated" }); + return done; + } + const r = await reserve(c, { + slot, + rel, + mode, + ev, + sub, + instanceId: gate.instanceId, + volumeKey: gate.volumeKey, + savePath: gate.savePath, + unitKeys: planUnits, + multiInflight: !isMovie, + }); + if (r.conflict) return defer("units_inflight", r.conflict, mode); + const planId = r.planId; + if (!planId) { + await note(c, { ...base, decision: "skipped", mode, rejected: [], reasons: ["slot_changed_or_reserved"] }); + return done; + } + c.out.plans.push(planId); + await note(c, { + ...base, + planId, + decision: ev.decision as "accept" | "upgrade", + mode, + rejected: [], + reasons: flat.reasons, + ...evalExtra, + unitKeys: planUnits, + }); + if (c.draft) return { planId, deferred: false }; + await enqueueTracked(d, { + kind: "submitDownload", + idempotencyKey: `plan:${planId}`, + input: { acquisitionPlanId: planId }, + subjectType: "acquisition_plan", + subjectId: planId, + }).then( + ({ taskId }) => db.updateTable("acquisition_plan").set({ task_id: taskId }).where("id", "=", planId).execute(), + (err) => + d.ctx.logger.warn({ err: { message: (err as Error).message }, planId }, "submitDownload 入队失败,留给巡检补投"), + ); + return { planId, deferred: false }; +} + +/** 分集质量账本:合成单元键 → 该单元当前版本的交付特征(含入库时的阶梯 pri) */ +async function unitQuality( + exec: Exec, + slot: AcquisitionSlotRow, + keys: readonly string[], +): Promise> { + const rows = await exec + .selectFrom("slot_unit as su") + .innerJoin("content_unit as u", "u.id", "su.content_unit_id") + .select(["u.season_number", "u.episode_number", "su.quality", "su.pri"]) + .where("su.slot_id", "=", slot.id) + .where("su.media_asset_id", "is not", null) + .execute(); + const out: Record = {}; + const want = new Set(keys); + for (const r of rows) { + if (r.episode_number === null) continue; + const k = episodeKey(r.season_number ?? 1, r.episode_number); + if (!want.has(k)) continue; + out[k] = { ...((r.quality ?? {}) as QualityFacts), ...(r.pri !== null ? { pri: r.pri } : {}) }; + } + return out; } interface GateOk { @@ -588,8 +813,12 @@ async function reserve( instanceId: string; volumeKey: string; savePath: string; + /** 计划预留的单元(剧集);电影为空 */ + unitKeys: string[]; + /** 剧集 Slot:允许与其它在途计划并存(单元不相交) */ + multiInflight: boolean; }, -): Promise { +): Promise<{ planId: string | null; conflict?: string[] }> { const { d, now } = c; const immediate = p.sub?.priority === "immediate"; const priority = PRIORITY[priorityOf(p.mode, immediate)]; @@ -599,6 +828,7 @@ async function reserve( instanceId: p.instanceId, savePath: p.savePath, mode: p.mode, + unitKeys: p.unitKeys, }; const evaluation = jsonb({ ...toContractSlotEvaluation(p.ev), @@ -607,6 +837,7 @@ async function reserve( volumeKey: p.volumeKey, savePath: p.savePath, }); + const purpose = p.mode === "upgrade" ? ("upgrade" as const) : ("fill" as const); if (c.draft) { // 草案:同一 Slot+发布复用未提交的草案;Slot 状态与预算不动,expected_revision 记下当前 Slot 版本 +1 const key = `draft:${p.slot.id}:${p.rel.id}`; @@ -616,20 +847,26 @@ async function reserve( .where("idempotency_key", "=", key) .executeTakeFirst(); if (hit) { - if (hit.status !== "draft") return null; + if (hit.status !== "draft") return { planId: null }; await d.ctx.db .updateTable("acquisition_plan") - .set({ evaluation, payload_hash: hashJson(payload), expected_revision: p.slot.revision + 1, priority }) + .set({ + evaluation, + payload_hash: hashJson(payload), + expected_revision: p.slot.revision + 1, + priority, + unit_keys: p.unitKeys, + }) .where("id", "=", hit.id) .execute(); - return hit.id; + return { planId: hit.id }; } const row = await d.ctx.db .insertInto("acquisition_plan") .values({ slot_id: p.slot.id, source_release_id: p.rel.id, - purpose: p.mode === "upgrade" ? "upgrade" : "fill", + purpose, status: "draft", priority, evaluation, @@ -642,63 +879,73 @@ async function reserve( payload_hash: hashJson(payload), expected_revision: p.slot.revision + 1, idempotency_key: key, + unit_keys: p.unitKeys, }) .onConflict((oc) => oc.column("idempotency_key").doNothing()) .returning("id") .executeTakeFirst(); - return row?.id ?? null; + return { planId: row?.id ?? null }; } - try { - return await d.ctx.db.transaction().execute(async (trx) => { - const slot = await trx - .selectFrom("acquisition_slot") - .selectAll() - .where("id", "=", p.slot.id) - .forUpdate() - .executeTakeFirst(); - if (!slot || slot.revision !== p.slot.revision || BUSY.includes(slot.status)) return null; - let st: SlotState = { status: slot.status, hasAsset: slot.current_asset_id !== null }; + return d.ctx.db.transaction().execute(async (trx) => { + const slot = await trx + .selectFrom("acquisition_slot") + .selectAll() + .where("id", "=", p.slot.id) + .forUpdate() + .executeTakeFirst(); + if (!slot) return { planId: null }; + const busy = BUSY.includes(slot.status); + if (p.multiInflight) { + // Slot×单元预留:在 Slot 行锁内复核与其它在途计划的单元不相交 + const others = await slotInflightPlans(trx, slot.id); + const hit = unitsOverlap( + p.unitKeys, + others.flatMap((o) => o.unitKeys), + ); + if (hit.length) return { planId: null, conflict: hit }; + } else if (slot.revision !== p.slot.revision || busy) { + return { planId: null }; + } + let st: SlotState = { status: slot.status, hasAsset: slot.current_asset_id !== null }; + if (!busy) { if (st.status === "satisfied") st = transition(st, { type: "upgrade_found" }); st = transition(st, { type: "reserve" }); - const plan = await trx - .insertInto("acquisition_plan") - .values({ - slot_id: slot.id, - source_release_id: p.rel.id, - purpose: p.mode === "upgrade" ? "upgrade" : "fill", - status: "reserved", - priority, - evaluation, - reserved_bytes: p.rel.size_bytes ?? 0, - budget_reservation: jsonb({ - bytes: p.rel.size_bytes, - countsAsNewTitle: p.mode === "new_title", - day: startOfDay(now).toISOString().slice(0, 10), - }), - payload_hash: hashJson(payload), - expected_revision: slot.revision + 1, - idempotency_key: `acq:${slot.id}:${p.rel.id}`, - }) - .onConflict((oc) => oc.column("idempotency_key").doNothing()) - .returning("id") - .executeTakeFirst(); - if (!plan) return null; - await trx - .updateTable("acquisition_slot") - .set({ - status: st.status, - reserved_plan_id: plan.id, - revision: slot.revision + 1, - last_evaluated_at: now, - status_reason: `reserved:${p.mode}`, - }) - .where("id", "=", slot.id) - .execute(); - return plan.id; - }); - } catch (err) { - // 部分唯一索引:同一 Slot 同时只能有一个有效预留 - if ((err as { code?: string }).code === "23505") return null; - throw err; - } + } + const plan = await trx + .insertInto("acquisition_plan") + .values({ + slot_id: slot.id, + source_release_id: p.rel.id, + purpose, + status: "reserved", + priority, + evaluation, + reserved_bytes: p.rel.size_bytes ?? 0, + budget_reservation: jsonb({ + bytes: p.rel.size_bytes, + countsAsNewTitle: p.mode === "new_title", + day: startOfDay(now).toISOString().slice(0, 10), + }), + payload_hash: hashJson(payload), + expected_revision: slot.revision + 1, + idempotency_key: `acq:${slot.id}:${p.rel.id}`, + unit_keys: p.unitKeys, + }) + .onConflict((oc) => oc.column("idempotency_key").doNothing()) + .returning("id") + .executeTakeFirst(); + if (!plan) return { planId: null }; + await trx + .updateTable("acquisition_slot") + .set({ + status: st.status, + ...(busy ? {} : { reserved_plan_id: plan.id }), + revision: slot.revision + 1, + last_evaluated_at: now, + status_reason: busy ? `reserved_more:${p.mode}` : `reserved:${p.mode}`, + }) + .where("id", "=", slot.id) + .execute(); + return { planId: plan.id }; + }); } diff --git a/apps/server/src/modules/acquisition-organize/services/downloads.ts b/apps/server/src/modules/acquisition-organize/services/downloads.ts index 4a9cbadd..6ac25fd4 100644 --- a/apps/server/src/modules/acquisition-organize/services/downloads.ts +++ b/apps/server/src/modules/acquisition-organize/services/downloads.ts @@ -1,10 +1,38 @@ -import { assertSafeToDeleteData, DownloaderError, type QbTorrent, SiteError, type TorrentSource } from "@umt/adapters"; +import { + assertSafeToDeleteData, + type DownloaderClient, + DownloaderError, + type QbTorrent, + SiteError, + type TorrentSource, +} from "@umt/adapters"; import type { DownloadStatus, PauseReason } from "@umt/contracts"; -import { type DB, type DownloadTaskRow, sql, type Transaction } from "@umt/db"; -import { mapQbState, type SlotEvent, transition } from "@umt/domain"; +import { + type AcquisitionPlanRow, + type DB, + type DownloadTaskRow, + type SourceReleaseRow, + sql, + type Transaction, +} from "@umt/db"; +import { + episodeKey, + fileKindOf, + mapQbState, + type PlanStage, + parseReleasePath, + parseUnitKey, + type SlotEvent, + type SlotState, + stageAfter, + transition, + transitionWithInflight, + unitsOverlap, +} from "@umt/domain"; import { mergeObservedDownloadState, toDomainDownloadStatus } from "@umt/mapping"; import { withAttemptToken } from "@umt/tasks"; import { errMessage, jsonb } from "../util.ts"; +import { parsedOf } from "./acquisition.ts"; import { type AoDeps, automationBlock, @@ -19,17 +47,69 @@ import { upsertDecision, } from "./common.ts"; import { hnrSatisfied, retentionElapsed } from "./retention.ts"; +import { + afterSlotVerified, + type SlotCompletion, + slotComplete, + slotCoveredUnits, + slotInflightPlans, +} from "./subscriptions.ts"; import { decideVolume, loadVolumes, usedPercent, type VolumeState } from "./volumes.ts"; // 下载:提交(幂等 + fencing)、qB 同步与标签接管、无进展证据、缓存卷硬阈值、缓存清理。 -/** Slot 状态推进(非法迁移只记日志不抛:并发下 Slot 可能已被人工处理) */ +/** 参与 Slot×单元多在途聚合的事件(其余事件按严格状态机) */ +const PLAN_EVENTS = new Set([ + "reserve", + "release", + "download_started", + "download_completed", + "download_failed", + "verify_passed", + "verify_failed", +]); + +/** 核验事件的来源计划与本次核验涉及的资产(organize 未传 planId 时按资产→下载任务→计划反查) */ +async function verifiedContext( + trx: Exec, + slotId: string, + assetId: string | null | undefined, +): Promise<{ planId: string | null; assets: string[] }> { + if (!assetId) return { planId: null, assets: [] }; + const a = await trx + .selectFrom("media_asset as a") + .leftJoin("download_task as dt", "dt.id", "a.download_task_id") + .leftJoin("acquisition_plan as p", "p.id", "dt.acquisition_plan_id") + .select(["dt.id as dt_id", "p.id as plan_id", "p.slot_id"]) + .where("a.id", "=", assetId) + .executeTakeFirst(); + const assets = new Set([assetId]); + if (a?.dt_id) { + const rows = await sql<{ media_asset_id: string }>` + select f.media_asset_id from organization_file_result f + join organization_record r on r.id = f.organization_record_id + where r.download_task_id = ${a.dt_id} and f.role = 'main' and f.media_asset_id is not null + and r.id = (select id from organization_record where download_task_id = ${a.dt_id} order by created_at desc limit 1)`.execute( + trx, + ); + for (const r of rows.rows) assets.add(r.media_asset_id); + } + return { planId: a?.slot_id === slotId ? (a?.plan_id ?? null) : null, assets: [...assets] }; +} + +/** + * Slot 状态推进(非法迁移只记日志不抛:并发下 Slot 可能已被人工处理)。 + * 剧集 Slot 可有多个在途计划(Slot×单元预留):按"其它在途计划的阶段 + 本计划事件后阶段"聚合, + * Slot 显示最靠后阶段;最后一个计划结束才落到静止态。 + * 核验通过时由本模块按季总集数账本重算完成(覆盖调用方传入的 complete),并写分集质量账本、 + * 完成订阅(写历史);计划结束(释放/失败/核验)时标记 Slot 到期重评,处理忙时延后的发布。 + */ export async function applySlotEvent( d: AoDeps, trx: Exec, slotId: string, event: SlotEvent, - extra: { currentAssetId?: string | null; reason?: string } = {}, + extra: { currentAssetId?: string | null; reason?: string; planId?: string | null } = {}, ): Promise { const slot = await trx .selectFrom("acquisition_slot") @@ -38,26 +118,63 @@ export async function applySlotEvent( .forUpdate() .executeTakeFirst(); if (!slot) return false; + let planId = extra.planId ?? null; + let completion: SlotCompletion | null = null; + let verified: string[] = []; + if (event.type === "verify_passed") { + const vc = await verifiedContext(trx, slotId, extra.currentAssetId); + planId ??= vc.planId; + verified = vc.assets; + const entity = await trx + .selectFrom("media_entity") + .select("kind") + .where("id", "=", slot.media_entity_id) + .executeTakeFirst(); + completion = await slotComplete(trx, slot, entity?.kind === "movie" ? "movie" : "tv", verified); + event = { type: "verify_passed", complete: completion.complete }; + } + let others: PlanStage[] = []; + if (PLAN_EVENTS.has(event.type)) { + const inflight = await slotInflightPlans(trx, slotId); + if (!planId && event.type === "verify_failed") { + // organize 阻塞/失败未传 planId:在途里处于核验阶段的那个视为本计划 + planId = inflight.find((p) => p.stage === "verifying")?.id ?? null; + } + others = inflight.filter((p) => p.id !== planId).map((p) => p.stage); + } + const state = { status: slot.status, hasAsset: slot.current_asset_id !== null }; + let next: SlotState | null; try { - const next = transition({ status: slot.status, hasAsset: slot.current_asset_id !== null }, event); - await trx - .updateTable("acquisition_slot") - .set({ - status: next.status, - revision: slot.revision + 1, - ...(extra.currentAssetId !== undefined ? { current_asset_id: extra.currentAssetId } : {}), - ...(next.status !== "reserved" && next.status !== "downloading" && next.status !== "verifying" - ? { reserved_plan_id: null } - : {}), - status_reason: extra.reason ?? event.type, - }) - .where("id", "=", slotId) - .execute(); - return true; + next = PLAN_EVENTS.has(event.type) ? transitionWithInflight(state, event, others) : transition(state, event); } catch (err) { d.ctx.logger.warn({ slotId, event: event.type, err: { message: errMessage(err) } }, "Slot 状态迁移被拒绝"); return false; } + if (!next) { + d.ctx.logger.warn({ slotId, event: event.type, from: slot.status }, "Slot 状态迁移被拒绝(已被人工处理?)"); + return false; + } + const ended = PLAN_EVENTS.has(event.type) && stageAfter(event.type) === null; + const busy = next.status === "reserved" || next.status === "downloading" || next.status === "verifying"; + await trx + .updateTable("acquisition_slot") + .set({ + status: next.status, + revision: slot.revision + 1, + ...(extra.currentAssetId !== undefined ? { current_asset_id: extra.currentAssetId } : {}), + ...(!busy ? { reserved_plan_id: null } : {}), + status_reason: + extra.reason === "verified" || extra.reason === "verified_partial" + ? completion?.complete + ? "verified" + : "verified_partial" + : (extra.reason ?? event.type), + ...(ended ? { next_evaluate_at: d.ports.now() } : {}), + }) + .where("id", "=", slotId) + .execute(); + if (completion) await afterSlotVerified(d, trx, { slotId, planId, verifiedAssets: verified, completion }); + return true; } async function emitStatus( @@ -212,7 +329,13 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?: .set({ status: "submitted", status_reason: r.kind }) .where("id", "=", plan.id) .execute(); - await applySlotEvent(d, trx, plan.slot_id, { type: "download_started" }, { reason: `download:${r.kind}` }); + await applySlotEvent( + d, + trx, + plan.slot_id, + { type: "download_started" }, + { reason: `download:${r.kind}`, planId: plan.id }, + ); await d.ctx.outbox.emit( trx, "download.submitted", @@ -224,6 +347,12 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?: ); if (result.kind === "rejected" || result.kind === "source_failed") return { status: "failed", reason: result.reason }; const hash = (result as Extract).hash; + // 季包/多集包:按缺集选文件(接管的已有任务不改文件选择);失败不影响提交结果 + if (result.kind === "added" && plan.unit_keys.length > 0 && plan.file_selection === null) { + await selectPackFiles(d, client, hash, plan, rel).catch((err) => + d.ctx.logger.warn({ planId: plan.id, err: { message: errMessage(err) } }, "季包选文件失败,按整包下载"), + ); + } if (!dtId) { // 账本已有结果(重放):按 hash 找回任务行 dtId = ( @@ -238,13 +367,107 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?: return { status: "submitted", reason: result.kind, ...(dtId ? { downloadTaskId: dtId } : {}) }; } +const PACK_FILE_LIST = { retries: 3, intervalMs: 300 } as const; + +/** + * 季包按缺集选文件(对照 MP 从整季包按集选文件,仅 qB):提交后读文件清单,把不需要单元的文件优先级置 0; + * 需要单元 = 计划预留的集(季通配 = 该季未覆盖、且不在其它在途计划里的集)。 + * 文件清单拿不到 / 有视频文件认不出集号 → 整包下载并记录理由;计划的 unit_keys 收敛为实际选中的集。 + */ +async function selectPackFiles( + d: AoDeps, + client: DownloaderClient, + hash: string, + plan: AcquisitionPlanRow, + rel: SourceReleaseRow, +): Promise { + const parsed = parsedOf(rel); + const isPack = !parsed.episodes && !parsed.airDate; + const multi = !!parsed.episodes && parsed.episodes.end > parsed.episodes.start; + if (!isPack && !multi) return; + const record = (sel: Record, unitKeys?: string[]) => + d.ctx.db + .updateTable("acquisition_plan") + .set({ file_selection: jsonb(sel), ...(unitKeys ? { unit_keys: unitKeys } : {}) }) + .where("id", "=", plan.id) + .execute(); + const files = await client.files(hash, PACK_FILE_LIST).catch(() => []); + if (files.length === 0) return void (await record({ mode: "whole", reason: "file_list_unavailable" })); + + // 季通配:展开为"该季未覆盖且不在其它在途计划里" + const wildSeasons = new Set( + plan.unit_keys + .map(parseUnitKey) + .filter((u) => u?.wildcard) + .map((u) => u!.season), + ); + const explicit = new Set(plan.unit_keys.filter((k) => !parseUnitKey(k)?.wildcard)); + let blocked = new Set(); + if (wildSeasons.size) { + const slot = await d.ctx.db + .selectFrom("acquisition_slot") + .select(["id", "current_asset_id", "media_entity_id"]) + .where("id", "=", plan.slot_id) + .executeTakeFirstOrThrow(); + const slots = await d.ctx.db + .selectFrom("acquisition_slot") + .select("id") + .where("media_entity_id", "=", slot.media_entity_id) + .execute(); + const covered = await slotCoveredUnits(d.ctx.db, slot, { entityWide: slots.length <= 1 }); + const others = await slotInflightPlans(d.ctx.db, plan.slot_id, plan.id); + blocked = new Set([...covered.keys, ...others.flatMap((o) => o.unitKeys)]); + } + const wantedKey = (k: string) => { + if (explicit.has(k)) return true; + const u = parseUnitKey(k); + return !!u && wildSeasons.has(u.season) && !blocked.has(k) && unitsOverlap([k], [...blocked]).length === 0; + }; + const rows = files.map((f) => { + const kind = fileKindOf(f.name); + const p = parseReleasePath(f.name); + const keys: string[] = []; + if (p.episodes) + for (let e = p.episodes.start; e <= p.episodes.end && keys.length < 200; e++) + keys.push(episodeKey(p.season ?? parsed.season ?? 1, e)); + return { index: f.index, name: f.name, kind, keys, wanted: keys.some(wantedKey) }; + }); + const videos = rows.filter((r) => r.kind === "video"); + const unmapped = videos.filter((r) => r.keys.length === 0); + if (videos.length === 0 || unmapped.length > 0) + return void (await record({ + mode: "whole", + reason: videos.length === 0 ? "no_video_files" : "unmapped_video_file", + files: unmapped.slice(0, 20).map((r) => r.name), + })); + const selected = [...new Set(videos.filter((r) => r.wanted).flatMap((r) => r.keys.filter(wantedKey)))]; + if (selected.length === 0) return void (await record({ mode: "whole", reason: "no_missing_units_in_pack" })); + // 视频不需要 → 不下;附属文件(字幕等)有集号且不需要 → 不下;无集号的附属文件保留 + const skip = rows.filter((r) => (r.kind === "video" ? !r.wanted : r.keys.length > 0 && !r.wanted)); + if (skip.length > 0) + await client.setFilePriority( + hash, + skip.map((r) => r.index), + 0, + ); + await record( + { + mode: skip.length > 0 ? "selected" : "all_needed", + selectedUnits: selected, + skippedFiles: skip.slice(0, 200).map((r) => r.name), + totalFiles: files.length, + }, + selected, + ); +} + async function failPlanTx(d: AoDeps, trx: Exec, planId: string, slotId: string, reason: string) { await trx .updateTable("acquisition_plan") .set({ status: "failed", status_reason: reason.slice(0, 500) }) .where("id", "=", planId) .execute(); - await applySlotEvent(d, trx, slotId, { type: "release" }, { reason: `submit_failed` }); + await applySlotEvent(d, trx, slotId, { type: "release" }, { reason: `submit_failed`, planId }); await upsertDecision(d, trx, { kind: "download_failed", title: "下载提交失败", @@ -256,6 +479,25 @@ async function failPlanTx(d: AoDeps, trx: Exec, planId: string, slotId: string, }); } +/** qB 里已确认没有该任务(用户移除):在途计划释放(失败),其单元不再算"已预留",Slot 到期重评 */ +async function releaseRemovedPlan(d: AoDeps, trx: Exec, planId: string) { + const plan = await trx + .updateTable("acquisition_plan") + .set({ status: "failed", status_reason: "download_removed" }) + .where("id", "=", planId) + .where("status", "=", "submitted") + .returning(["id", "slot_id"]) + .executeTakeFirst(); + if (plan) + await applySlotEvent( + d, + trx, + plan.slot_id, + { type: "download_failed", reason: "download_removed" }, + { reason: "download_removed", planId: plan.id }, + ); +} + async function failPlan(d: AoDeps, planId: string, slotId: string, reason: string): Promise { await d.ctx.db.transaction().execute((trx) => failPlanTx(d, trx, planId, slotId, reason)); return { status: "failed", reason }; @@ -407,6 +649,7 @@ async function syncInstance(d: AoDeps, topo: InstanceTopology, stats: SyncStats) .where("id", "=", row.id) .execute(); await emitStatus(d, trx, row.id, row.status, "removed", null); + if (row.acquisition_plan_id) await releaseRemovedPlan(d, trx, row.acquisition_plan_id); stats.removed++; } }); @@ -486,11 +729,11 @@ async function observe(d: AoDeps, row: DownloadTaskRow, t: QbTorrent, now: Date, const plan = row.acquisition_plan_id ? await trx .selectFrom("acquisition_plan") - .select(["slot_id"]) + .select(["id", "slot_id"]) .where("id", "=", row.acquisition_plan_id) .executeTakeFirst() : null; - if (plan) await applySlotEvent(d, trx, plan.slot_id, { type: "download_completed" }); + if (plan) await applySlotEvent(d, trx, plan.slot_id, { type: "download_completed" }, { planId: plan.id }); } if (stalledNow) { // 无进展:只记证据 + 待处理,保留原任务,不换源(REQ §8.4.2) diff --git a/apps/server/src/modules/acquisition-organize/services/subscriptions.ts b/apps/server/src/modules/acquisition-organize/services/subscriptions.ts new file mode 100644 index 00000000..8342ee99 --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/services/subscriptions.ts @@ -0,0 +1,800 @@ +import type { AcquisitionSlot, SubscriptionParams, SubscriptionProgress, UnitProgressState } from "@umt/contracts"; +import type { AcquisitionSlotRow, SubscriptionRow } from "@umt/db"; +import { sql } from "@umt/db"; +import { + episodeKey, + movieCompletion, + type ParsedRelease, + parseUnitKey, + priorityLadder, + type QualityFacts, + resolveEpisodeTotal, + type SeasonProgress, + seasonProgress, + type UnitRange, +} from "@umt/domain"; +import { slotSpecFromRow } from "@umt/mapping"; +import { errMessage, jsonb, toApi } from "../util.ts"; +import { type AoDeps, type Exec, upsertDecision } from "./common.ts"; + +// W5-C 订阅业务:季总集数账本、缺集/在途、完成判定与历史、Slot 重评调度、分集质量账本、主动搜索调度。 +// 对照 docs/porting/moviepilot-subscribe.md §3/§7/§8/§9;差距清单 docs/plan/05-subscription-gap.md。 + +/** 在途计划状态(等价 MP note:提交成功即计入,失败/移除释放) */ +export const INFLIGHT_PLAN_STATUSES = ["reserved", "submitted"] as const; +/** 仍参与评估/搜索的订阅状态(pending = TMDB 未就绪:可搜索/评估,不可完成) */ +export const TRACKING_SUB_STATUSES = ["active", "pending"] as const; + +export const DEFAULT_SLOT_REEVALUATE_MS = 6 * 3600_000; +export const DEFAULT_METADATA_REFRESH_MS = 6 * 3600_000; + +export function subParams(sub: Pick | null | undefined): SubscriptionParams { + return (sub?.params ?? {}) as SubscriptionParams; +} + +export function rangeOf(sub: Pick | null | undefined): UnitRange | undefined { + const r = subParams(sub).episodeRange; + if (!r || (r.start === undefined && r.end === undefined)) return undefined; + return { ...(r.start !== undefined ? { start: r.start } : {}), ...(r.end !== undefined ? { end: r.end } : {}) }; +} + +// ── 季总集数账本 ── + +export interface SeasonLedger { + seasonUnitId: string | null; + seasonNumber: number; + total: number | null; + known: boolean; + source: string | null; + locked: boolean; +} + +/** 作品各季的总集数账本(季单元 episode_total);只列季号 ≥ 1 的正片季 */ +export async function seasonLedgers(exec: Exec, entityId: string): Promise> { + const rows = await exec + .selectFrom("content_unit") + .select(["id", "season_number", "episode_total", "episode_total_source", "episode_total_locked"]) + .where("media_entity_id", "=", entityId) + .where("kind", "=", "season") + .where("status", "!=", "retired") + .where("season_number", "is not", null) + .orderBy("created_at") + .execute(); + const out = new Map(); + for (const r of rows) { + const n = r.season_number!; + if (n < 1 || out.has(n)) continue; + out.set(n, { + seasonUnitId: r.id, + seasonNumber: n, + total: r.episode_total, + known: r.episode_total !== null && r.episode_total > 0, + source: r.episode_total_source, + locked: r.episode_total_locked, + }); + } + return out; +} + +/** 找或建季单元(TMDB 季信息/人工设置总集数时用;编号来自元数据,状态 confirmed) */ +export async function ensureSeasonUnit(exec: Exec, entityId: string, season: number): Promise { + const hit = await exec + .selectFrom("content_unit") + .select("id") + .where("media_entity_id", "=", entityId) + .where("kind", "=", "season") + .where("season_number", "=", season) + .where("status", "!=", "retired") + .orderBy("created_at") + .executeTakeFirst(); + if (hit) return hit.id; + return ( + await exec + .insertInto("content_unit") + .values({ media_entity_id: entityId, kind: "season", season_number: season, status: "confirmed" }) + .returning("id") + .executeTakeFirstOrThrow() + ).id; +} + +/** 某季已入库覆盖的最高集号(回退兜底用) */ +async function coveredMaxEpisode(exec: Exec, entityId: string, season: number): Promise { + const r = await sql<{ max: number | null }>` + select max(u.episode_number)::int as max from asset_unit_coverage c + join content_unit u on u.id = c.content_unit_id + join media_asset a on a.id = c.media_asset_id + where u.media_entity_id = ${entityId} and u.kind = 'episode' and u.season_number = ${season} + and a.status = 'active'`.execute(exec); + return r.rows[0]?.max ?? null; +} + +/** + * 写入季总集数(元数据刷新 / 注入 / 人工):按 resolveEpisodeTotal 只升、回退不低于已覆盖最高集、锁定不改。 + * manual=true 时为人工设置:可调低并可锁定。 + */ +export async function applyEpisodeTotal( + exec: Exec, + input: { + entityId: string; + season: number; + fresh?: number | null; + injected?: number | null; + manual?: { total: number; locked: boolean }; + now: Date; + }, +): Promise<{ seasonUnitId: string; total: number | null; known: boolean; changed: boolean }> { + const unitId = await ensureSeasonUnit(exec, input.entityId, input.season); + const cur = await exec + .selectFrom("content_unit") + .select(["episode_total", "episode_total_locked", "episode_total_source"]) + .where("id", "=", unitId) + .forUpdate() + .executeTakeFirstOrThrow(); + if (input.manual) { + await exec + .updateTable("content_unit") + .set({ + episode_total: input.manual.total, + episode_total_source: "manual", + episode_total_locked: input.manual.locked, + episode_total_at: input.now, + }) + .where("id", "=", unitId) + .execute(); + return { + seasonUnitId: unitId, + total: input.manual.total, + known: true, + changed: input.manual.total !== cur.episode_total, + }; + } + const r = resolveEpisodeTotal({ + previous: cur.episode_total, + fresh: input.fresh ?? null, + injected: input.injected ?? null, + coveredMax: await coveredMaxEpisode(exec, input.entityId, input.season), + locked: cur.episode_total_locked, + }); + if (r.changed) { + await exec + .updateTable("content_unit") + .set({ episode_total: r.total, episode_total_source: r.source, episode_total_at: input.now }) + .where("id", "=", unitId) + .execute(); + } + return { seasonUnitId: unitId, total: r.total, known: r.known, changed: r.changed }; +} + +// ── 覆盖 / 在途 ── + +/** Slot 已覆盖的内容单元:该 Slot 计划产出资产 + 当前资产 + 分集账本(content_unit id 与合成键两种口径) */ +export async function slotCoveredUnits( + exec: Exec, + slot: Pick, + opts: { entityWide?: boolean } = {}, +): Promise<{ keys: Set; episodes: Map> }> { + const r = await sql<{ + unit_id: string; + season_number: number | null; + episode_number: number | null; + kind: string; + air_date: string | null; + }>` + with assets as ( + select a.id from media_asset a + left join download_task dt on dt.id = a.download_task_id + left join acquisition_plan p on p.id = dt.acquisition_plan_id + where a.status = 'active' and a.parent_asset_id is null + and (p.slot_id = ${slot.id} or a.id = ${slot.current_asset_id} + or a.id in (select media_asset_id from slot_unit where slot_id = ${slot.id}) + ${opts.entityWide ? sql`or a.media_entity_id = ${slot.media_entity_id}` : sql``}) + ) + select u.id as unit_id, u.season_number, u.episode_number, u.kind, to_char(u.air_date, 'YYYYMMDD') as air_date + from asset_unit_coverage c + join assets on assets.id = c.media_asset_id + join content_unit u on u.id = c.content_unit_id + union + select u.id, u.season_number, u.episode_number, u.kind, to_char(u.air_date, 'YYYYMMDD') from slot_unit su + join content_unit u on u.id = su.content_unit_id + where su.slot_id = ${slot.id} and su.media_asset_id is not null`.execute(exec); + const keys = new Set(); + const episodes = new Map>(); + for (const row of r.rows) { + keys.add(row.unit_id); + // 综艺按播出日期成单元:与发布侧合成键 d{YYYYMMDD} 同口径 + if (row.air_date) keys.add(`d${row.air_date}`); + if (row.kind === "episode" && row.episode_number !== null) { + const s = row.season_number ?? 1; + keys.add(episodeKey(s, row.episode_number)); + const set = episodes.get(s) ?? new Set(); + set.add(row.episode_number); + episodes.set(s, set); + } + } + // 入库时记在资产上的合成键(无 content_unit 的旧数据) + const legacy = await sql<{ keys: string[] | null }>` + select array(select jsonb_array_elements_text(coalesce(a.media_info->'_umt'->'unitKeys', '[]'::jsonb))) as keys + from media_asset a + left join download_task dt on dt.id = a.download_task_id + left join acquisition_plan p on p.id = dt.acquisition_plan_id + where a.status = 'active' and a.parent_asset_id is null + and (p.slot_id = ${slot.id} or a.id = ${slot.current_asset_id})`.execute(exec); + for (const row of legacy.rows) + for (const k of row.keys ?? []) { + keys.add(k); + const p = parseUnitKey(k); + if (p?.episode !== undefined) { + const set = episodes.get(p.season) ?? new Set(); + set.add(p.episode); + episodes.set(p.season, set); + } + } + return { keys, episodes }; +} + +export interface InflightPlan { + id: string; + status: string; + unitKeys: string[]; + stage: "reserved" | "downloading" | "verifying"; +} + +/** Slot 的在途计划(reserved/submitted)及其阶段 */ +export async function slotInflightPlans(exec: Exec, slotId: string, excludePlanId?: string): Promise { + let q = exec + .selectFrom("acquisition_plan as p") + .leftJoin("download_task as dt", (j) => + j.onRef("dt.acquisition_plan_id", "=", "p.id").on("dt.removed_at", "is", null), + ) + .select(["p.id", "p.status", "p.unit_keys", "dt.status as dt_status"]) + .where("p.slot_id", "=", slotId) + .where("p.status", "in", [...INFLIGHT_PLAN_STATUSES]); + if (excludePlanId) q = q.where("p.id", "!=", excludePlanId); + const rows = await q.execute(); + const byId = new Map(); + for (const r of rows) { + const stage: InflightPlan["stage"] = + r.status === "reserved" + ? "reserved" + : r.dt_status === "completed" || r.dt_status === "importing" + ? "verifying" + : "downloading"; + const prev = byId.get(r.id); + if (!prev || stage === "verifying") byId.set(r.id, { id: r.id, status: r.status, unitKeys: r.unit_keys, stage }); + } + return [...byId.values()]; +} + +// ── 完成判定 ── + +export interface SlotCompletion { + complete: boolean; + seasons: SeasonProgress[]; + /** 继续追踪时的疑似缺集(总集数未知,已见最高集以下的空洞) */ + suspectedMissing: { season: number; episodes: number[] }[]; +} + +/** 剧集 Slot 覆盖的季:Slot 限定季单元 > 订阅季单元 > 作品已知的所有正片季 */ +async function scopeSeasons( + exec: Exec, + slot: Pick, + sub: Pick | null, + covered: Map>, + ledgers: Map, +): Promise { + const unitId = slot.content_unit_id ?? sub?.season_unit_id ?? null; + if (unitId) { + const u = await exec + .selectFrom("content_unit") + .select(["season_number", "kind"]) + .where("id", "=", unitId) + .executeTakeFirst(); + if (u?.season_number !== null && u?.season_number !== undefined) return [u.season_number]; + } + const eps = await exec + .selectFrom("content_unit") + .select("season_number") + .distinct() + .where("media_entity_id", "=", slot.media_entity_id) + .where("kind", "=", "episode") + .where("content_type", "=", "main") + .where("status", "!=", "retired") + .where("season_number", ">=", 1) + .execute(); + const set = new Set([...ledgers.keys(), ...covered.keys(), ...eps.map((e) => e.season_number!)]); + return [...set].filter((s) => s >= 1).sort((a, b) => a - b); +} + +/** + * Slot 完成判定。电影按内容版本;剧集/综艺按季单元总集数(TMDB 账本)与实际文件覆盖: + * 总集数未知 → 不完成、继续追踪并报告疑似缺集;订阅 pending(TMDB 未就绪)→ 不完成。 + */ +export async function slotComplete( + exec: Exec, + slot: Pick< + AcquisitionSlotRow, + "id" | "media_entity_id" | "content_unit_id" | "completion" | "subscription_id" | "current_asset_id" + >, + mediaType: "movie" | "tv", + newAssets: readonly string[], +): Promise { + const completion = (slot.completion ?? {}) as { kind?: string; cuts?: string[] }; + if (mediaType === "movie") { + const assets = newAssets.length + ? await exec + .selectFrom("media_asset") + .select(["id", "content_edition"]) + .where("id", "in", [...newAssets]) + .execute() + : []; + const complete = movieCompletion( + completion.cuts as never, + assets.map((a) => ({ + assetId: a.id, + verified: true, + ...(a.content_edition ? { cut: a.content_edition as never } : {}), + })), + ).complete; + return { complete, seasons: [], suspectedMissing: [] }; + } + const sub = slot.subscription_id + ? ((await exec.selectFrom("subscription").selectAll().where("id", "=", slot.subscription_id).executeTakeFirst()) ?? + null) + : null; + const slotCount = await exec + .selectFrom("acquisition_slot") + .select((eb) => eb.fn.countAll().as("n")) + .where("media_entity_id", "=", slot.media_entity_id) + .executeTakeFirstOrThrow(); + // 单 Slot 作品:作品下所有资产都算该 Slot 的覆盖(手动整理/历史导入);多 Slot 只算本 Slot 的 + const covered = await slotCoveredUnits(exec, slot, { entityWide: Number(slotCount.n) <= 1 }); + const inflight = await slotInflightPlans(exec, slot.id); + const inflightEps = new Map(); + for (const p of inflight) + for (const k of p.unitKeys) { + const u = parseUnitKey(k); + if (u?.episode !== undefined) inflightEps.set(u.season, [...(inflightEps.get(u.season) ?? []), u.episode]); + } + const ledgers = await seasonLedgers(exec, slot.media_entity_id); + const seasons = await scopeSeasons(exec, slot, sub, covered.episodes, ledgers); + const range = rangeOf(sub); + const progress = seasons.map((s) => { + const l = ledgers.get(s); + return seasonProgress({ + season: s, + total: l?.total ?? null, + totalKnown: !!l?.known, + covered: [...(covered.episodes.get(s) ?? [])], + inflight: inflightEps.get(s) ?? [], + ...(range ? { range } : {}), + }); + }); + const complete = sub?.status !== "pending" && progress.length > 0 && progress.every((p) => p.complete); + return { + complete, + seasons: progress, + suspectedMissing: progress + .filter((p) => p.suspectedMissing.length > 0) + .map((p) => ({ season: p.season, episodes: p.suspectedMissing })), + }; +} + +// ── 核验通过后:分集账本、订阅完成、重评调度 ── + +const qualityOf = (mediaInfo: unknown): QualityFacts => + ((mediaInfo as { _umt?: { quality?: QualityFacts } })?._umt?.quality ?? {}) as QualityFacts; + +/** + * 核验入库后(organize.verifyAndImport 在同一事务里调用): + * 1) 分集质量账本:该计划覆盖的单元指向新资产(升级计划只替换计划内单元;其余单元缺账本时补登); + * 2) 订阅完成:once 订阅完成;continuous + 仅本季 的剧集订阅季完整也完成;pending 不完成;写历史; + * 3) 疑似缺集记待处理(info);4) Slot 标记到期重评(处理忙时延后的发布)。 + */ +export async function afterSlotVerified( + d: AoDeps, + trx: Exec, + input: { slotId: string; planId: string | null; verifiedAssets: readonly string[]; completion: SlotCompletion }, +): Promise { + const slot = await trx.selectFrom("acquisition_slot").selectAll().where("id", "=", input.slotId).executeTakeFirst(); + if (!slot) return; + const now = d.ports.now(); + const plan = input.planId + ? await trx + .selectFrom("acquisition_plan") + .select(["id", "purpose", "unit_keys"]) + .where("id", "=", input.planId) + .executeTakeFirst() + : undefined; + if (input.verifiedAssets.length) { + const spec = slotSpecFromRow(toApi(slot)); + const cov = await sql<{ + asset_id: string; + unit_id: string; + media_info: unknown; + season: number | null; + ep: number | null; + }>` + select c.media_asset_id as asset_id, c.content_unit_id as unit_id, a.media_info, + u.season_number as season, u.episode_number as ep + from asset_unit_coverage c + join media_asset a on a.id = c.media_asset_id + join content_unit u on u.id = c.content_unit_id + where c.media_asset_id in (${sql.join([...input.verifiedAssets])}) and u.kind = 'episode'`.execute(trx); + const planUnits = new Set(plan?.unit_keys ?? []); + for (const r of cov.rows) { + const q = qualityOf(r.media_info); + const pri = priorityLadder(spec, { raw: "", hdr: [], isEncode: false, ...q } as unknown as ParsedRelease).pri; + const inPlan = planUnits.has(r.unit_id) || (r.ep !== null && planUnits.has(episodeKey(r.season ?? 1, r.ep))); + const replace = plan?.purpose === "upgrade" ? inPlan : true; + const values = { + slot_id: slot.id, + content_unit_id: r.unit_id, + media_asset_id: r.asset_id, + plan_id: plan?.id ?? null, + quality: jsonb(q), + pri, + }; + if (replace) { + await trx + .insertInto("slot_unit") + .values(values) + .onConflict((oc) => + oc.columns(["slot_id", "content_unit_id"]).doUpdateSet({ + media_asset_id: values.media_asset_id, + plan_id: values.plan_id, + quality: values.quality, + pri: values.pri, + }), + ) + .execute(); + } else { + await trx + .insertInto("slot_unit") + .values(values) + .onConflict((oc) => oc.columns(["slot_id", "content_unit_id"]).doNothing()) + .execute(); + } + } + } + + if (input.completion.suspectedMissing.length) { + const desc = input.completion.suspectedMissing + .map((s) => `S${String(s.season).padStart(2, "0")}:${s.episodes.map((e) => `E${e}`).join(",")}`) + .join(";"); + await upsertDecision(d, trx, { + kind: "other", + severity: "info", + title: "疑似缺集(总集数未知,继续追踪)", + summary: `已入库的最高集以下仍缺:${desc}`.slice(0, 500), + subjectType: "acquisition_slot", + subjectId: slot.id, + dedupeKey: `suspected_missing:${slot.id}`, + evidence: { slotId: slot.id, suspectedMissing: input.completion.suspectedMissing }, + }); + } + + if (input.completion.complete && slot.subscription_id) { + const sub = await trx + .selectFrom("subscription") + .selectAll() + .where("id", "=", slot.subscription_id) + .executeTakeFirst(); + const seasonDone = sub?.media_kind !== "movie" && sub?.season_scope === "current_season"; + if (sub && sub.status === "active" && (sub.mode === "once" || seasonDone)) { + await completeSubscription(d, trx, sub, "slot_satisfied", { seasons: input.completion.seasons }); + } + } + // 计划结束:Slot 到期重评(忙时延后的发布、连续更新的下一集) + await markSlotDue(trx, slot.id, now); +} + +/** Slot 标记立即到期重评(事务内写,由编排循环 sweepDueSlots 消费) */ +export async function markSlotDue(exec: Exec, slotId: string, now: Date): Promise { + await exec.updateTable("acquisition_slot").set({ next_evaluate_at: now }).where("id", "=", slotId).execute(); +} + +/** 订阅完成:状态 completed + 写 subscription_history(行保留,可改回 active 重新订阅) */ +export async function completeSubscription( + d: AoDeps, + trx: Exec, + sub: SubscriptionRow, + reason: string, + progress: Record = {}, +): Promise { + const done = await trx + .updateTable("subscription") + .set({ status: "completed", status_reason: reason, next_search_at: null }) + .where("id", "=", sub.id) + .where("status", "in", [...TRACKING_SUB_STATUSES]) + .returningAll() + .executeTakeFirst(); + if (!done) return false; + await trx + .insertInto("subscription_history") + .values({ + subscription_id: done.id, + media_entity_id: done.media_entity_id, + season_unit_id: done.season_unit_id, + media_kind: done.media_kind, + mode: done.mode, + reason, + snapshot: jsonb(toApi(done)), + progress: jsonb(progress), + completed_at: d.ports.now(), + }) + .execute(); + d.ctx.logger.info({ subscriptionId: sub.id, reason }, "订阅完成,已写历史"); + return true; +} + +// ── 延后与重评调度 ── + +export async function recordDeferral( + exec: Exec, + input: { slotId: string; sourceReleaseId: string; reason: string; unitKeys: readonly string[] }, +): Promise { + await sql` + insert into slot_deferral (slot_id, source_release_id, reason, unit_keys) + values (${input.slotId}, ${input.sourceReleaseId}, ${input.reason}, ${[...input.unitKeys]}) + on conflict (slot_id, source_release_id) do update + set reason = excluded.reason, unit_keys = excluded.unit_keys, attempts = slot_deferral.attempts + 1, + updated_at = now()`.execute(exec); +} + +export async function clearDeferral(exec: Exec, slotId: string, sourceReleaseId: string): Promise { + await exec + .deleteFrom("slot_deferral") + .where("slot_id", "=", slotId) + .where("source_release_id", "=", sourceReleaseId) + .execute(); +} + +/** + * 到期 Slot:next_evaluate_at <= now → 入队 evaluateAcquisition{acquisitionSlotId}(先评 deferred,再评最近发布); + * 仍需追踪的 Slot 顺延一个周期,否则清空。返回入队数。 + */ +export async function sweepDueSlots(d: AoDeps): Promise { + const now = d.ports.now(); + const every = d.ports.options.slotReevaluateIntervalMs ?? DEFAULT_SLOT_REEVALUATE_MS; + const due = await d.ctx.db.transaction().execute(async (trx) => { + const rows = await sql<{ id: string; status: string; sub_status: string | null; has_sub: boolean; due: Date }>` + select s.id, s.status, sub.status as sub_status, s.subscription_id is not null as has_sub, s.next_evaluate_at as due + from acquisition_slot s left join subscription sub on sub.id = s.subscription_id + where s.next_evaluate_at is not null and s.next_evaluate_at <= ${now} + order by s.next_evaluate_at limit 20 + for update of s skip locked`.execute(trx); + for (const r of rows.rows) { + const tracking = + !r.has_sub || (r.sub_status !== null && (TRACKING_SUB_STATUSES as readonly string[]).includes(r.sub_status)); + const keep = tracking && every > 0 && r.status !== "satisfied"; + await trx + .updateTable("acquisition_slot") + .set({ next_evaluate_at: keep ? new Date(now.getTime() + every) : null }) + .where("id", "=", r.id) + .execute(); + } + return rows.rows; + }); + let n = 0; + for (const r of due) { + try { + await d.ctx.engine.enqueue( + "evaluateAcquisition", + { acquisitionSlotId: r.id }, + { idempotencyKey: `slot-due:${r.id}:${new Date(r.due).getTime()}`, traceId: r.id }, + ); + n++; + } catch (err) { + d.ctx.logger.warn({ slotId: r.id, err: { message: errMessage(err) } }, "Slot 重评入队失败"); + } + } + return n; +} + +/** 首次进入追踪的 Slot 排上周期重评(只在未排期时设置,不覆盖"立即到期"标记) */ +export async function scheduleSlotReevaluate(d: AoDeps, exec: Exec, slotId: string): Promise { + const every = d.ports.options.slotReevaluateIntervalMs ?? DEFAULT_SLOT_REEVALUATE_MS; + if (every <= 0) return; + await exec + .updateTable("acquisition_slot") + .set({ next_evaluate_at: new Date(d.ports.now().getTime() + every) }) + .where("id", "=", slotId) + .where("next_evaluate_at", "is", null) + .where("status", "!=", "satisfied") + .execute(); +} + +/** 手动/新建订阅的主动搜索入队(searchSubscription 由 sources-identity 注册) */ +export async function enqueueSubscriptionSearch( + d: AoDeps, + subscriptionId: string, + reason: "created" | "manual" | "schedule", +): Promise<{ runId: string; deduplicated: boolean }> { + const now = d.ports.now(); + const key = + reason === "schedule" + ? `searchSubscription:${subscriptionId}:${Math.floor(now.getTime() / 3600_000)}` + : `searchSubscription:${subscriptionId}:${reason}:${now.getTime()}`; + return d.ctx.engine.enqueue( + "searchSubscription", + { subscriptionId, manual: reason === "manual" }, + { idempotencyKey: key, traceId: subscriptionId }, + ); +} + +/** 周期主动搜索:next_search_at 到期的追踪中订阅(新建时由搜索任务排期;0 = 关闭周期搜索) */ +export async function sweepSubscriptionSearch(d: AoDeps): Promise { + const now = d.ports.now(); + const rows = await d.ctx.db.transaction().execute(async (trx) => { + const due = await sql<{ id: string }>` + select id from subscription + where status in ('active', 'pending') and next_search_at is not null and next_search_at <= ${now} + order by next_search_at limit 10 for update skip locked`.execute(trx); + for (const r of due.rows) { + // 先顺延一小时租约:任务成功后会按周期重写 + await trx + .updateTable("subscription") + .set({ next_search_at: new Date(now.getTime() + 3600_000) }) + .where("id", "=", r.id) + .execute(); + } + return due.rows; + }); + let n = 0; + for (const r of rows) { + try { + await enqueueSubscriptionSearch(d, r.id, "schedule"); + n++; + } catch (err) { + d.ctx.logger.warn({ subscriptionId: r.id, err: { message: errMessage(err) } }, "订阅搜索入队失败"); + } + } + return n; +} + +/** 追踪中剧集的元数据刷新(总集数):上次刷新超过周期(或从未刷新)且有 TMDB 身份 → 入队 refreshMetadata */ +export async function sweepMetadataRefresh(d: AoDeps): Promise { + const every = d.ports.options.metadataRefreshIntervalMs ?? DEFAULT_METADATA_REFRESH_MS; + if (every <= 0) return 0; + const now = d.ports.now(); + const before = new Date(now.getTime() - every); + const rows = await sql<{ id: string }>` + select distinct e.id from subscription s + join media_entity e on e.id = s.media_entity_id + where s.status in ('active', 'pending') and s.media_kind <> 'movie' and e.status <> 'archived' + and (e.last_metadata_refresh_at is null or e.last_metadata_refresh_at < ${before}) + and exists (select 1 from external_identity x where x.media_entity_id = e.id and x.provider = 'tmdb' and x.status = 'accepted') + limit 20`.execute(d.ctx.db); + const slot = Math.floor(now.getTime() / every); + let n = 0; + for (const r of rows.rows) { + try { + await d.ctx.engine.enqueue( + "refreshMetadata", + { mediaEntityId: r.id }, + { idempotencyKey: `refreshMetadata:tracking:${r.id}:${slot}`, traceId: r.id }, + ); + n++; + } catch (err) { + d.ctx.logger.warn({ mediaEntityId: r.id, err: { message: errMessage(err) } }, "元数据刷新入队失败"); + } + } + return n; +} + +// ── 订阅进度(API / UI) ── + +export async function subscriptionProgress(exec: Exec, subId: string): Promise { + const sub = await exec.selectFrom("subscription").selectAll().where("id", "=", subId).executeTakeFirst(); + if (!sub) return null; + const slots = sub.media_entity_id + ? await exec + .selectFrom("acquisition_slot") + .selectAll() + .where("media_entity_id", "=", sub.media_entity_id) + .where((eb) => eb.or([eb("subscription_id", "=", sub.id), eb("subscription_id", "is", null)])) + .orderBy("created_at") + .execute() + : []; + const slotOut: SubscriptionProgress["slots"] = []; + for (const s of slots) { + const inflight = await slotInflightPlans(exec, s.id); + const deferred = await exec + .selectFrom("slot_deferral") + .select((eb) => eb.fn.countAll().as("n")) + .where("slot_id", "=", s.id) + .executeTakeFirstOrThrow(); + slotOut.push({ + slotId: s.id, + name: s.name, + status: s.status, + inflightPlans: inflight.map((p) => ({ planId: p.id, status: p.status, unitKeys: p.unitKeys })), + deferred: Number(deferred.n), + nextEvaluateAt: s.next_evaluate_at ? s.next_evaluate_at.toISOString() : null, + }); + } + const seasons: SubscriptionProgress["seasons"] = []; + if (sub.media_entity_id && sub.media_kind !== "movie") { + const main = slots[0]; + const ledgers = await seasonLedgers(exec, sub.media_entity_id); + const covered = main + ? await slotCoveredUnits(exec, main, { entityWide: slots.length <= 1 }) + : { keys: new Set(), episodes: new Map>() }; + const inflightEps = new Map>(); + for (const s of slotOut) + for (const p of s.inflightPlans) + for (const k of p.unitKeys) { + const u = parseUnitKey(k); + if (u?.episode !== undefined) + inflightEps.set(u.season, (inflightEps.get(u.season) ?? new Set()).add(u.episode)); + } + const nums = main + ? await scopeSeasons(exec, main, sub, covered.episodes, ledgers) + : [...ledgers.keys()].sort((a, b) => a - b); + const units = await exec + .selectFrom("content_unit") + .select(["id", "season_number", "episode_number"]) + .where("media_entity_id", "=", sub.media_entity_id) + .where("kind", "=", "episode") + .where("status", "!=", "retired") + .execute(); + const ledgerRows = main + ? await exec + .selectFrom("slot_unit") + .select(["content_unit_id", "quality", "pri"]) + .where("slot_id", "=", main.id) + .execute() + : []; + const range = rangeOf(sub); + for (const n of nums) { + const l = ledgers.get(n); + const p = seasonProgress({ + season: n, + total: l?.total ?? null, + totalKnown: !!l?.known, + covered: [...(covered.episodes.get(n) ?? [])], + inflight: [...(inflightEps.get(n) ?? [])], + ...(range ? { range } : {}), + }); + const maxEp = Math.max(p.total ?? 0, ...p.covered, ...p.inflight, ...p.suspectedMissing, 0); + const list: SubscriptionProgress["seasons"][number]["units"] = []; + for (let e = 1; e <= maxEp && e <= 2000; e++) { + const cu = units.find((u) => (u.season_number ?? 1) === n && u.episode_number === e); + const lr = cu ? ledgerRows.find((r) => r.content_unit_id === cu.id) : undefined; + const q = (lr?.quality ?? null) as { resolution?: string; source?: string } | null; + const state: UnitProgressState = + (range?.start !== undefined && e < range.start) || (range?.end !== undefined && e > range.end) + ? "out_of_range" + : p.covered.includes(e) + ? "covered" + : p.inflight.includes(e) + ? "inflight" + : p.suspectedMissing.includes(e) + ? "suspected_missing" + : "missing"; + list.push({ + episode: e, + state, + contentUnitId: cu?.id ?? null, + quality: lr ? { resolution: q?.resolution ?? null, source: q?.source ?? null, pri: lr.pri } : null, + }); + } + seasons.push({ + seasonUnitId: l?.seasonUnitId ?? null, + seasonNumber: n, + episodeTotal: l?.total ?? null, + totalKnown: !!l?.known, + totalSource: l?.source ?? null, + totalLocked: !!l?.locked, + units: list, + missing: p.missing, + suspectedMissing: p.suspectedMissing, + complete: p.complete, + }); + } + } + const completable = + sub.status !== "pending" && + (sub.media_kind === "movie" || (seasons.length > 0 && seasons.every((s) => s.totalKnown))); + return { subscriptionId: sub.id, status: sub.status, completable, seasons, slots: slotOut }; +} diff --git a/apps/server/src/modules/sources-identity/service/metadata-totals.ts b/apps/server/src/modules/sources-identity/service/metadata-totals.ts new file mode 100644 index 00000000..28a368f3 --- /dev/null +++ b/apps/server/src/modules/sources-identity/service/metadata-totals.ts @@ -0,0 +1,82 @@ +import type { CandidateSeason } from "@umt/adapters"; +import { sql } from "@umt/db"; +import { resolveEpisodeTotal } from "@umt/domain"; +import type { Exec } from "./common.ts"; + +// W5-C 季总集数账本(content_unit(kind=season).episode_total):refreshMetadata 用 TMDB seasons[].episodeCount 刷新。 +// 规则在 domain resolveEpisodeTotal:只升;TMDB 回退不低于已覆盖最高集;人工锁定不改;未公布(0)保持未知。 +// 季单元不存在时按元数据建(confirmed);不预建集单元(集单元仍由发布/文件懒建,provisional 不代表总数)。 + +export interface SeasonTotalsResult { + seasons: { season: number; total: number | null; known: boolean; changed: boolean }[]; + /** 本次后该作品的追踪订阅从 pending 转 active 的数量 */ + activated: number; +} + +async function ensureSeason(exec: Exec, entityId: string, season: number, name: string | null): Promise { + const hit = await exec + .selectFrom("content_unit") + .select("id") + .where("media_entity_id", "=", entityId) + .where("kind", "=", "season") + .where("season_number", "=", season) + .where("status", "!=", "retired") + .orderBy("created_at") + .executeTakeFirst(); + if (hit) return hit.id; + return ( + await exec + .insertInto("content_unit") + .values({ media_entity_id: entityId, kind: "season", season_number: season, status: "confirmed", title: name }) + .returning("id") + .executeTakeFirstOrThrow() + ).id; +} + +export async function applySeasonTotals( + exec: Exec, + entityId: string, + seasons: readonly CandidateSeason[], + now: Date, +): Promise { + const out: SeasonTotalsResult = { seasons: [], activated: 0 }; + for (const s of seasons) { + if (s.seasonNumber < 1) continue; // S00 特别篇不计入正片季 + const unitId = await ensureSeason(exec, entityId, s.seasonNumber, s.name); + const cur = await exec + .selectFrom("content_unit") + .select(["episode_total", "episode_total_locked"]) + .where("id", "=", unitId) + .forUpdate() + .executeTakeFirstOrThrow(); + const covered = await sql<{ max: number | null }>` + select max(u.episode_number)::int as max from asset_unit_coverage c + join content_unit u on u.id = c.content_unit_id + join media_asset a on a.id = c.media_asset_id + where u.media_entity_id = ${entityId} and u.kind = 'episode' and u.season_number = ${s.seasonNumber} + and a.status = 'active'`.execute(exec); + const r = resolveEpisodeTotal({ + previous: cur.episode_total, + fresh: s.episodeCount, + coveredMax: covered.rows[0]?.max ?? null, + locked: cur.episode_total_locked, + }); + if (r.changed) + await exec + .updateTable("content_unit") + .set({ episode_total: r.total, episode_total_source: r.source, episode_total_at: now }) + .where("id", "=", unitId) + .execute(); + out.seasons.push({ season: s.seasonNumber, total: r.total, known: r.known, changed: r.changed }); + } + // TMDB 已就绪:因元数据未就绪而 pending 的订阅转 active(仍按总集数是否已知决定能否完成) + const act = await exec + .updateTable("subscription") + .set({ status: "active", status_reason: "metadata_ready" }) + .where("media_entity_id", "=", entityId) + .where("status", "=", "pending") + .where("status_reason", "=", "metadata_pending") + .executeTakeFirst(); + out.activated = Number(act.numUpdatedRows ?? 0); + return out; +} diff --git a/apps/server/src/modules/sources-identity/service/metadata.ts b/apps/server/src/modules/sources-identity/service/metadata.ts index 763e1cc5..43c89044 100644 --- a/apps/server/src/modules/sources-identity/service/metadata.ts +++ b/apps/server/src/modules/sources-identity/service/metadata.ts @@ -6,6 +6,7 @@ import { jsonb } from "../rows.ts"; import type { SourcesRuntime } from "../runtime.ts"; import { type Actor, audit, type Exec } from "./common.ts"; import { resolveEntityId, seedAliases } from "./entities.ts"; +import { applySeasonTotals, type SeasonTotalsResult } from "./metadata-totals.ts"; // refreshMetadata:字段级刷新(REQ §7.3)。来源值写 metadata_field;最终值 = 覆盖(锁)> 已锁字段保持 > 来源值。 // 只更新 media_entity 的展示字段与别名,不触碰 path_binding(路径绑定不随元数据变化)。 @@ -34,6 +35,8 @@ export interface RefreshResult { reason?: string; changedFields: string[]; lockedFields: string[]; + /** 剧集:季总集数账本刷新结果(W5-C) */ + seasons?: SeasonTotalsResult["seasons"]; } export async function refreshMetadata(rt: SourcesRuntime, mediaEntityId: string): Promise { @@ -95,8 +98,13 @@ export async function refreshMetadata(rt: SourcesRuntime, mediaEntityId: string) .execute(); await seedAliases(trx, id, detail); const r = await project(trx, id, now); + // 季总集数账本:只升;回退不低于已覆盖最高集;锁定不改(订阅完成判定据此) + const totals = + ident.entity_type === "tv" && detail.seasons?.length + ? await applySeasonTotals(trx, id, detail.seasons, now) + : null; await trx.updateTable("media_entity").set({ last_metadata_refresh_at: now }).where("id", "=", id).execute(); - return { status: "refreshed" as const, ...r }; + return { status: "refreshed" as const, ...r, ...(totals ? { seasons: totals.seasons } : {}) }; }); } diff --git a/apps/server/src/modules/sources-identity/tasks.ts b/apps/server/src/modules/sources-identity/tasks.ts index c662c3a7..0ab662be 100644 --- a/apps/server/src/modules/sources-identity/tasks.ts +++ b/apps/server/src/modules/sources-identity/tasks.ts @@ -5,6 +5,7 @@ import { ingestFeed } from "./service/feeds.ts"; import { refreshMetadata } from "./service/metadata.ts"; import { resolveRelease } from "./service/resolve.ts"; import { snapshotSite } from "./service/snapshot.ts"; +import { liveSearch } from "./service/workbench.ts"; // 持久任务:输入只放业务 ID,处理器读业务表。外部请求失败的退避/熔断由适配器状态与业务表承担, // 凭据失效返回正常结果(不重试);只有意外错误(DB、TMDB 网络)抛出交给引擎重试。 @@ -53,6 +54,14 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu retryPolicy: NETWORK_RETRY, handler: (t) => t.step.run("refresh", () => refreshMetadata(rt(), t.input.mediaEntityId)), }), + defineTask({ + name: "searchSubscription", + version: 1, + inputSchema: z.object({ subscriptionId: id, manual: z.boolean().optional() }), + retryPolicy: NETWORK_RETRY, + handler: (t) => + t.step.run("search", () => searchSubscription(rt(), t.input.subscriptionId, { manual: !!t.input.manual })), + }), defineTask({ name: "snapshotSite", version: 1, @@ -70,6 +79,106 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu ); } +// ── 订阅主动搜索(W5-C,对照 MP SubscribeSearchOwner):按作品标题/原名/别名在已开启站点搜索 ── + +/** 主动搜索周期缺省 24h;subscription.params.searchIntervalHours 覆盖(0 = 关闭周期搜索) */ +export const DEFAULT_SEARCH_INTERVAL_HOURS = 24; +/** 每次最多用几个检索词(标题、原名、别名),控制站点请求量 */ +export const SEARCH_MAX_KEYWORDS = 3; + +export interface SubscriptionSearchResult { + status: "ok" | "skipped"; + reason?: string; + keywords: string[]; + sites: string[]; + searches: number; + errors: { siteId: string; keyword: string; error: string }[]; + nextSearchAt: string | null; +} + +/** + * 主动搜索:对订阅站点范围内已开启且健康的站点,按标题/原名/别名经站点适配器 search(共享限流器、实时补查合并与失败退避, + * 记 site_snapshot kind=live_search);新结果 upsert 为 source_release 并发 release.observed → 识别 → release.resolved → + * evaluateAcquisition(与 RSS 同一路径);已识别的旧结果由本任务末尾入队的订阅重评覆盖。 + */ +export async function searchSubscription( + rt: SourcesRuntime, + subscriptionId: string, + opts: { manual?: boolean } = {}, +): Promise { + const { db } = rt.ctx; + const none = { keywords: [], sites: [], searches: 0, errors: [], nextSearchAt: null }; + const sub = await db.selectFrom("subscription").selectAll().where("id", "=", subscriptionId).executeTakeFirst(); + if (!sub) return { status: "skipped", reason: "not_found", ...none }; + if (!opts.manual && sub.status !== "active" && sub.status !== "pending") + return { status: "skipped", reason: `subscription_${sub.status}`, ...none }; + if (!sub.media_entity_id) return { status: "skipped", reason: "no_entity", ...none }; + const automation = await rt.ctx.settings.get("automation"); + if (!opts.manual && (!automation.enabled || automation.readOnlyReconcile)) + return { status: "skipped", reason: "automation_blocked", ...none }; + const entity = await db + .selectFrom("media_entity") + .select(["id", "title", "original_title"]) + .where("id", "=", sub.media_entity_id) + .executeTakeFirst(); + if (!entity) return { status: "skipped", reason: "entity_not_found", ...none }; + const aliases = await db + .selectFrom("title_alias") + .select(["alias", "kind"]) + .where("media_entity_id", "=", entity.id) + .where("content_unit_id", "is", null) + .orderBy("created_at") + .limit(20) + .execute(); + const seen = new Set(); + const keywords: string[] = []; + for (const k of [entity.title, entity.original_title, ...aliases.map((a) => a.alias)]) { + const v = (k ?? "").trim(); + const norm = v.toLowerCase(); + if (!v || seen.has(norm)) continue; + seen.add(norm); + keywords.push(v); + if (keywords.length >= SEARCH_MAX_KEYWORDS) break; + } + const params = (sub.params ?? {}) as { siteIds?: string[]; searchIntervalHours?: number }; + let sq = db.selectFrom("site").selectAll().where("enabled", "=", true).where("health_status", "!=", "failing"); + if (params.siteIds?.length) sq = sq.where("id", "in", params.siteIds); + const sites = await sq.orderBy("name").execute(); + const errors: SubscriptionSearchResult["errors"] = []; + let searches = 0; + for (const site of sites) { + for (const kw of keywords) { + searches++; + const r = await liveSearch(rt, site, kw).catch((e) => ({ error: (e as Error).message, coalesced: false })); + if (r.error) errors.push({ siteId: site.id, keyword: kw, error: r.error }); + } + } + const hours = params.searchIntervalHours ?? DEFAULT_SEARCH_INTERVAL_HOURS; + const now = rt.now(); + const next = hours > 0 ? new Date(now.getTime() + hours * 3600_000) : null; + await db + .updateTable("subscription") + .set({ last_searched_at: now, next_search_at: next }) + .where("id", "=", sub.id) + .execute(); + // 已识别的旧结果:按订阅重评(新结果走 release.observed → resolved → evaluate) + await rt.ctx.engine + .enqueue( + "evaluateAcquisition", + { subscriptionId: sub.id }, + { idempotencyKey: `search-eval:${sub.id}:${now.getTime()}`, traceId: sub.id }, + ) + .catch(() => {}); + return { + status: "ok", + keywords, + sites: sites.map((s) => s.id), + searches, + errors, + nextSearchAt: next ? next.toISOString() : null, + }; +} + type P = Record; export const sourcesOutboxRoutes: Record = { diff --git a/apps/server/test/e2e/e9-continuous-series.test.ts b/apps/server/test/e2e/e9-continuous-series.test.ts new file mode 100644 index 00000000..2d8f7ff2 --- /dev/null +++ b/apps/server/test/e2e/e9-continuous-series.test.ts @@ -0,0 +1,274 @@ +import { + createMockRssServer, + type MockFeedItem, + type MockRssServer, + TMDB_FIXTURES, + type TmdbFixture, +} from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { TORRENT_SIZE, torrentFor } from "./fixtures/modules.ts"; +import { completeDownload, createE2E, type E2E, pgAvailable, seedBase, waitFor } from "./harness.ts"; + +// E9 连续更新剧集(W5-C,对照 MP 缺集 − note 与批量下载):全部经真实任务引擎 + outbox 中继。 +// 季共 3 集(mock TMDB),新作品策略按 once 订阅。E01 还在下载(Slot 忙)时 E02、E03 与 E02 的另一个版本先后到达: +// · E02/E03 与在途的 E01 单元不相交 → 各自得到计划并同时下载(Slot×单元预留),不再因 slot_busy 丢集; +// · E02 另一版本与在途 E02 单元冲突 → 延后(slot_deferral),计划结束后重评时已覆盖,不重复下载; +// · refreshMetadata(编排循环按周期入队)把 TMDB 季集数写进季单元 → 三集全部入库后 once 订阅完成并写历史。 + +const hasPg = await pgAvailable(); +const BASE = "https://audiences.example"; +const EP = (n: number, grp = "x264-GRP") => `Three.Eps.S01E0${n}.1080p.WEB-DL.${grp}`; +const ALT = EP(2, "x265-ALT"); + +const SHOW: TmdbFixture = { + type: "tv", + id: 990001, + original_title: "Three Eps", + original_language: "en", + origin_country: ["US"], + date: "2026-09-01", + titles: { "zh-CN": "三集剧", "en-US": "Three Eps" }, + popularity: 50, + alternative_titles: [], + external_ids: {}, + seasons: [ + { + season_number: 1, + air_date: "2026-09-01", + names: { "zh-CN": "第 1 季", "en-US": "Season 1" }, + episodes: [1, 2, 3].map((i) => ({ + episode_number: i, + air_date: `2026-09-0${i}`, + names: { "zh-CN": `第 ${i} 集`, "en-US": `Episode ${i}` }, + })), + }, + ], +}; + +const item = (id: number, title: string): MockFeedItem => ({ + guid: `aud-${id}`, + title, + link: `${BASE}/details.php?id=${id}&hit=1`, + enclosureUrl: `${BASE}/download.php?id=${id}&passkey=E9PASSKEY`, + size: TORRENT_SIZE, + pubDate: new Date(), + categories: ["剧集"], +}); + +describe.skipIf(!hasPg)("E9 连续更新剧集:Slot 忙时到达的集全部入库", () => { + let e: E2E; + let rss: MockRssServer; + let feedId: string; + let entityId: string; + let slotId: string; + let subId: string; + + let lastSlot = -1; + const fetchFeed = async () => { + const slotOf = () => Math.floor(Date.now() / 10_000); + if (slotOf() === lastSlot) await new Promise((r) => setTimeout(r, (lastSlot + 1) * 10_000 - Date.now() + 50)); + lastSlot = slotOf(); + const r = await e.api("POST", `/feed-sources/${feedId}/fetch`); + expect(r.status).toBe(200); + }; + const dtOf = (title: string) => + e.db.selectFrom("download_task").selectAll().where("torrent_hash", "=", torrentFor(title).hash).executeTakeFirst(); + const importedFor = async (title: string) => { + const dt = await waitFor(async () => dtOf(title), 60_000, `${title} download_task`); + return waitFor( + async () => + e.db + .selectFrom("organization_record") + .selectAll() + .where("download_task_id", "=", dt.id) + .where("status", "in", ["succeeded", "failed", "partial"]) + .executeTakeFirst(), + 90_000, + `${title} 整理记录结束`, + ); + }; + + beforeAll(async () => { + e = await createE2E("e9", { tmdbFixtures: [...TMDB_FIXTURES, SHOW] }); + const base = await seedBase(e, { baseUrl: `${BASE}/` }); + rss = await createMockRssServer({ passkey: "E9PASSKEY", items: [item(930001, EP(1))] }); + const urlRef = await e.booted.ctx.secrets.create({ purpose: "feed_url", plaintext: rss.feedUrl }); + feedId = ( + await e.db + .insertInto("feed_source") + .values({ + site_id: base.siteId, + name: "观众-剧集", + purpose: "series", + url_secret_ref: urlRef, + url_display: rss.feedUrl.replace(/passkey=[^&]+/, "passkey=[REDACTED]"), + poll_interval_sec: 3600, + }) + .returning("id") + .executeTakeFirstOrThrow() + ).id; + const created = await e.api("POST", "/policy-revisions", { + policyKey: "series", + kind: "acquisition", + title: "剧集 1080p", + scope: { mediaKinds: ["series"] }, + rules: { + newTitle: { + enabled: true, + windowDays: 3650, + mediaKinds: ["series"], + slots: [{ name: "1080p", hardConditions: { resolutions: ["1080p"] } }], + }, + }, + }); + expect(created.status).toBe(201); + const pub = await e.api("POST", `/policy-revisions/${created.body.data.id}/publish`, { + expectedRevision: created.body.data.revision, + }); + expect(pub.status).toBe(200); + }, 120_000); + + afterAll(async () => { + await rss?.close(); + await e?.close(); + }, 60_000); + + it("E01 下载中;TMDB 季集数写入季单元(总集数 3)", async () => { + await fetchFeed(); + const rel = await waitFor( + async () => + e.db + .selectFrom("source_release") + .selectAll() + .where("title", "=", EP(1)) + .where("resolution_status", "=", "resolved") + .executeTakeFirst(), + 60_000, + "E01 识别完成", + ); + entityId = rel.media_entity_id!; + const sub = await waitFor( + async () => + e.db.selectFrom("subscription").selectAll().where("media_entity_id", "=", entityId).executeTakeFirst(), + 30_000, + "新作品订阅", + ); + subId = sub.id; + expect(sub).toMatchObject({ mode: "once", status: "active" }); + slotId = ( + await e.db + .selectFrom("acquisition_slot") + .select("id") + .where("media_entity_id", "=", entityId) + .executeTakeFirstOrThrow() + ).id; + await waitFor(async () => e.qb.get(torrentFor(EP(1)).hash), 60_000, "qB 收到 E01"); + const season = await waitFor( + async () => + e.db + .selectFrom("content_unit") + .select(["episode_total", "episode_total_source"]) + .where("media_entity_id", "=", entityId) + .where("kind", "=", "season") + .where("episode_total", "is not", null) + .executeTakeFirst(), + 60_000, + "季总集数刷新", + ); + expect(season).toEqual({ episode_total: 3, episode_total_source: "tmdb" }); + }, 180_000); + + it("Slot 忙时 E02、E03 与 E02 另一版本到达:E02/E03 同时在途,另一版本延后", async () => { + rss.addItem(item(930002, EP(2))); + rss.addItem(item(930003, EP(3))); + rss.addItem(item(930004, ALT)); + await fetchFeed(); + // E02(任一版本)与 E03 都进了 qB;E01 仍未完成 → 三个在途计划并存 + await waitFor(async () => e.qb.get(torrentFor(EP(3)).hash), 90_000, "qB 收到 E03"); + const e02 = await waitFor( + async () => e.qb.get(torrentFor(EP(2)).hash) ?? e.qb.get(torrentFor(ALT).hash), + 60_000, + "qB 收到 E02", + ); + const inflight = await e.db + .selectFrom("acquisition_plan") + .select(["unit_keys", "status"]) + .where("slot_id", "=", slotId) + .where("status", "in", ["reserved", "submitted"]) + .execute(); + expect(inflight.map((p) => p.unit_keys[0]).sort()).toEqual(["s1e1", "s1e2", "s1e3"]); + const slot = await e.db + .selectFrom("acquisition_slot") + .selectAll() + .where("id", "=", slotId) + .executeTakeFirstOrThrow(); + expect(slot.status).toBe("downloading"); + // 另一版本 E02:延后而不是丢弃 + const loser = e02.name.includes("x265-ALT") ? EP(2) : ALT; + const loserRel = await waitFor( + async () => + e.db + .selectFrom("source_release") + .select("id") + .where("title", "=", loser) + .where("resolution_status", "=", "resolved") + .executeTakeFirst(), + 60_000, + "另一版本识别", + ); + await waitFor( + async () => + e.db + .selectFrom("slot_deferral") + .selectAll() + .where("slot_id", "=", slotId) + .where("source_release_id", "=", loserRel.id) + .executeTakeFirst(), + 60_000, + "另一版本延后", + ); + }, 180_000); + + it("三集依次完成全部入库;once 订阅完成并写历史;延后的版本重评后不重复下载", async () => { + const e02Title = e.qb.get(torrentFor(EP(2)).hash) ? EP(2) : ALT; + for (const t of [EP(3), e02Title, EP(1)]) { + await completeDownload(e, t); + const rec = await importedFor(t); + expect(rec.status).toBe("succeeded"); + } + const cov = await e.db + .selectFrom("asset_unit_coverage as c") + .innerJoin("content_unit as u", "u.id", "c.content_unit_id") + .innerJoin("media_asset as a", "a.id", "c.media_asset_id") + .select(["u.episode_number"]) + .where("a.media_entity_id", "=", entityId) + .where("a.status", "=", "active") + .orderBy("u.episode_number") + .execute(); + expect(cov.map((c) => c.episode_number)).toEqual([1, 2, 3]); + const sub = await waitFor( + async () => { + const s = await e.db.selectFrom("subscription").selectAll().where("id", "=", subId).executeTakeFirstOrThrow(); + return s.status === "completed" ? s : null; + }, + 30_000, + "订阅完成", + ); + expect(sub.status_reason).toBe("slot_satisfied"); + const slot = await e.db + .selectFrom("acquisition_slot") + .selectAll() + .where("id", "=", slotId) + .executeTakeFirstOrThrow(); + expect(slot.status).toBe("satisfied"); + const hist = await e.api("GET", `/subscription-history?subscriptionId=${subId}`); + expect(hist.body.data.items).toHaveLength(1); + expect(hist.body.data.items[0]).toMatchObject({ reason: "slot_satisfied", mediaEntityId: entityId }); + // 进度 API:三集已覆盖,总集数已知 + const prog = await e.api("GET", `/subscriptions/${subId}/progress`); + expect(prog.body.data.seasons[0]).toMatchObject({ episodeTotal: 3, totalKnown: true, complete: true }); + // 延后的另一版本:计划结束后重评,E02 已覆盖 → 不下载;三个种子各提交一次 + await new Promise((r) => setTimeout(r, 1500)); + expect(e.qb.callsTo("torrents/add")).toHaveLength(3); + }, 300_000); +}); diff --git a/apps/server/test/sources-identity/mteam-tasks.test.ts b/apps/server/test/sources-identity/mteam-tasks.test.ts index 5ffc3428..39583ecc 100644 --- a/apps/server/test/sources-identity/mteam-tasks.test.ts +++ b/apps/server/test/sources-identity/mteam-tasks.test.ts @@ -72,7 +72,13 @@ describe.skipIf(!up)("M-Team 快照与任务注册", () => { it("模块注册 4 个任务与 3 条 outbox 路由;任务处理器经 step.run 调服务", async () => { const reg = new TaskRegistry(); h.module.registerTasks(reg, h.ctx); - expect(reg.names()).toEqual(["ingestFeed", "refreshMetadata", "resolveObservation", "snapshotSite"]); + expect(reg.names()).toEqual([ + "ingestFeed", + "refreshMetadata", + "resolveObservation", + "searchSubscription", + "snapshotSite", + ]); expect(Object.keys(h.module.outboxRoutes!(h.ctx)).sort()).toEqual([ "feed.due", "feed.gap_suspected", diff --git a/apps/server/test/sources-identity/subscription-search.test.ts b/apps/server/test/sources-identity/subscription-search.test.ts new file mode 100644 index 00000000..6f33298e --- /dev/null +++ b/apps/server/test/sources-identity/subscription-search.test.ts @@ -0,0 +1,149 @@ +import { createMockNexusPhpSite, type MockNexusPhpSite } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { applySeasonTotals } from "../../src/modules/sources-identity/service/metadata-totals.ts"; +import { searchSubscription } from "../../src/modules/sources-identity/tasks.ts"; +import { createHarness, type Harness, json, pgAvailable } from "./harness.ts"; + +// W5-C 订阅主动搜索(mock NexusPHP 站点:适配器 search + 限流 + 快照)与季总集数账本(refreshMetadata 的写入部分)。 + +const up = await pgAvailable(); + +describe.skipIf(!up)("订阅主动搜索与季总集数", () => { + let h: Harness; + let mock: MockNexusPhpSite; + let siteId: string; + + beforeAll(async () => { + mock = await createMockNexusPhpSite(); + h = await createHarness("subsearch"); + const ok = await json( + await h.req("/sites", { + body: { + key: "audiences", + name: "观众", + adapter: "nexusphp", + baseUrl: `${mock.baseUrl}/`, + rateLimit: { minIntervalMs: 0, concurrency: 4 }, + }, + }), + ); + expect(ok.status).toBe(200); + siteId = ok.body.data.id; + await h.req("/site-accounts", { body: { siteId, label: "主号", authKind: "cookie", credential: mock.cookie } }); + }); + afterAll(async () => { + await h?.close(); + await mock?.close(); + }); + + const entity = async (title: string, kind: "movie" | "series" = "movie") => + ( + await h.db + .insertInto("media_entity") + .values({ kind, title, original_title: title, year: 2021, status: "active" }) + .returning("id") + .executeTakeFirstOrThrow() + ).id; + const subscribe = async (entityId: string, extra: Record = {}) => + ( + await h.db + .insertInto("subscription") + .values({ media_entity_id: entityId, media_kind: "movie", mode: "once", ...extra }) + .returning("id") + .executeTakeFirstOrThrow() + ).id; + + it("按标题在已开启站点搜索:结果 upsert 为 source_release(新结果发 release.observed),排下次搜索并入队订阅重评", async () => { + const e = await entity("Dune"); + const subId = await subscribe(e); + const searches = () => mock.requests.filter((r) => r.path.includes("search=")).length; + const before = searches(); + const r = await searchSubscription(h.rt, subId); + expect(r).toMatchObject({ status: "ok", keywords: ["Dune"], sites: [siteId], searches: 1, errors: [] }); + expect(searches() - before).toBe(1); + const rels = await h.db + .selectFrom("source_release") + .select(["id", "title"]) + .where("site_id", "=", siteId) + .execute(); + expect(rels.some((x) => /Dune/i.test(x.title))).toBe(true); + expect((await h.outbox("release.observed")).length).toBeGreaterThanOrEqual(1); + const snap = await h.db + .selectFrom("site_snapshot") + .selectAll() + .where("query", "=", "Dune") + .executeTakeFirstOrThrow(); + expect(snap.kind).toBe("live_search"); + const sub = await h.db.selectFrom("subscription").selectAll().where("id", "=", subId).executeTakeFirstOrThrow(); + expect(sub.last_searched_at).not.toBeNull(); + expect(sub.next_search_at!.getTime() - sub.last_searched_at!.getTime()).toBe(24 * 3600_000); + expect(h.engine.enqueued.at(-1)).toMatchObject({ task: "evaluateAcquisition", input: { subscriptionId: subId } }); + }); + + it("站点范围外不搜;暂停订阅只允许手动;searchIntervalHours=0 关闭周期", async () => { + const e = await entity("Bear"); + const scoped = await subscribe(e, { + params: JSON.stringify({ siteIds: ["00000000-0000-4000-8000-000000000000"] }), + }); + expect(await searchSubscription(h.rt, scoped)).toMatchObject({ status: "ok", sites: [], searches: 0 }); + const paused = await subscribe(e, { status: "paused", params: JSON.stringify({ searchIntervalHours: 0 }) }); + expect(await searchSubscription(h.rt, paused)).toMatchObject({ status: "skipped", reason: "subscription_paused" }); + const manual = await searchSubscription(h.rt, paused, { manual: true }); + expect(manual).toMatchObject({ status: "ok", searches: 1, nextSearchAt: null }); + }); + + it("季总集数:TMDB 集数写季单元(S00 不计、未公布保持未知);回退接受但被已覆盖最高集兜底;锁定不改;pending 订阅转 active", async () => { + const e = await entity("Series T", "series"); + const sub = await subscribe(e, { media_kind: "series", status: "pending", status_reason: "metadata_pending" }); + const now = new Date(); + const r1 = await h.db.transaction().execute((trx) => + applySeasonTotals( + trx, + e, + [ + { seasonNumber: 0, episodeCount: 3, name: "特别篇", airDate: null }, + { seasonNumber: 1, episodeCount: 8, name: "第 1 季", airDate: null }, + { seasonNumber: 2, episodeCount: 0, name: "第 2 季", airDate: null }, + ], + now, + ), + ); + expect(r1.seasons).toEqual([ + { season: 1, total: 8, known: true, changed: true }, + { season: 2, total: null, known: false, changed: false }, + ]); + expect(r1.activated).toBe(1); + expect( + (await h.db.selectFrom("subscription").select("status").where("id", "=", sub).executeTakeFirstOrThrow()).status, + ).toBe("active"); + const r2 = await h.db + .transaction() + .execute((trx) => + applySeasonTotals(trx, e, [{ seasonNumber: 1, episodeCount: 12, name: null, airDate: null }], now), + ); + expect(r2.seasons[0]).toMatchObject({ total: 12, changed: true }); + await h.db + .updateTable("content_unit") + .set({ episode_total_locked: true }) + .where("media_entity_id", "=", e) + .where("season_number", "=", 1) + .execute(); + const r3 = await h.db + .transaction() + .execute((trx) => + applySeasonTotals(trx, e, [{ seasonNumber: 1, episodeCount: 20, name: null, airDate: null }], now), + ); + expect(r3.seasons[0]).toMatchObject({ total: 12, changed: false }); + const units = await h.db + .selectFrom("content_unit") + .select(["kind", "season_number", "episode_total", "episode_total_source"]) + .where("media_entity_id", "=", e) + .orderBy("season_number") + .execute(); + // 只建季单元,不预建集单元 + expect(units).toEqual([ + { kind: "season", season_number: 1, episode_total: 12, episode_total_source: "tmdb" }, + { kind: "season", season_number: 2, episode_total: null, episode_total_source: null }, + ]); + }); +});