From e7de7e2e69e36359e01565a4b3b93478eb192e34 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 29 Sep 2026 20:54:19 +0800 Subject: [PATCH] feat(organize): identify takeover downloads by file name and close organize decisions Takeover identify (item 2): - Adopted tasks without source release / plan: on completion planImport parses main video path (file + parent + grandparent dirs, like MP MetaInfoPath) merged with the torrent name, then runs ID-2 matching via identifyParsed (local recall + TMDB; trigger "takeover" may query TMDB). auto -> bind/create entity, backfill download_task.media_entity_id and organize right away; no pool entry, no identity decision. - Failure / ambiguity: organize_identity decision with evidence and options: candidates, select entity (local / TMDB), provisional entity, dismiss. Decision closure (item 3): - decision-hooks registry: the decisions resolve route runs the owning module's executor before closing; errors keep the decision open. - organize_identity: backfill entity, freeze manual plan, enqueue run. - organize_blocked with target_exists: replace (verify + trash old) / keep both (rename " (n)" with sidecars) / skip. - Successful organize closes both organize_identity and organize_blocked of the same download task (resolution manual_organize), also for manual path organizes matched to download tasks via path mappings. Co-Authored-By: Claude Opus 5.5 --- .../acquisition-organize/__tests__/harness.ts | 10 +- .../__tests__/takeover.test.ts | 283 +++++++++++++ .../src/modules/acquisition-organize/index.ts | 4 + .../src/modules/acquisition-organize/ports.ts | 44 ++ .../acquisition-organize/services/common.ts | 6 +- .../acquisition-organize/services/organize.ts | 86 +++- .../acquisition-organize/services/takeover.ts | 390 ++++++++++++++++++ apps/server/src/modules/decision-hooks.ts | 38 ++ .../sources-identity/routes/decisions.ts | 21 +- .../sources-identity/service/resolve.ts | 187 ++++++++- .../sources-identity/lazy-identity.test.ts | 44 +- docs/runbook/files.md | 12 + docs/runbook/identity.md | 19 + 13 files changed, 1127 insertions(+), 17 deletions(-) create mode 100644 apps/server/src/modules/acquisition-organize/__tests__/takeover.test.ts create mode 100644 apps/server/src/modules/acquisition-organize/services/takeover.ts create mode 100644 apps/server/src/modules/decision-hooks.ts diff --git a/apps/server/src/modules/acquisition-organize/__tests__/harness.ts b/apps/server/src/modules/acquisition-organize/__tests__/harness.ts index 9e689fed..c04b34fc 100644 --- a/apps/server/src/modules/acquisition-organize/__tests__/harness.ts +++ b/apps/server/src/modules/acquisition-organize/__tests__/harness.ts @@ -14,6 +14,7 @@ import { hashApiKey } from "../../../http/auth.ts"; import { EventBus } from "../../../http/sse.ts"; import { createLogger } from "../../../logger.ts"; import { createCapabilityRegistry } from "../../agent/capability-registry.ts"; +import { createSourcesIdentityModule } from "../../sources-identity/index.ts"; import type { ServerContext } from "../../types.ts"; import { type AoOverrides, createAcquisitionOrganizeModule } from "../index.ts"; import type { AoDeps } from "../services/common.ts"; @@ -129,7 +130,12 @@ export interface Harness { close(): Promise; } -export async function createHarness(name: string, overrides: AoOverrides = {}): Promise { +export async function createHarness( + name: string, + overrides: AoOverrides = {}, + /** withSources:同时装配 sources-identity 模块(待处理中心路由等) */ + opts: { withSources?: boolean } = {}, +): Promise { const schema = `ao_test_${name}_${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`; const admin = new pg.Client({ connectionString: DB_URL }); await admin.connect(); @@ -207,7 +213,7 @@ export async function createHarness(name: string, overrides: AoOverrides = {}): role: "all", version: "test", db, - modules: [module], + modules: opts.withSources ? [module, createSourcesIdentityModule()] : [module], ctx, apiKeyStore: { async findByHash(hash) { diff --git a/apps/server/src/modules/acquisition-organize/__tests__/takeover.test.ts b/apps/server/src/modules/acquisition-organize/__tests__/takeover.test.ts new file mode 100644 index 00000000..3a5f280f --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/__tests__/takeover.test.ts @@ -0,0 +1,283 @@ +import { existsSync, readdirSync, statSync } from "node:fs"; +import { join } from "node:path"; +import { createMockQb, createTempTree, type MockQb, type TempTree } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { addAliasIfAbsent } from "../../sources-identity/service/entities.ts"; +import { syncDownloads } from "../services/downloads.ts"; +import { executeRecord, planImport } from "../services/organize.ts"; +import { organizeDecisionHook, pickMainVideo, takeoverParsed } from "../services/takeover.ts"; +import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "./harness.ts"; + +// QB-1:接管任务(用户手动加进 qB、贴管理标签,没有站点资源)完成后按 种子名 + 主视频文件名 识别 → 直接整理; +// 识别失败建带选项的事项(候选 / 选择作品 / 临时作品 / 忽略),处理后执行并闭环;目标已存在 → 替换 / 保留两者 / 跳过; +// 手动按路径整理也关闭同一下载任务的整理事项;作品详情下载历史列出接管任务与整理记录。 + +const hasPg = await pgAvailable(); +const SIZE = 300_000; + +function walkFiles(dir: string): string[] { + const out: string[] = []; + for (const n of readdirSync(dir)) { + const p = join(dir, n); + if (statSync(p).isDirectory()) out.push(...walkFiles(p)); + else out.push(p); + } + return out; +} + +describe("接管识别:文件名解析", () => { + it("主视频取要下载的最大视频,跳过 sample;文件名 + 目录 + 种子名合并", () => { + expect( + pickMainVideo([ + { name: "Movie.2021/Sample/sample.mkv", size: 9e9 }, + { name: "Movie.2021/Movie.2021.1080p.mkv", size: 5e9 }, + { name: "Movie.2021/Extra.mkv", size: 6e9, priority: 0 }, + { name: "Movie.2021/Movie.2021.nfo", size: 1e3 }, + ]), + ).toBe("Movie.2021/Movie.2021.1080p.mkv"); + expect(pickMainVideo([{ name: "a.nfo", size: 1 }])).toBeNull(); + // 文件名只有集号:标题 / 季从目录名(=种子根目录)补 + const p = takeoverParsed("The.Bear.S02.1080p.WEB-DL", "The.Bear.S02.1080p.WEB-DL/E03.mkv"); + expect(p.title?.toLowerCase()).toContain("bear"); + expect(p.season).toBe(2); + expect(p.episodes).toEqual({ start: 3, end: 3 }); + // 单文件种子:种子名即文件名 + const m = takeoverParsed("Oppenheimer.2023.2160p.UHD.BluRay.mkv", "Oppenheimer.2023.2160p.UHD.BluRay.mkv"); + expect(m.year).toBe(2023); + // 文件名解析不出标题:用种子名 + expect(takeoverParsed("Heat.1995.1080p.BluRay", "Heat.1995.1080p.BluRay/0001.mkv").year).toBe(1995); + }); +}); + +describe.skipIf(!hasPg)("QB-1 接管任务识别与事项闭环(真实 PG + mock qB + 临时目录)", () => { + let h: Harness; + let qb: MockQb; + let lib: TempTree; + let cache: TempTree; + let n = 0; + const actor = { type: "user", id: "tester" }; + + beforeAll(async () => { + qb = await createMockQb({ addVisibleAfterPolls: 0 }); + lib = await createTempTree({}, { sentinel: true, prefix: "umt-lib-" }); + cache = await createTempTree({}, { prefix: "umt-cache-" }); + h = await createHarness("takeover", {}, { withSources: true }); + await seedBase(h.db, { libraryPath: lib.root, cachePath: cache.root, qbUrl: qb.url }); + await h.setSettings("retention", { seedRetentionDays: 0 }); + }, 60_000); + + afterAll(async () => { + await h?.close(); + await qb?.close(); + await lib?.cleanup(); + await cache?.cleanup(); + }); + + /** 用户在 qB 里手动加的任务(贴管理标签)→ 同步接管 → 完成落盘 */ + async function adoptCompleted(name: string, file: string, size = SIZE) { + n++; + const hash = n.toString(16).padStart(40, "a"); + qb.seed({ hash, name, tags: ["umt"], save_path: cache.root }); + await qb.complete(hash, { files: [{ path: `${name}/${file}`, size }] }); + const sync = await syncDownloads(h.d); + expect(sync.adopted).toBeGreaterThanOrEqual(1); + const dt = await h.db + .selectFrom("download_task") + .selectAll() + .where("torrent_hash", "=", hash) + .executeTakeFirstOrThrow(); + expect(dt).toMatchObject({ origin: "adopted", source_release_id: null, status: "completed" }); + return dt; + } + + const decisionOf = (key: string) => + h.db + .selectFrom("decision") + .selectAll() + .where("dedupe_key", "=", key) + .orderBy("created_at", "desc") + .executeTakeFirst(); + + let movieId: string; + let firstDtId: string; + + it("识别成功(本地别名召回):不建事项,直接整理入库并回填下载任务的作品", async () => { + movieId = await seedEntity(h.db, { kind: "movie", title: "Quiet Harbor", year: 2021 }); + await addAliasIfAbsent(h.db, movieId, "Quiet Harbor", "original", "user"); + const dt = await adoptCompleted( + "Quiet.Harbor.2021.1080p.BluRay.x264-GRP", + "Quiet.Harbor.2021.1080p.BluRay.x264-GRP.mkv", + ); + firstDtId = dt.id; + const plan = await planImport(h.d, { downloadTaskId: dt.id, trigger: "auto", actorType: "system" }); + expect(plan.status).toBe("running"); + const done = await executeRecord(h.d, plan.recordId!); + expect(done.status).toBe("succeeded"); + const row = await h.db.selectFrom("download_task").selectAll().where("id", "=", dt.id).executeTakeFirstOrThrow(); + expect(row).toMatchObject({ media_entity_id: movieId, status: "imported" }); + expect(await decisionOf(`organize_identity:${dt.id}`)).toBeUndefined(); + }); + + let unknownDtId: string; + + it("识别失败:事项带 选择作品(本地 / TMDB)/ 临时作品 / 忽略;选临时作品 → 建临时作品并整理,事项闭环", async () => { + const dt = await adoptCompleted("Zzqx.Vorlanth.2019.1080p.WEB-DL", "Zzqx.Vorlanth.2019.1080p.WEB-DL.mkv"); + unknownDtId = dt.id; + const plan = await planImport(h.d, { downloadTaskId: dt.id, trigger: "auto", actorType: "system" }); + expect(plan).toMatchObject({ status: "failed", reason: "identity_unknown" }); + const dec = (await decisionOf(`organize_identity:${dt.id}`))!; + expect(dec).toBeTruthy(); + const keys = (dec.options as { key: string }[]).map((o) => o.key); + expect(keys).toEqual(expect.arrayContaining(["select_entity", "select_tmdb", "provisional", "dismiss"])); + expect(dec.evidence).toMatchObject({ downloadTaskId: dt.id, outcome: "no_match" }); + expect((dec.evidence as { parsed: { year: number } }).parsed.year).toBe(2019); + + const hook = organizeDecisionHook(h.d); + expect(hook.match(dec)).toBe(true); + // 选择作品(本地)但没给作品 → 错误,事项保持打开 + await expect(hook.execute(dec, { optionKey: "select_entity" }, actor)).rejects.toThrow(/mediaEntityId/); + const out = await hook.execute(dec, { optionKey: "provisional" }, actor); + expect(out).toMatchObject({ kind: "organize", status: "running" }); + const ent = await h.db + .selectFrom("media_entity") + .selectAll() + .where("id", "=", out.mediaEntityId as string) + .executeTakeFirstOrThrow(); + expect(ent).toMatchObject({ status: "provisional", year: 2019 }); + expect(h.engine.enqueued.some((e) => (e.input as { recordId?: string }).recordId === out.recordId)).toBe(true); + const done = await executeRecord(h.d, out.recordId as string); + expect(done.status).toBe("succeeded"); + const closed = (await decisionOf(`organize_identity:${dt.id}`))!; + expect(closed.status).toBe("resolved"); + expect(closed.resolution).toMatchObject({ key: "manual_organize", organizationRecordId: out.recordId }); + const row = await h.db.selectFrom("download_task").selectAll().where("id", "=", dt.id).executeTakeFirstOrThrow(); + expect(row.media_entity_id).toBe(ent.id); + }); + + it("待处理中心 API:选候选 / 本地作品 → 执行并关闭;缺作品 → 400 且事项保持打开", async () => { + const target = await seedEntity(h.db, { kind: "movie", title: "Amber Coast", year: 2017 }); + const dt = await adoptCompleted("Qwzz.Nomatch.2017.1080p.WEB", "Qwzz.Nomatch.2017.1080p.WEB.mkv"); + await planImport(h.d, { downloadTaskId: dt.id, trigger: "auto", actorType: "system" }); + const dec = (await decisionOf(`organize_identity:${dt.id}`))!; + const bad = await h.api("POST", `/decisions/${dec.id}/resolve`, { optionKey: "select_entity" }); + expect(bad.status).toBe(400); + expect((await decisionOf(`organize_identity:${dt.id}`))?.status).toBe("open"); + const ok = await h.api("POST", `/decisions/${dec.id}/resolve`, { + optionKey: "select_entity", + identity: { mediaEntityId: target }, + }); + expect(ok.status).toBe(200); + expect(ok.body.data.status).toBe("resolved"); + expect(ok.body.data.resolution.execution).toMatchObject({ kind: "organize", mediaEntityId: target }); + const row = await h.db.selectFrom("download_task").selectAll().where("id", "=", dt.id).executeTakeFirstOrThrow(); + expect(row.media_entity_id).toBe(target); + const recId = ok.body.data.resolution.execution.recordId as string; + expect((await executeRecord(h.d, recId)).status).toBe("succeeded"); + }); + + it("识别失败后在文件管理按路径手动整理:同一下载任务的整理事项关闭,回填作品", async () => { + const other = await seedEntity(h.db, { kind: "movie", title: "Glass Orchard", year: 2018 }); + const dt = await adoptCompleted("Xqqz.Unmatched.2018.720p.HDTV", "Xqqz.Unmatched.2018.720p.HDTV.mkv"); + await planImport(h.d, { downloadTaskId: dt.id, trigger: "auto", actorType: "system" }); + expect((await decisionOf(`organize_identity:${dt.id}`))?.status).toBe("open"); + const src = join(cache.root, "Xqqz.Unmatched.2018.720p.HDTV"); + const plan = await planImport(h.d, { + sourcePaths: [src], + mediaEntityId: other, + trigger: "manual", + actorType: "user", + actorId: "tester", + }); + expect(plan.status).toBe("running"); + expect((await executeRecord(h.d, plan.recordId!)).status).toBe("succeeded"); + const closed = (await decisionOf(`organize_identity:${dt.id}`))!; + expect(closed).toMatchObject({ status: "resolved" }); + expect(closed.resolution).toMatchObject({ key: "manual_organize" }); + const row = await h.db.selectFrom("download_task").selectAll().where("id", "=", dt.id).executeTakeFirstOrThrow(); + expect(row.media_entity_id).toBe(other); + }); + + it("目标已存在:事项给 替换 / 保留两者 / 跳过;保留两者 → 新文件加序号入库,旧文件不动,事项闭环", async () => { + // 同一作品的另一份(不同内容、同名)→ 目标文件已存在 + const dt = await adoptCompleted( + "Quiet.Harbor.2021.1080p.BluRay.x264-GRP", + "Quiet.Harbor.2021.1080p.BluRay.x264-GRP.mkv", + SIZE + 1000, + ); + // 同名种子在 qB 里是新 hash;接管识别仍命中同一作品 + const plan = await planImport(h.d, { downloadTaskId: dt.id, trigger: "auto", actorType: "system" }); + expect(plan.status).toBe("failed"); + expect(plan.reason).toContain("target_exists"); + const dec = (await decisionOf(`organize_blocked:${dt.id}`))!; + expect((dec.options as { key: string }[]).map((o) => o.key)).toEqual(["replace", "keep_both", "skip"]); + const before = walkFiles(lib.root).filter((p) => p.endsWith(".mkv")); + expect(before.filter((p) => p.includes("Quiet"))).toHaveLength(1); + const hook = organizeDecisionHook(h.d); + const out = await hook.execute(dec, { optionKey: "keep_both" }, actor); + expect(out).toMatchObject({ kind: "organize", choice: "keep_both", status: "running" }); + expect((await executeRecord(h.d, out.recordId as string)).status).toBe("succeeded"); + const after = walkFiles(lib.root).filter((p) => p.endsWith(".mkv")); + expect(after).toHaveLength(before.length + 1); + const renamed = after.find((p) => p.includes(" (2).")); + expect(renamed && statSync(renamed).size).toBe(SIZE + 1000); + for (const p of before) expect(existsSync(p)).toBe(true); + expect((await decisionOf(`organize_blocked:${dt.id}`))?.status).toBe("resolved"); + }); + + it("目标已存在 → 跳过:其余照常,已存在的不整理;替换:核验后替换", async () => { + const dt = await adoptCompleted( + "Quiet.Harbor.2021.1080p.BluRay.x264-GRP", + "Quiet.Harbor.2021.1080p.BluRay.x264-GRP.mkv", + SIZE + 2000, + ); + await planImport(h.d, { downloadTaskId: dt.id, trigger: "auto", actorType: "system" }); + const dec = (await decisionOf(`organize_blocked:${dt.id}`))!; + const hook = organizeDecisionHook(h.d); + const skip = await hook.execute(dec, { optionKey: "skip" }, actor); + const rec = await h.db + .selectFrom("organization_record") + .selectAll() + .where("id", "=", skip.recordId as string) + .executeTakeFirstOrThrow(); + expect(rec.file_skipped).toBeGreaterThanOrEqual(1); + + const dt2 = await adoptCompleted( + "Quiet.Harbor.2021.1080p.BluRay.x264-GRP", + "Quiet.Harbor.2021.1080p.BluRay.x264-GRP.mkv", + SIZE + 3000, + ); + await planImport(h.d, { downloadTaskId: dt2.id, trigger: "auto", actorType: "system" }); + const dec2 = (await decisionOf(`organize_blocked:${dt2.id}`))!; + const rep = await hook.execute(dec2, { optionKey: "replace" }, actor); + expect(rep.status).toBe("running"); + expect((await executeRecord(h.d, rep.recordId as string)).status).toBe("succeeded"); + // 被替换的旧文件进库根回收站(.umt-trash),不在库内 + const inLib = walkFiles(lib.root).filter( + (p) => p.endsWith(".mkv") && p.includes("Quiet") && !p.includes("/.umt-trash/"), + ); + const sizes = inLib.map((p) => statSync(p).size); + expect(sizes).toContain(SIZE + 3000); + expect(sizes).not.toContain(SIZE); + expect(walkFiles(lib.root).some((p) => p.includes("/.umt-trash/") && statSync(p).size === SIZE)).toBe(true); + }); + + it("作品详情下载历史:列出接管的下载任务与整理记录", async () => { + const r = await h.api("GET", `/library/${movieId}`); + expect(r.status).toBe(200); + const ids = r.body.data.downloads.map((d: { downloadTask: { id: string } }) => d.downloadTask.id); + expect(ids).toContain(firstDtId); + const first = r.body.data.downloads.find((d: { downloadTask: { id: string } }) => d.downloadTask.id === firstDtId); + expect(first.downloadTask).toMatchObject({ origin: "adopted", mediaEntityId: movieId }); + expect(first.lastOrganizationStatus).toBe("succeeded"); + expect(r.body.data.organizationRecords.length).toBeGreaterThanOrEqual(3); + const prov = await h.db + .selectFrom("download_task") + .select("media_entity_id") + .where("id", "=", unknownDtId) + .executeTakeFirstOrThrow(); + const r2 = await h.api("GET", `/library/${prov.media_entity_id}`); + expect(r2.body.data.downloads.map((d: { downloadTask: { id: string } }) => d.downloadTask.id)).toContain( + unknownDtId, + ); + }); +}); diff --git a/apps/server/src/modules/acquisition-organize/index.ts b/apps/server/src/modules/acquisition-organize/index.ts index 7ec51daf..8313eb33 100644 --- a/apps/server/src/modules/acquisition-organize/index.ts +++ b/apps/server/src/modules/acquisition-organize/index.ts @@ -1,5 +1,6 @@ import { type EventPayload, EventType } from "@umt/contracts"; import { sql } from "@umt/db"; +import { registerDecisionHook } from "../decision-hooks.ts"; import { registerScheduledJob } from "../schedules/registry.ts"; import type { ServerContext, ServerModule } from "../types.ts"; import { registerAoCapabilities } from "./capabilities.ts"; @@ -15,6 +16,7 @@ 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 { organizeDecisionHook } from "./services/takeover.ts"; import { loadVolumes, usedPercent } from "./services/volumes.ts"; import { errMessage } from "./util.ts"; @@ -66,6 +68,8 @@ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): Se }, registerRoutes(app, ctx) { const d = lazy(ctx); + // 待处理"处理":接管识别失败 / 目标已存在 两类整理事项在关闭前由本模块执行所选处理 + registerDecisionHook(ctx, organizeDecisionHook(d)); registerRuleRoutes(app, d); registerDownloadRoutes(app, d); registerOrganizationRoutes(app, d); diff --git a/apps/server/src/modules/acquisition-organize/ports.ts b/apps/server/src/modules/acquisition-organize/ports.ts index 3b94d883..b878e087 100644 --- a/apps/server/src/modules/acquisition-organize/ports.ts +++ b/apps/server/src/modules/acquisition-organize/ports.ts @@ -15,7 +15,16 @@ import { type TorrentSource, type VerifyMethod, } from "@umt/adapters"; +import type { ActorType } from "@umt/contracts"; import type { DownloaderInstanceRow, MediaServerInstanceRow, SiteRow, SourceReleaseRow } from "@umt/db"; +import type { ParsedRelease } from "@umt/domain"; +import { SourcesRuntime } from "../sources-identity/runtime.ts"; +import { + createEntityFromTmdbCandidate, + enqueueNewEntityRefresh, +} from "../sources-identity/service/entity-from-candidate.ts"; +import { metadataLanguages } from "../sources-identity/service/metadata-rich.ts"; +import { identifyParsed, type NameIdentifyResult } from "../sources-identity/service/resolve.ts"; import type { ServerContext } from "../types.ts"; // 外部依赖端口:默认实现接真实 qB / 站点 / 文件系统 / Emby;测试注入 fake(mock qB、临时目录、固定磁盘用量)。 @@ -58,9 +67,21 @@ export interface AoPorts { diskUsage(path: string): Promise; probe(path: string): Promise; mediaServer(inst: MediaServerInstanceRow): Promise; + /** + * QB-1 接管任务按名称识别作品(sources-identity 的 ID-2 打分;trigger=takeover 允许查 TMDB)。 + * auto 时已绑定 / 新建作品;其余返回候选。 + */ + identify(input: { parsed: ParsedRelease; evidence?: Record }): Promise; + /** 按 TMDB id 找或建本地作品(待处理"按 TMDB 指定"),返回作品 id */ + entityFromTmdb( + ref: { type: "movie" | "tv"; id: string }, + actor: { type: string; id: string | null }, + ): Promise; options: AoOptions; } +export type { NameIdentifyResult }; + export const DEFAULT_OPTIONS: AoOptions = { backgroundLoops: true, noProgressMs: 6 * 3600_000, @@ -90,8 +111,31 @@ export function createPorts( } }; + let sources: SourcesRuntime | null = null; + const rt = () => { + sources ??= new SourcesRuntime(ctx); + return sources; + }; + const ports: AoPorts = { now: () => new Date(), + async identify({ parsed, evidence }) { + const r = await identifyParsed(rt(), parsed, { trigger: "takeover", ...(evidence ? { evidence } : {}) }); + if (r.created && r.mediaEntityId) await enqueueNewEntityRefresh(ctx, r.mediaEntityId); + return r; + }, + async entityFromTmdb(ref, actor) { + const type: ActorType = (["user", "agent", "system", "api_key"] as const).find((t) => t === actor.type) ?? "user"; + const r = await createEntityFromTmdbCandidate( + ctx.db, + () => rt().tmdb(), + { kind: ref.type, externalId: ref.id }, + { type, id: actor.id }, + { languages: await metadataLanguages(ctx) }, + ); + if (r.created) await enqueueNewEntityRefresh(ctx, r.mediaEntityId); + return r.mediaEntityId; + }, async downloader(inst) { const stamp = `${inst.updated_at.toISOString()}|${inst.base_url}`; const hit = clients.get(inst.id); diff --git a/apps/server/src/modules/acquisition-organize/services/common.ts b/apps/server/src/modules/acquisition-organize/services/common.ts index 3363c003..a676f45a 100644 --- a/apps/server/src/modules/acquisition-organize/services/common.ts +++ b/apps/server/src/modules/acquisition-organize/services/common.ts @@ -26,7 +26,7 @@ export interface DecisionInput { /** 同一 dedupeKey 同时只有一个未关闭条目;重复写入只累加 attempts、刷新证据 */ dedupeKey: string; evidence?: Record; - options?: { key: string; label: string; description?: string }[]; + options?: { key: string; label: string; description?: string; input?: Record }[]; taskId?: string | null; } @@ -38,6 +38,10 @@ export async function upsertDecision(d: AoDeps, exec: Exec, input: DecisionInput ${jsonb(input.evidence)}::jsonb, ${JSON.stringify(input.options ?? [])}::jsonb, ${input.taskId ?? null}) on conflict (dedupe_key) where status in ('open', 'investigating', 'awaiting_user') do update set attempts = decision.attempts + 1, evidence = excluded.evidence, summary = excluded.summary, + subject_type = coalesce(excluded.subject_type, decision.subject_type), + subject_id = coalesce(excluded.subject_id, decision.subject_id), + -- 重新判定时选项随最新证据刷新(如接管识别的候选);没给选项的保留原有(如 Agent 补的"重新校验") + options = case when jsonb_array_length(excluded.options) > 0 then excluded.options else decision.options end, updated_at = now() returning id, (xmax = 0) as inserted`.execute(exec); const row = r.rows[0]!; diff --git a/apps/server/src/modules/acquisition-organize/services/organize.ts b/apps/server/src/modules/acquisition-organize/services/organize.ts index 0f8483df..e2b776bc 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize.ts @@ -46,6 +46,16 @@ import { PathGuardError } from "./organize-guard.ts"; import { type FrozenTransfer, finalizeTransfer, planTransfer, type TransferSpec } from "./organize-transfer.ts"; import { hnrSatisfied } from "./retention.ts"; import { slotComplete as sharedSlotComplete, subscriptionLibraryRoot } from "./subscriptions.ts"; +import { + CONFLICT_OPTIONS, + downloadTasksUnder, + identifyTakeover, + identityOptions, + keepBothTargets, + type QbFileLike, + type TakeoverIdentify, + takeoverSummary, +} from "./takeover.ts"; // 入库:冻结整理计划(organization_record + 文件明细)→ 逐文件可恢复执行(操作账本 fencing)→ // 资产/位置/路径绑定同事务登记 → Slot 推进 → Move 后移除 qB 任务记录(不删数据)。 @@ -70,6 +80,8 @@ export interface ImportRequest { onExisting?: ExistingPolicy | undefined; /** 已获替换授权(审批/手动确认) */ grantReplace?: boolean | undefined; + /** QB-1 目标已存在时"保留两者":新文件(连同附属文件)改名加序号放在旁边,已有文件不动 */ + keepBoth?: boolean | undefined; /** * W5-A2 逐文件覆写(文件管理整理向导的人工确认,优先于文件名解析):键为源文件或目录的绝对路径 * (目录对其下文件生效,最长前缀优先)。season/episodes 决定剧集季目录与命名以及集单元覆盖; @@ -312,6 +324,7 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise ({ name: f.path, size: f.size })); } if (list.length === 0) return fail("file_list_unavailable"); + qbFiles = list; sourceRoot = dt.content_path ? hostPathOf(topo, dt.content_path) : null; for (const f of list) { if (f.priority === 0) continue; @@ -389,16 +403,41 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise c.kind === "hnr_unsatisfied" || c.kind === "hnr_unknown"); + const existsConflict = blocking.some((c) => c.kind === "target_exists"); await upsertDecision(d, trx, { kind: hnrConflict ? "approval_required" : "path_conflict", title: hnrConflict ? `做种义务未满足，已阻止移动：${entity.title}` : `整理冲突：${entity.title}`, @@ -718,13 +766,15 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise f.source_path)].filter((x): x is string => !!x)), + ]; + for (const t of await downloadTasksUnder(trx, srcs)) { + await closeOrganizeDecisions(d, trx, t.id, rec); + await trx + .updateTable("download_task") + .set({ media_entity_id: frozen.entityId }) + .where("id", "=", t.id) + .where("media_entity_id", "is", null) + .execute(); + } } // 手动整理指定的 Slot(文件管理整理向导):核验通过后以新资产满足该 Slot(获取中的 Slot 不动,由其计划推进) if (!dt && frozen.slotId && okAny) { @@ -1654,7 +1726,7 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise best.size) best = f; + } + return best?.name ?? null; +} + +/** + * 种子名 + 主视频文件名 → 解析结果。主视频路径(qB 文件名含种子根目录)按 文件 → 上级 → 上上级 目录合并; + * 种子名作为再上一级兜底(补标题 / 年份 / 季);文件名解析不出标题时直接用种子名。 + */ +export function takeoverParsed(torrentName: string, mainFile: string | null): ParsedRelease { + const byName = parseRelease(torrentName, { isFile: fileKindOf(torrentName) !== undefined }); + if (!mainFile) return byName; + const byPath = parseReleasePath(mainFile); + if (!byPath.title || byPath.auxiliaryOnly) return byName; + // 多文件种子:根目录就是种子名,parseReleasePath 已合并过;单文件种子文件名≈种子名,再合并一次无害 + return mergeParsed(byPath, byName, "parent_dir"); +} + +export interface TakeoverIdentify { + entityId: string | null; + mainFile: string | null; + parsed: ParsedRelease; + result: NameIdentifyResult; +} + +/** 识别接管任务;识别到作品时回填 download_task.media_entity_id */ +export async function identifyTakeover( + d: AoDeps, + dt: DownloadTaskRow, + files: readonly QbFileLike[], +): Promise { + const mainFile = pickMainVideo(files); + const parsed = takeoverParsed(dt.name, mainFile); + const result = await d.ports.identify({ + parsed, + evidence: { downloadTaskId: dt.id, torrentName: dt.name, mainFile }, + }); + const entityId = result.outcome === "auto" ? result.mediaEntityId : null; + if (entityId) + await d.ctx.db + .updateTable("download_task") + .set({ media_entity_id: entityId }) + .where("id", "=", dt.id) + .where("media_entity_id", "is", null) + .execute(); + return { entityId, mainFile, parsed, result }; +} + +export interface DecisionOptionInput { + key: string; + label: string; + description?: string; + input?: Record; +} + +const OUTCOME_TEXT: Record = { + auto: "已识别", + proposed: "识别到候选，但自动化未开启，需要确认", + ambiguous: "有多个相近的作品，无法自动确定", + no_match: "本地与 TMDB 都没有匹配的作品", + lookup_failed: "查询 TMDB 失败", + unparseable: "文件名解析不出标题", +}; + +export function takeoverSummary(t: TakeoverIdentify | null): string { + if (!t) return "下载内容未关联作品。识别或指定作品后再整理"; + const what = [t.parsed.title, t.parsed.year].filter(Boolean).join(" "); + return `按文件名识别${what ? `「${what}」` : ""}：${OUTCOME_TEXT[t.result.outcome]}。选择作品、作为临时作品入库，或忽略`; +} + +/** 接管识别失败 / 歧义的事项选项:候选(最多 3 个)、选择作品(本地 / TMDB)、作为临时作品入库、忽略 */ +export function identityOptions(t: TakeoverIdentify | null, fallbackTitle: string): DecisionOptionInput[] { + const out: DecisionOptionInput[] = []; + for (const [i, c] of (t?.result.candidates ?? []).slice(0, 3).entries()) { + const target = c.localEntityId + ? { mediaEntityId: c.localEntityId } + : c.provider === "tmdb" && (c.entityType === "movie" || c.entityType === "tv") + ? { tmdb: { type: c.entityType, id: c.externalId } } + : null; + if (!target) continue; + out.push({ + key: `candidate_${i}`, + label: `选择：${c.title}${c.year ? `（${c.year}）` : ""}`, + description: c.localEntityId ? "本地已有作品" : `TMDB ${c.entityType}/${c.externalId}，本地没有时自动建档`, + input: { candidate: target }, + }); + } + out.push( + { + key: "select_entity", + label: "选择作品（本地）", + description: "在本地作品里搜索并指定", + input: { requires: "mediaEntityId" }, + }, + { + key: "select_tmdb", + label: "选择作品（TMDB）", + description: "按 TMDB ID 指定，本地没有时自动建档", + input: { requires: "tmdb" }, + }, + ); + const title = t?.parsed.title ?? fallbackTitle; + if (title) { + const tvLike = + t?.parsed.mediaTypeHint === "tv" || t?.parsed.season !== undefined || t?.parsed.episodes !== undefined; + out.push({ + key: "provisional", + label: "作为临时作品入库", + description: "按文件名建本地临时作品后整理，之后可合并或补外部身份", + input: { provisional: { kind: tvLike ? "series" : "movie", title, year: t?.parsed.year ?? null } }, + }); + } + out.push({ key: "dismiss", label: "忽略" }); + return out; +} + +/** 目标已存在(target_exists)冲突的事项选项 */ +export const CONFLICT_OPTIONS: DecisionOptionInput[] = [ + { key: "replace", label: "替换", description: "新文件核验通过后替换已有文件，旧文件进回收站" }, + { key: "keep_both", label: "保留两者", description: "新文件改名（加序号）后放在旁边，已有文件不动" }, + { key: "skip", label: "跳过", description: "已存在的文件不整理，其余文件照常入库" }, +]; + +// ── 保留两者:把目标已存在的主视频(连同附属文件)改名为 "<名称> (n)" ── + +function withSuffix(p: string, stem: string, n: number): string { + const dir = dirname(p); + const name = basename(p); + if (name.startsWith(stem)) return join(dir, `${stem} (${n})${name.slice(stem.length)}`); + const ext = extname(name); + return join(dir, `${name.slice(0, name.length - ext.length)} (${n})${ext}`); +} + +export async function keepBothTargets(plan: OrganizePlan): Promise { + const held = plan.conflicts.filter((c) => c.kind === "target_exists" && c.blocking && c.src); + if (held.length === 0) return; + const mains = new Set(); + for (const c of held) { + const f = plan.files.find((x) => x.src === c.src); + if (f) mains.add(f.role === "main" ? f.src : (f.mainSrc ?? f.src)); + } + const taken = new Set(plan.files.map((f) => f.dst)); + const resolved = new Set(); + for (const m of mains) { + const group = plan.files.filter((f) => f.src === m || f.mainSrc === m); + const main = group.find((f) => f.src === m) ?? group[0]!; + const ext = extname(main.dst); + const stem = basename(main.dst).slice(0, basename(main.dst).length - ext.length); + for (let n = 2; n < 100; n++) { + const alts = group.map((f) => withSuffix(f.dst, stem, n)); + let free = true; + for (const a of alts) { + if (taken.has(a) || (await statSafe(a)).kind !== "missing") { + free = false; + break; + } + } + if (!free) continue; + group.forEach((f, i) => { + taken.add(alts[i]!); + f.dst = alts[i]!; + if (f.action === "hold") f.action = "transfer"; + resolved.add(f.src); + }); + break; + } + } + plan.conflicts = plan.conflicts.filter((c) => !(c.kind === "target_exists" && c.src && resolved.has(c.src))); + for (const pc of plan.preconditions) + if (pc.kind === "target_absent") pc.paths = plan.files.filter((f) => f.action === "transfer").map((f) => f.dst); + plan.requiresApproval = plan.conflicts.some((c) => c.blocking); +} + +// ── 手动整理(待处理里选了处理方式):冻结计划 + 入队执行 ── + +export async function startManualImport( + d: AoDeps, + req: Omit, + actor: DecisionHookActor, + via: string, +): Promise<{ recordId: string | null; status: string; reason: string | null; taskId: string }> { + const db = d.ctx.db; + const idem = `${via}:${req.downloadTaskId ?? "x"}:${randomUUID()}`; + const by = actor.id ? `${actor.type}:${actor.id}` : actor.type; + const taskId = await ensureTaskRow(db, { + kind: "verifyAndImport", + idempotencyKey: idem, + input: { ...req, via }, + subjectType: req.downloadTaskId ? "download_task" : "organize_request", + subjectId: req.downloadTaskId ?? null, + triggeredBy: by, + }); + const actorType = actor.type === "api_key" || actor.type === "agent" || actor.type === "system" ? actor.type : "user"; + const plan = await planImport(d, { + ...req, + trigger: "manual", + actorType, + actorId: by, + taskId, + }); + if (plan.status === "running" || plan.status === "queued") { + const { runId } = await d.ctx.engine.enqueue( + "verifyAndImport", + { recordId: plan.recordId, downloadTaskId: req.downloadTaskId, taskId }, + { idempotencyKey: `verifyAndImport:${idem}`, traceId: taskId }, + ); + await db.updateTable("task").set({ engine_run_id: runId }).where("id", "=", taskId).execute(); + } else { + await markTask(db, taskId, "failed", { error: plan.reason ?? plan.status }); + } + return { recordId: plan.recordId, status: plan.status, reason: plan.reason, taskId }; +} + +// ── 待处理执行钩子:organize_identity / organize_blocked ── + +class HookError extends Error { + constructor( + readonly code: "bad_request" | "conflict" | "not_found", + message: string, + ) { + super(message); + } +} + +export { HookError as OrganizeDecisionError }; + +async function createProvisional( + exec: Exec, + p: { kind: "movie" | "series"; title: string; year: number | null }, +): Promise { + const e = await exec + .insertInto("media_entity") + .values({ + kind: p.kind, + title: p.title, + original_title: null, + year: p.year, + status: "provisional", + metadata: jsonb({}), + }) + .returning(["id"]) + .executeTakeFirstOrThrow(); + await addAliasIfAbsent(exec, e.id, p.title, "other", "user"); + return e.id; +} + +async function chosenEntity( + d: AoDeps, + body: DecisionResolveBody, + opt: DecisionOptionInput | undefined, + actor: DecisionHookActor, +): Promise { + const input = (opt?.input ?? {}) as { + requires?: string; + candidate?: { mediaEntityId?: string; tmdb?: { type: "movie" | "tv"; id: string } }; + provisional?: { kind: "movie" | "series"; title: string; year: number | null }; + }; + const pick = + input.requires === "mediaEntityId" + ? { mediaEntityId: body.identity?.mediaEntityId } + : input.requires === "tmdb" + ? { tmdb: body.identity?.tmdb } + : (input.candidate ?? null); + if (input.provisional) return createProvisional(d.ctx.db, input.provisional); + if (pick?.mediaEntityId) { + const e = await d.ctx.db + .selectFrom("media_entity") + .select(["id", "status", "merged_into_id"]) + .where("id", "=", pick.mediaEntityId) + .executeTakeFirst(); + if (!e) throw new HookError("not_found", "作品不存在"); + return e.status === "merged" && e.merged_into_id ? e.merged_into_id : e.id; + } + if (pick?.tmdb) return d.ports.entityFromTmdb(pick.tmdb, actor); + throw new HookError("bad_request", input.requires ? `需要 identity.${input.requires}` : "该选项没有指定作品"); +} + +export function organizeDecisionHook(d: AoDeps): DecisionHook { + return { + match: (row) => row.dedupe_key.startsWith("organize_identity:") || row.dedupe_key.startsWith("organize_blocked:"), + async execute(row, body, actor) { + const dismissing = body.dismiss === true || body.optionKey === "dismiss"; + if (dismissing) return { kind: "none", reason: "dismissed" }; + const options = (row.options ?? []) as DecisionOptionInput[]; + const opt = options.find((o) => o.key === body.optionKey); + const ev = (row.evidence ?? {}) as { downloadTaskId?: string | null }; + const dtId = ev.downloadTaskId ?? row.dedupe_key.split(":")[1] ?? null; + const dt = dtId + ? await d.ctx.db.selectFrom("download_task").selectAll().where("id", "=", dtId).executeTakeFirst() + : undefined; + if (row.dedupe_key.startsWith("organize_identity:")) { + if (!dt) throw new HookError("not_found", "关联的下载任务不存在"); + const entityId = await chosenEntity(d, body, opt, actor); + await d.ctx.db + .updateTable("download_task") + .set({ media_entity_id: entityId }) + .where("id", "=", dt.id) + .execute(); + const r = await startManualImport(d, { downloadTaskId: dt.id, mediaEntityId: entityId }, actor, "decision"); + return { kind: "organize", mediaEntityId: entityId, ...r }; + } + // organize_blocked:目标已存在 → 替换 / 保留两者 / 跳过 + const key = body.optionKey; + if (key !== "replace" && key !== "keep_both" && key !== "skip") return { kind: "none", reason: `option_${key}` }; + if (!dt) throw new HookError("not_found", "关联的下载任务不存在"); + const rec = await d.ctx.db + .selectFrom("organization_record") + .select(["id", "media_entity_id", "target_library_root_id"]) + .where("download_task_id", "=", dt.id) + .orderBy("created_at", "desc") + .executeTakeFirst(); + const r = await startManualImport( + d, + { + downloadTaskId: dt.id, + ...(rec?.media_entity_id ? { mediaEntityId: rec.media_entity_id } : {}), + ...(rec?.target_library_root_id ? { targetLibraryRootId: rec.target_library_root_id } : {}), + // 操作方式不沿用原记录:仍按策略 / 设置决定(Move 的授权与做种义务照常把关) + ...(key === "replace" + ? { onExisting: "replace_with_verify" as const, grantReplace: true } + : key === "skip" + ? { onExisting: "skip" as const } + : { keepBoth: true }), + }, + actor, + `decision-${key}`, + ); + if (r.status !== "running" && r.status !== "queued") + throw new HookError("conflict", `重新整理仍未能执行：${r.reason ?? r.status}`); + return { kind: "organize", choice: key, ...r }; + }, + }; +} + +/** 已登记下载任务里,内容路径(本机视图)与这些源路径重叠的(手动按路径整理时找回对应下载任务) */ +export async function downloadTasksUnder(exec: Exec, paths: readonly string[]): Promise { + if (paths.length === 0) return []; + const topos = await loadInstances(exec, { enabledOnly: false }); + const rows = await exec + .selectFrom("download_task") + .selectAll() + .where("removed_at", "is", null) + .where("content_path", "is not", null) + .execute(); + const out: DownloadTaskRow[] = []; + for (const r of rows) { + const topo = topos.find((t) => t.inst.id === r.instance_id); + const host = topo && r.content_path ? hostPathOf(topo, r.content_path) : null; + if (!host) continue; + const root = host.replace(/\/+$/, ""); + if (paths.some((p) => p === root || p.startsWith(`${root}/`) || root.startsWith(`${p.replace(/\/+$/, "")}/`))) + out.push(r); + } + return out; +} diff --git a/apps/server/src/modules/decision-hooks.ts b/apps/server/src/modules/decision-hooks.ts new file mode 100644 index 00000000..abab4e05 --- /dev/null +++ b/apps/server/src/modules/decision-hooks.ts @@ -0,0 +1,38 @@ +import type { DecisionRow } from "@umt/db"; +import type { ServerContext } from "./types.ts"; + +// 待处理"处理"端点(sources-identity routes/decisions.ts)的执行钩子:其它模块按 dedupe_key 认领自己的事项, +// 在事项关闭前同步执行所选处理(失败抛错 → 事项保持打开,错误回给用户)。按 ctx 隔离(测试里多个装配互不影响)。 + +export interface DecisionResolveBody { + optionKey?: string | undefined; + dismiss?: boolean | undefined; + note?: string | undefined; + identity?: + | { mediaEntityId?: string | undefined; tmdb?: { type: "movie" | "tv"; id: string } | undefined } + | undefined; +} + +export interface DecisionHookActor { + type: string; + id: string | null; +} + +export interface DecisionHook { + /** 认领该事项(通常按 dedupe_key 前缀) */ + match(d: DecisionRow): boolean; + /** 执行所选处理;返回写进 resolution.execution 的摘要 */ + execute(d: DecisionRow, body: DecisionResolveBody, actor: DecisionHookActor): Promise>; +} + +const hooks = new WeakMap(); + +export function registerDecisionHook(ctx: ServerContext, hook: DecisionHook): void { + const list = hooks.get(ctx) ?? []; + list.push(hook); + hooks.set(ctx, list); +} + +export function decisionHookFor(ctx: ServerContext, d: DecisionRow): DecisionHook | undefined { + return hooks.get(ctx)?.find((h) => h.match(d)); +} diff --git a/apps/server/src/modules/sources-identity/routes/decisions.ts b/apps/server/src/modules/sources-identity/routes/decisions.ts index 273ba903..91538f88 100644 --- a/apps/server/src/modules/sources-identity/routes/decisions.ts +++ b/apps/server/src/modules/sources-identity/routes/decisions.ts @@ -3,6 +3,7 @@ import { type Decision, DecisionKind, DecisionStatus, Id, type PoolCandidate } f import { sql } from "@umt/db"; import { z } from "zod"; import { ApiException } from "../../../http/errors.ts"; +import { decisionHookFor } from "../../decision-hooks.ts"; import type { ModuleApp } from "../../types.ts"; import { D, DecisionMinSeverity } from "../defs.ts"; import { route } from "../http.ts"; @@ -223,6 +224,19 @@ export function decisionRoutes(app: ModuleApp, rt: () => SourcesRuntime) { { idempotencyKey: `refreshSiteAccount:decision:${d.id}:${Date.now()}`, traceId: randomUUID() }, ); } + // 其它模块认领的事项(如整理:接管识别 / 目标已存在):关闭前同步执行,失败则事项保持打开 + let execution: Record | null = null; + const hook = decisionHookFor(rt().ctx, d); + if (hook) { + try { + execution = await hook.execute(d, body, { type: actor.type, id: actor.id ?? null }); + } catch (e) { + const code = (e as { code?: string }).code; + if (code === "bad_request" || code === "conflict" || code === "not_found") + throw new ApiException(code, (e as Error).message); + throw e; + } + } // 上面的执行路径已按 dedupeKey 关闭;其余类型在这里记录处理结果 const row = await db() .transaction() @@ -239,7 +253,12 @@ export function decisionRoutes(app: ModuleApp, rt: () => SourcesRuntime) { .updateTable("decision") .set({ status, - resolution: jsonb({ optionKey: body.optionKey ?? null, note: body.note ?? null, by }), + resolution: jsonb({ + optionKey: body.optionKey ?? null, + note: body.note ?? null, + by, + ...(execution ? { execution } : {}), + }), resolved_by: by, resolved_at: new Date(), }) diff --git a/apps/server/src/modules/sources-identity/service/resolve.ts b/apps/server/src/modules/sources-identity/service/resolve.ts index d68377e0..0f4ea625 100644 --- a/apps/server/src/modules/sources-identity/service/resolve.ts +++ b/apps/server/src/modules/sources-identity/service/resolve.ts @@ -65,7 +65,11 @@ export interface ResolveInput { trigger?: ResolveTrigger | undefined; } -export type ResolveTrigger = "observed" | "review" | "manual" | "search"; +/** + * takeover = 接管的下载任务(用户手动加进 qB、没有站点资源)完成后按种子名 / 主视频文件名识别(QB-1,对照 MP transfer + * 无下载历史时按文件名 MetaInfoPath 识别):视同用户动作,允许查 TMDB。 + */ +export type ResolveTrigger = "observed" | "review" | "manual" | "search" | "takeover"; export interface ResolveResult { outcome: @@ -118,10 +122,10 @@ function poolCandidate(r: RankedCandidate): PoolCandidate { async function localCandidates( exec: Exec, parsed: ParsedRelease, - release: SourceReleaseRow, + release: SourceReleaseRow | null, ): Promise { const ids = new Set(); - const explicit = [...parsed.explicitIds, ...siteExplicitIds(release.raw)]; + const explicit = [...parsed.explicitIds, ...(release ? siteExplicitIds(release.raw) : [])]; for (const x of explicit) { const row = await exec .selectFrom("external_identity") @@ -408,7 +412,14 @@ export async function resolveRelease(rt: SourcesRuntime, input: ResolveInput): P } /** 为什么要查 TMDB;null = 不查(惰性识别) */ -export type LookupReason = "tracked_alias" | "manual" | "search" | "previously_looked_up" | "eager" | "new_title_rule"; +export type LookupReason = + | "tracked_alias" + | "manual" + | "search" + | "takeover" + | "previously_looked_up" + | "eager" + | "new_title_rule"; async function lookupReason( rt: SourcesRuntime, @@ -419,6 +430,7 @@ async function lookupReason( ): Promise { if (trigger === "manual") return "manual"; if (trigger === "search") return "search"; + if (trigger === "takeover") return "takeover"; if (tracked) return "tracked_alias"; // 池条目此前已查过 TMDB(旧行为 / 手动 / 规则触发):复查延续,仍受识别缓存与有界退避约束 if (trigger === "review" && poolReason && poolReason !== "unmatched") return "previously_looked_up"; @@ -820,3 +832,170 @@ async function toPool( } export { recordAttempt }; + +// ── 按名称识别(没有站点资源行):接管的下载任务完成后用 ── + +export interface NameIdentifyCandidate { + provider: string; + entityType: string; + externalId: string; + title: string; + year: number | null; + localEntityId: string | null; + level: string; +} + +export interface NameIdentifyResult { + outcome: "auto" | "proposed" | "ambiguous" | "no_match" | "lookup_failed" | "unparseable"; + mediaEntityId: string | null; + contentUnitIds: string[]; + reason: string; + /** 候选(歧义 / 待确认时给用户挑;最多 5 个) */ + candidates: NameIdentifyCandidate[]; + /** 本次是否新建了作品 */ + created: boolean; +} + +/** + * 按已解析的名称识别作品(ID-2 同一套打分:本地召回 + TMDB 候选 → resolveIdentity)。不写待识别池、不开待处理, + * 由调用方决定失败后怎么办。auto 且自动化开启时绑定 / 新建作品并确保季集单元;其余只返回候选。 + */ +export async function identifyParsed( + rt: SourcesRuntime, + parsed: ParsedRelease, + opts: { trigger?: ResolveTrigger; evidence?: Record } = {}, +): Promise { + const { ctx } = rt; + const none = { mediaEntityId: null, contentUnitIds: [] as string[], candidates: [], created: false }; + if (!parsed.title || parsed.auxiliaryOnly) return { outcome: "unparseable", reason: "unparseable_title", ...none }; + const trigger = opts.trigger ?? "takeover"; + const local = await localCandidates(ctx.db, parsed, null); + const lookup = await lookupReason(rt, parsed, trigger, local.length > 0, null); + let remote: MetadataCandidate[] = []; + let lookupError: string | null = null; + let recognize: { cacheHit: boolean; meta: RecognizeMeta | null } = { cacheHit: false, meta: null }; + if (lookup) { + try { + const r = await tmdbCandidates(rt, parsed, true); + remote = r.remote; + recognize = r; + } catch (e) { + if (!(e instanceof TmdbError)) throw e; + lookupError = `${e.kind}: ${e.message}`; + } + } + const resolution = resolveIdentity(parsed, [...local, ...remote.map(tmdbToCandidate)]); + if (recognize.meta && !recognize.cacheHit && !lookupError) { + const top = resolution.status === "auto" ? resolution.best?.candidate : undefined; + const hit = + top?.provider === "tmdb" + ? remote.find((m) => m.externalId === top.externalId && m.entityType === top.entityType) + : undefined; + const info = hit ?? (resolution.status === "no_match" ? null : undefined); + await saveRecognizeCache(rt, recognize.meta, info); + } + const candidates = resolution.ranked.slice(0, 5).map((r) => { + const c = poolCandidate(r); + return { + provider: c.provider, + entityType: c.entityType, + externalId: c.externalId, + title: c.title, + year: c.year, + localEntityId: c.localEntityId, + level: String(c.level), + }; + }); + const automation = await ctx.settings.get("automation"); + const autoAccept = automation.enabled && !automation.readOnlyReconcile; + if (resolution.status === "auto" && resolution.best && autoAccept) { + const best = resolution.best; + const level: EvidenceLevel = toContractEvidenceLevel(best.level) ?? "needs_review"; + let detail = + best.candidate.provider === "tmdb" + ? remote.find((m) => m.externalId === best.candidate.externalId && m.entityType === best.candidate.entityType) + : undefined; + if (detail && !detail.detailed) { + const existing = await entityByExternal(ctx.db, { + provider: "tmdb", + entityType: detail.entityType, + externalId: detail.externalId, + }); + if (!existing) { + try { + detail = await (await rt.tmdb())!.details(detail.entityType, detail.externalId); + } catch (e) { + if (!(e instanceof TmdbError)) throw e; + } + } + } + for (let attempt = 0; ; attempt++) { + try { + const bound = await ctx.db.transaction().execute(async (trx) => { + let entityId: string | null = null; + let created = false; + if (best.candidate.localEntityId) entityId = await resolveEntityId(trx, best.candidate.localEntityId); + if (!entityId && best.candidate.provider === "tmdb") { + entityId = await entityByExternal(trx, { + provider: "tmdb", + entityType: best.candidate.entityType, + externalId: best.candidate.externalId, + }); + if (!entityId && detail) { + const r = await createEntityFromTmdb(trx, detail, { + level, + evidence: { + ...(opts.evidence ?? {}), + trigger, + rules: best.evidence, + titleMatch: best.titleMatch, + yearMatch: best.yearMatch, + }, + actor: SYSTEM, + variety: looksLikeVariety(parsed, null), + }); + entityId = r.entityId; + created = r.created; + } + } + if (!entityId) return null; + const entity = await trx + .selectFrom("media_entity") + .select(["id", "kind"]) + .where("id", "=", entityId) + .executeTakeFirstOrThrow(); + const units = await ensureUnits(trx, entityId, entity.kind, parsed); + return { entityId, created, unitIds: units.unitIds }; + }); + if (!bound) break; + return { + outcome: "auto", + mediaEntityId: bound.entityId, + contentUnitIds: bound.unitIds, + reason: resolution.reason, + candidates, + created: bound.created, + }; + } catch (e) { + if (e instanceof IdentityConflictError && attempt < 2) continue; + throw e; + } + } + } + const outcome: NameIdentifyResult["outcome"] = + resolution.status === "auto" + ? "proposed" + : resolution.status === "ambiguous" + ? "ambiguous" + : lookupError + ? "lookup_failed" + : "no_match"; + return { + outcome, + mediaEntityId: null, + contentUnitIds: [], + reason: lookupError ?? (resolution.status === "auto" && !autoAccept ? "automation_off" : resolution.reason), + candidates, + created: false, + }; +} diff --git a/apps/server/test/sources-identity/lazy-identity.test.ts b/apps/server/test/sources-identity/lazy-identity.test.ts index 3941c8c4..3d279281 100644 --- a/apps/server/test/sources-identity/lazy-identity.test.ts +++ b/apps/server/test/sources-identity/lazy-identity.test.ts @@ -1,9 +1,9 @@ import { sql } from "@umt/db"; -import { normalizeTitle } from "@umt/domain"; +import { normalizeTitle, parseReleasePath } from "@umt/domain"; import { createMockTmdb, type MockTmdb, TMDB_FIXTURES, type TmdbFixture } from "@umt/testkit"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { down, up } from "../../../../packages/db/migrations/0047_identity_pool.ts"; -import { resolveRelease } from "../../src/modules/sources-identity/service/resolve.ts"; +import { identifyParsed, resolveRelease } from "../../src/modules/sources-identity/service/resolve.ts"; import { createHarness, type Harness, json, pgAvailable, seedSite } from "./harness.ts"; // DEC-1 惰性识别(对照 MP:RSS / 站点资源只在匹配订阅或用户搜索时识别,识别失败不告警)与待识别池治理。 @@ -37,6 +37,8 @@ describe.skipIf(!pg)("DEC-1 惰性识别 + 待识别池", () => { twin(112), twin(211, "Nightfall", "夜幕", "1995-03-01"), twin(212, "Nightfall", "夜幕", "1995-03-01"), + twin(311, "Harborlight", "港灯", "2003-05-01"), + twin(312, "Harborlight", "港灯", "2003-05-01"), ], }); h = await createHarness("lazyid", { tmdbUrl: tmdb.url }); @@ -271,6 +273,44 @@ describe.skipIf(!pg)("DEC-1 惰性识别 + 待识别池", () => { .executeTakeFirstOrThrow(); expect(p).toMatchObject({ status: "archived", next_review_at: null }); }); + + it("QB-1 接管任务按文件名识别(trigger=takeover):不需要本地别名也查 TMDB,auto 建档;歧义只返回候选,不进池不开待处理", async () => { + const calls = tmdbCalls(); + const parsed = parseReleasePath( + "Dune.Part.Two.2024.2160p.UHD.BluRay.x265-GRP/Dune.Part.Two.2024.2160p.UHD.BluRay.x265-GRP.mkv", + ); + const r = await identifyParsed(h.rt, parsed, { trigger: "takeover" }); + expect(tmdbCalls()).toBeGreaterThan(calls); + expect(r.outcome).toBe("auto"); + expect(r.mediaEntityId).toBeTruthy(); + const ent = await h.db + .selectFrom("media_entity") + .selectAll() + .where("id", "=", r.mediaEntityId!) + .executeTakeFirstOrThrow(); + expect(ent.year).toBe(2024); + // 同名两部 → 歧义:只给候选 + const pool = await h.db + .selectFrom("identity_candidate") + .select((eb) => eb.fn.countAll().as("n")) + .executeTakeFirstOrThrow(); + const amb = await identifyParsed(h.rt, parseReleasePath("Harborlight.2003.1080p.BluRay.mkv"), { + trigger: "takeover", + }); + expect(amb.outcome).toBe("ambiguous"); + expect(amb.candidates.map((c) => c.externalId).sort()).toEqual(["311", "312"]); + expect(amb.mediaEntityId).toBeNull(); + const pool2 = await h.db + .selectFrom("identity_candidate") + .select((eb) => eb.fn.countAll().as("n")) + .executeTakeFirstOrThrow(); + expect(pool2.n).toBe(pool.n); + // 默认 observed 触发:不相关资源不查 TMDB(惰性识别不变) + const before = tmdbCalls(); + const lazy = await identifyParsed(h.rt, parseReleasePath("Zqxv.Kxlmn.2011.720p.mkv"), { trigger: "observed" }); + expect(lazy.outcome).toBe("no_match"); + expect(tmdbCalls()).toBe(before); + }); }); describe.skipIf(!pg)("迁移 0047:存量 identity_unresolved 清理", () => { diff --git a/docs/runbook/files.md b/docs/runbook/files.md index bb788d17..793c0a13 100644 --- a/docs/runbook/files.md +++ b/docs/runbook/files.md @@ -79,3 +79,15 @@ - NFO/图片投影(scrape 的 `writeNfo/images`)、硬链接、蓝光原盘整目录操作仍未做。 - 搬运 move 在终态提交前删源侧空目录;若进程恰在两者之间退出,重跑 finalize 幂等,但极端情况下可能留下空目录(无数据风险)。 - 路径守卫未下沉到 `packages/adapters/src/fs/guard.ts`(本模块自带按根解析的守卫,整理仍用 `sourceGuard`)。 + +## 整理事项闭环(QB-1) + +- 整理成功(自动重试或手动,有资产入库)后,同一下载任务的 `organize_identity:`(无法确定作品)与 `organize_blocked:` + (整理冲突 / 做种义务)事项一并关闭,`resolution.key = manual_organize`(自动重试为 `organize_retry`),记整理记录 id。 +- 文件管理按路径手动整理(没有 `downloadTaskId`):按下载器路径映射找内容路径与源路径重叠的下载任务,同样关闭其整理事项, + 并把作品回填到 `download_task.media_entity_id`(作品详情「下载历史」据此列出)。 +- 目标已存在(`target_exists`)的整理冲突事项选项为「替换 / 保留两者 / 跳过」,处理时按原记录的作品与库根重新冻结计划并执行(操作方式仍按策略 / 设置,Move 的做种义务照常把关): + 替换 = `onExisting=replace_with_verify` + 授权(新文件核验后旧文件进库根回收站);保留两者 = 新主视频连同附属文件改名 + `<名称> (n)`(n 从 2 起取第一个不存在的);跳过 = `onExisting=skip`,其余文件照常入库。仍有其它阻塞冲突 → 409,事项保持打开。 +- 作品详情(`GET /library/{id}`):`downloads` 按 计划 Slot / 站点资源 / `download_task.media_entity_id` 归属(接管任务按识别结果、 + 手动指定或整理后回填);`organizationRecords` 列该作品最近 100 条整理记录(含没有下载任务的手动整理)。 diff --git a/docs/runbook/identity.md b/docs/runbook/identity.md index 235769db..7558d3d6 100644 --- a/docs/runbook/identity.md +++ b/docs/runbook/identity.md @@ -16,6 +16,7 @@ |---|---| | `manual` | 任务输入 `trigger: "manual"`:池里点「重新识别」、Agent / 能力 `identity.resolve` | | `search` | 任务输入 `trigger: "search"`:搜索结果 | + | `takeover` | 接管的下载任务按文件名识别(`identifyParsed(…, { trigger: "takeover" })`,见下文「接管任务按文件名识别」) | | `tracked_alias` | 本地召回有候选(标题命中订阅 / 本地作品的别名) | | `previously_looked_up` | 周期复查,且这条池条目以前查过 TMDB(旧数据、手动、规则触发) | | `eager` | 设置 `identity.eagerTmdbLookup = true` | @@ -38,6 +39,24 @@ 搜索模块想让搜索结果直接识别,入队 `resolveObservation` 时传 `trigger: "search"`(目前搜索结果经 `release.observed` 进来, 按 observed 处理;命中订阅别名的照样会查)。 +## 接管任务按文件名识别(QB-1) + +用户手动加进 qB、贴了管理标签的任务(`download_task.origin = adopted`,没有 `source_release` / 计划)下载完成后,`verifyAndImport` +的 `planImport` 先按 MP transfer 的做法识别作品(`acquisition-organize/services/takeover.ts`): + +1. 主视频 = 要下载(priority≠0)的视频文件里最大的一个(跳过 sample);`parseReleasePath(主视频路径)` 合并 文件 → 上级 → 上上级 + 目录名(qB 文件名含种子根目录,相当于 MP `MetaInfoPath`),再用种子名补缺(`mergeParsed`);文件名解析不出标题时直接用种子名。 +2. `identifyParsed`(`service/resolve.ts`):同一套 ID-2 打分(本地召回 + TMDB 候选 → `resolveIdentity`),`trigger: "takeover"` 视为 + 允许查 TMDB;不写待识别池、不开身份事项。auto 且自动化开启 → 关联 / 新建作品(新建的入队一次 full 刷新),回填 + `download_task.media_entity_id`,直接整理。 +3. 失败 / 歧义 / 自动化关闭:整理记录 `identity_unknown`,事项 `organize_identity:<下载任务 id>`(`identity_ambiguous`), + 证据带种子名、主视频、解析出的标题 / 年份 / 季集、识别结果与候选。选项:候选(最多 3 个,本地作品或 TMDB)、 + 「选择作品（本地）」(`requires: mediaEntityId`)、「选择作品（TMDB）」(`requires: tmdb`,本地没有时建档)、 + 「作为临时作品入库」(建 `provisional` 作品 + 别名)、「忽略」。 +4. 处理(`POST /decisions/{id}/resolve`):整理模块注册的执行钩子(`modules/decision-hooks.ts`)在关闭前同步执行 —— 回填 + `download_task.media_entity_id` → 冻结手动整理计划 → 入队执行;执行结果记在 `resolution.execution`。缺输入 / 作品不存在 + → 400 / 404,事项保持打开。 + ## 待处理只收相关的 | 结果 | 开事项? |