From 407cb7eb3612a695c1919a82ee868b46933d3b3a Mon Sep 17 00:00:00 2001 From: root Date: Tue, 13 Oct 2026 20:38:06 +0800 Subject: [PATCH] fix(organize): recover legacy requests without dropping one-off intent --- .../services/organize-records.ts | 7 +- .../services/organize-retry.ts | 150 +++++++- .../src/modules/file-manager/organize.ts | 89 ++++- apps/server/src/modules/file-manager/tree.ts | 45 ++- .../organize-identity-recovery.test.ts | 7 +- .../legacy-organize-recovery.test.ts | 350 ++++++++++++++++++ 6 files changed, 616 insertions(+), 32 deletions(-) create mode 100644 apps/server/test/file-manager/legacy-organize-recovery.test.ts diff --git a/apps/server/src/modules/acquisition-organize/services/organize-records.ts b/apps/server/src/modules/acquisition-organize/services/organize-records.ts index 2850426f..c7797dd0 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize-records.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize-records.ts @@ -5,6 +5,7 @@ import { type AoDeps, type Exec, enqueueSweep, timeSlot, upsertDecision } from " import { SWEEP_SLOT_MS } from "./downloads.ts"; import { isHoldReason, ROOT_OFFLINE } from "./organize-exec.ts"; import { finalizeRecord, removeMovedDownload } from "./organize-finalize.ts"; +import { resolveOrganizeRecovery } from "./organize-retry.ts"; // 入库:冻结整理计划(organization_record + 文件明细)→ 逐文件可恢复执行(操作账本 fencing)→ // 资产/位置/路径绑定同事务登记 → Slot 推进 → Move 后移除 qB 任务记录(不删数据)。 @@ -203,7 +204,10 @@ export async function loadRecordsPublic(exec: Exec, rows: OrganizationRecordRow[ message: (err as { message?: string }).message ?? JSON.stringify(err).slice(0, 300), } : null; - return rows.map((r) => { + const recoveries = includeFiles + ? await Promise.all(rows.map(async (r) => (await resolveOrganizeRecovery(exec, r)).recovery)) + : []; + return rows.map((r, index) => { const e = entities.find((x) => x.id === r.media_entity_id); const u = units.find((x) => x.id === r.content_unit_id); const targetRef = { @@ -213,6 +217,7 @@ export async function loadRecordsPublic(exec: Exec, rows: OrganizationRecordRow[ libraryRootId: r.target_library_root_id, }; return { + ...(includeFiles ? { recovery: recoveries[index] } : {}), id: r.id, taskId: r.task_id, downloadTaskId: r.download_task_id, diff --git a/apps/server/src/modules/acquisition-organize/services/organize-retry.ts b/apps/server/src/modules/acquisition-organize/services/organize-retry.ts index 9a00448a..48ae7aa8 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize-retry.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize-retry.ts @@ -1,10 +1,12 @@ import { randomUUID } from "node:crypto"; +import { dirname } from "node:path"; +import { OrganizationRecovery, type OrganizationRecovery as Recovery } from "@umt/contracts"; import { type OrganizationRecordRow, sql } from "@umt/db"; import type { DecisionHookActor } from "../../decision-hooks.ts"; -import { type AoDeps, ensureTaskRow, markTask } from "./common.ts"; +import { type AoDeps, type Exec, ensureTaskRow, hostPathOf, loadInstances, markTask } from "./common.ts"; import { sourceGuard } from "./organize-guard.ts"; import { planImport } from "./organize-plan.ts"; -import type { ImportRequest, StoredImportRequest } from "./organize-types.ts"; +import type { FrozenPlan, ImportRequest, StoredImportRequest } from "./organize-types.ts"; /** 只保留业务参数；路径守卫与任务身份在恢复时重新装配。 */ export function storedImportRequest(req: ImportRequest): StoredImportRequest { @@ -30,6 +32,137 @@ export function organizeRequestFromRecord( return null; } +/** 先找完整请求；冻结结果只提供已知事实，未知的一次性设置由用户确认。 */ +export async function resolveOrganizeRecovery(db: Exec, record: OrganizationRecordRow) { + const snapshot = organizeRequestFromRecord(record); + if (snapshot) return withDownloadSource(db, recoveryResult(snapshot, "request", [])); + const task = record.task_id + ? await db.selectFrom("task").select(["kind", "input"]).where("id", "=", record.task_id).executeTakeFirst() + : undefined; + const input = task?.input as (Partial & { planId?: string; recordId?: string }) | undefined; + // 文件向导的老 task.input 只有分组基本参数；planId 表示缺少完整一次性设置，不能当完整请求。 + if ( + task?.kind === "verifyAndImport" && + input && + !input.planId && + !input.recordId && + (input.downloadTaskId || input.sourcePaths?.length) + ) { + const request = storedImportRequest({ ...input, trigger: "manual", actorType: "system" }); + return withDownloadSource(db, recoveryResult(request, "task", [])); + } + const plan = record.plan as unknown as Partial; + const request: StoredImportRequest = {}; + const missing: Recovery["missing"] = ["options"]; + if (record.download_task_id) request.downloadTaskId = record.download_task_id; + if (record.media_entity_id ?? plan.entityId) request.mediaEntityId = record.media_entity_id ?? plan.entityId; + if (record.content_unit_id) request.contentUnitId = record.content_unit_id; + if (["copy", "move", "in_place"].includes(record.action)) request.action = record.action as ImportRequest["action"]; + if (record.target_library_root_id ?? plan.libraryRootId) + request.targetLibraryRootId = record.target_library_root_id ?? plan.libraryRootId; + if (plan.sourceConfigId) request.sourceConfigId = plan.sourceConfigId; + if (plan.slotId) request.slotId = plan.slotId; + if (plan.naming?.mode) request.namingMode = plan.naming.mode; + if (plan.scrapeAfterImport !== undefined) request.scrapeAfterImport = plan.scrapeAfterImport; + if (plan.replacement) { + request.importMode = "replace"; + request.replacementAssetIds = plan.replacement.assets.map((a) => a.assetId); + } + const paths = [ + ...(plan.organize?.files.filter((f) => f.role === "main").map((f) => f.src) ?? []), + ...(plan.organize?.excluded.filter((f) => f.kind === "video").map((f) => f.src) ?? []), + ]; + if (!paths.length) { + const files = await db + .selectFrom("organization_file_result") + .select("source_path") + .where("organization_record_id", "=", record.id) + .where("role", "=", "main") + .execute(); + paths.push(...files.map((f) => f.source_path)); + } + if (!paths.length && input?.sourcePaths?.length) paths.push(...input.sourcePaths); + if (!paths.length && record.source_path && record.source_path_view === "local") paths.push(record.source_path); + if (!paths.length && record.download_task_id) { + const download = await db + .selectFrom("download_task") + .selectAll() + .where("id", "=", record.download_task_id) + .executeTakeFirst(); + if (download?.content_path) { + const [topology] = await loadInstances(db, { enabledOnly: false, ids: [download.instance_id] }); + const path = topology ? hostPathOf(topology, download.content_path) : null; + if (path) paths.push(path); + } + } + if (!paths.length) { + const decisions = await db + .selectFrom("decision") + .select("evidence") + .where("subject_type", "=", "organization_record") + .where("subject_id", "=", record.id) + .execute(); + for (const decision of decisions) { + const evidence = decision.evidence as { sourcePaths?: string[]; mainFile?: string } | null; + if (evidence?.sourcePaths?.length) paths.push(...evidence.sourcePaths); + else if (evidence?.mainFile?.startsWith("/")) paths.push(evidence.mainFile); + } + } + if (paths.length) request.sourcePaths = [...new Set(paths)]; + if (!request.downloadTaskId && !request.sourcePaths?.length) missing.push("sourcePaths"); + // 实际目标目录可以预填，但不把它冒充原来的 targetDir 选项；确认后才按这次目录执行。 + const dirs = new Set(plan.organize?.files.filter((f) => f.role === "main").map((f) => dirname(f.dst)) ?? []); + if (dirs.size === 1) request.targetDir = [...dirs][0]; + if (plan.episodeStructure) { + request.episodeStructure = { + groupId: plan.episodeStructure.groupId, + layout: null, + appliedToEntity: plan.episodeStructure.appliedToEntity, + }; + if (plan.episodeStructure.temporary) missing.push("episodeStructure"); + } + for (const [path, facts] of Object.entries(plan.facts ?? {})) { + if (!facts.episodeResolution && (facts.season !== undefined || facts.episodes || facts.unitIds?.length)) { + request.fileOverrides ??= {}; + request.fileOverrides[path] = { season: facts.season, episodes: facts.episodes, contentUnitIds: facts.unitIds }; + } + } + return recoveryResult(request, plan.organize ? "plan" : "record", missing); +} + +/** 下载任务原请求不增加 sourcePaths，避免改变下载器的选中文件语义；向导只预填当前已知宿主来源。 */ +async function withDownloadSource(db: Exec, resolved: ReturnType) { + const request = resolved.recovery.request; + if (!request.sourcePaths?.length && request.downloadTaskId) { + const download = await db + .selectFrom("download_task") + .selectAll() + .where("id", "=", request.downloadTaskId) + .executeTakeFirst(); + if (download?.content_path) { + const [topology] = await loadInstances(db, { enabledOnly: false, ids: [download.instance_id] }); + const path = topology ? hostPathOf(topology, download.content_path) : null; + if (path) request.sourcePaths = [path]; + } + } + return resolved; +} + +function recoveryResult(request: StoredImportRequest, source: Recovery["source"], missing: Recovery["missing"]) { + const recovery = OrganizationRecovery.parse({ ready: !missing.length, source, missing, request }); + return { recovery, request }; +} + +export class OrganizeRecoveryError extends Error { + readonly code = "conflict"; + constructor(readonly recovery: Recovery) { + const labels = { sourcePaths: "来源文件", options: "本次整理设置", episodeStructure: "季集结构" }; + super( + `历史整理缺少${recovery.missing.map((key) => labels[key]).join("、")}，已保留可用信息，请在整理向导补齐后继续`, + ); + } +} + interface RetryOptions { mediaEntityId?: string; idempotencyKey?: string; @@ -80,8 +213,17 @@ async function retryLocked( if (existing?.task_id) return { recordId: existing.id, taskId: existing.task_id, status: existing.status, reason: existing.reason }; } - const request = organizeRequestFromRecord(record); - if (!request) throw new Error("整理记录没有可恢复的来源，重新选择来源后整理"); + const resolved = await resolveOrganizeRecovery(d.ctx.db, record); + if (!resolved.recovery.ready) throw new OrganizeRecoveryError(resolved.recovery); + const request = resolved.request; + if (overrides.mediaEntityId && overrides.mediaEntityId !== request.mediaEntityId) { + delete request.contentUnitId; + delete request.episodeStructure; + delete request.slotId; + delete request.fileOverrides; + delete request.importMode; + delete request.replacementAssetIds; + } if (overrides.sourcePaths) { const files = await d.ctx.db .selectFrom("organization_file_result") diff --git a/apps/server/src/modules/file-manager/organize.ts b/apps/server/src/modules/file-manager/organize.ts index d5a62354..266b38fa 100644 --- a/apps/server/src/modules/file-manager/organize.ts +++ b/apps/server/src/modules/file-manager/organize.ts @@ -1,6 +1,7 @@ import { randomUUID } from "node:crypto"; import { join, relative } from "node:path"; import type { FilesOrganizeGroup, FilesOrganizeRequest, FilesOrganizeResult } from "@umt/contracts"; +import { overrideFor } from "@umt/domain"; import type { Principal } from "../../http/auth.ts"; import { ApiException } from "../../http/errors.ts"; import { vlstat as stat } from "../../storage/vfs.ts"; @@ -14,6 +15,8 @@ import { requestKey, sourceGuard, } from "../acquisition-organize/services/organize-guard.ts"; +import { walk } from "../acquisition-organize/services/organize-plan.ts"; +import { resolveOrganizeRecovery } from "../acquisition-organize/services/organize-retry.ts"; import { organizeStructure, setEpisodeGroup } from "../sources-identity/service/episode-groups.ts"; import type { Layout } from "../sources-identity/service/episode-structure.ts"; import { entityFromCandidate, externalKey } from "./candidate.ts"; @@ -21,6 +24,7 @@ import { actorOfP, errMessage, FilesError, type FmDeps, hashJson, isAdmin, resol import { audit, filesChanged, loadPlan, retirePlan, savePlan } from "./ledger.ts"; import { isWritableTarget } from "./list.ts"; import { entitiesUnder } from "./records.ts"; +import { recordSources } from "./tree.ts"; // 整理向导落地:按 (mediaEntityId, contentUnitId, 操作, 目标目录, Slot) 分组,每组走 planImport(与 POST /organization-records // 同一实现,同格式整理记录)。逐项季/集/单元覆写以 fileOverrides 传入 planImport(计划前后路径即反映覆写), @@ -193,6 +197,42 @@ export async function organizeFiles( const db = d.ctx.db; const actorType = principal?.kind === "api_key" ? "api_key" : "user"; const actor = actorOfP(principal); + const recoveryRecord = body.recoveryRecordId + ? await db.selectFrom("organization_record").selectAll().where("id", "=", body.recoveryRecordId).executeTakeFirst() + : undefined; + if (body.recoveryRecordId && !recoveryRecord) throw new FilesError("not_found", "原整理记录不存在"); + const recovered = recoveryRecord ? await resolveOrganizeRecovery(db, recoveryRecord) : undefined; + const recoveryPaths = new Set(); + const recoveryDirectories: string[] = []; + const movedOverrides: Record = {}; + const recoverySources = recoveryRecord ? await recordSources(d, recoveryRecord.id) : undefined; + if (recoveryRecord) { + const results = await db + .selectFrom("organization_file_result") + .selectAll() + .where("organization_record_id", "=", recoveryRecord.id) + .execute(); + const roots = await db.selectFrom("library_root").select(["id", "local_path"]).execute(); + for (const file of results) { + recoveryPaths.add(file.source_path); + const root = roots.find((r) => r.id === file.target_library_root_id); + if (root && file.target_path && file.status === "succeeded") { + const target = join(root.local_path, file.target_path); + recoveryPaths.add(target); + const override = overrideFor(recovered?.request.fileOverrides, file.source_path); + if (override) movedOverrides[target] = override; + } + } + const frozen = recoveryRecord.plan as unknown as Partial; + for (const file of frozen.organize?.files ?? []) recoveryPaths.add(file.src); + for (const file of frozen.organize?.excluded ?? []) recoveryPaths.add(file.src); + if (!recoveryPaths.size) { + for (const path of recovered?.recovery.request.sourcePaths ?? []) { + recoveryPaths.add(path); + if ((await stat(path).catch(() => null))?.isDirectory()) recoveryDirectories.push(path); + } + } + } // 先核验整批人工集号,避免无效输入创建外部作品或冻结计划。数字范围由请求 schema 校验。 const localIds = [...new Set(body.items.flatMap((it) => (it.mediaEntityId ? [it.mediaEntityId] : [])))]; @@ -232,9 +272,9 @@ export async function organizeFiles( const entityOf = (it: (typeof body.items)[number]) => it.mediaEntityId ?? created.get(externalKey(it.external!))!.entityId; const fromExternal = new Map([...created.values()].map((c) => [c.entityId, c])); - const minSizeBytes = body.minSizeBytes ?? (await settingsMinSize(d)); - const onExisting = body.onExisting ?? "conflict"; - const namingMode = body.naming ?? "template"; + const minSizeBytes = body.minSizeBytes ?? recovered?.request.minSizeBytes ?? (await settingsMinSize(d)); + const onExisting = body.onExisting ?? recovered?.request.onExisting ?? "conflict"; + const namingMode = body.naming ?? recovered?.request.namingMode ?? "template"; const entityIds = [...new Set(body.items.map(entityOf))]; const entities = new Map( @@ -287,6 +327,22 @@ export async function organizeFiles( const roots = new Set(); for (const it of body.items) { const r = await resolveIn(d, it.root, it.path); + if (recovered && !recoverySources?.recovery?.missing.includes("sourcePaths")) { + const selected: string[] = []; + await walk(r.abs, selected, d.ao.ports.options.walkMaxFiles + 1); + if ( + !selected.length || + selected.some( + (path) => + !recoveryPaths.has(path) && + !recoveryDirectories.some((dir) => { + const rel = relative(dir, path); + return rel !== ".." && !rel.startsWith("../") && !rel.startsWith("/"); + }), + ) + ) + throw new FilesError("bad_request", "重新整理只能选择原记录中的文件；请单独整理其它来源"); + } if (r.rel === "") throw new FilesError("bad_request", "不能整理整个登记目录"); if (r.isSymlink) throw new FilesError("bad_request", `不整理符号链接：${r.rel}`); const action = it.operation ?? body.action; @@ -420,6 +476,7 @@ export async function organizeFiles( const hasOverrides = Object.keys(g.overrides).length > 0; const st = structures.get(g.entity); const reqHash = hashJson({ + ...(body.recoveryRecordId ? { recoveryRecordId: body.recoveryRecordId, recovered: recovered?.request } : {}), key: requestKey(req), ...(body.sourceConfigId && g.action !== "in_place" ? { sourceConfigId: body.sourceConfigId } : {}), ...(hasOverrides ? { overrides: g.overrides } : {}), @@ -447,10 +504,30 @@ export async function organizeFiles( onExisting, ...(minSizeBytes !== undefined ? { minSizeBytes } : {}), }; + const inherited = { ...recovered?.request }; + if (recovered && g.entity !== recovered.request.mediaEntityId) { + delete inherited.contentUnitId; + delete inherited.episodeStructure; + delete inherited.slotId; + delete inherited.fileOverrides; + delete inherited.importMode; + delete inherited.replacementAssetIds; + } else if (inherited.fileOverrides) { + inherited.fileOverrides = { ...inherited.fileOverrides, ...movedOverrides }; + } + // 恢复向导提交完整的目标/来源/季集选择；省略代表选择自动或跟随作品，不能沿用旧覆写。 + delete inherited.targetDir; + delete inherited.targetLibraryRootId; + delete inherited.sourceConfigId; + delete inherited.episodeStructure; const importReq = (dryRun: boolean, taskId?: string) => ({ + ...inherited, ...req, + ...(body.recoveryRecordId ? { retryOfRecordId: body.recoveryRecordId } : {}), ...(body.sourceConfigId && g.action !== "in_place" ? { sourceConfigId: body.sourceConfigId } : {}), - ...(hasOverrides ? { fileOverrides: g.overrides } : {}), + ...(hasOverrides || inherited.fileOverrides + ? { fileOverrides: { ...inherited.fileOverrides, ...g.overrides } } + : {}), ...(g.targetDir ? { targetDir: g.targetDir } : {}), ...(g.slotId ? { slotId: g.slotId } : {}), ...(minSizeBytes !== undefined ? { minSizeBytes } : {}), @@ -458,8 +535,8 @@ export async function organizeFiles( // 向导里显式选择"替换"即为人工授权;替换走 replace_with_verify(新文件核验通过后旧文件进回收站) ...(onExisting === "replace_with_verify" ? { grantReplace: true } : {}), namingMode, - skipSuccessfulRecords: body.skipSuccessfulRecords, - scrapeAfterImport: body.scrapeAfterImport, + skipSuccessfulRecords: body.skipSuccessfulRecords ?? recovered?.request.skipSuccessfulRecords, + scrapeAfterImport: body.scrapeAfterImport ?? recovered?.request.scrapeAfterImport, // 勾了「同时设为作品的季集结构」:执行前已切换作品,之后按作品(新结构)整理;预览按临时结构算,命名相同 ...(st ? { diff --git a/apps/server/src/modules/file-manager/tree.ts b/apps/server/src/modules/file-manager/tree.ts index 8d8fc975..55db1ce7 100644 --- a/apps/server/src/modules/file-manager/tree.ts +++ b/apps/server/src/modules/file-manager/tree.ts @@ -7,9 +7,10 @@ import type { OrganizationStatus, } from "@umt/contracts"; import { sql } from "@umt/db"; -import { isRemotePath, vlstat as lstat, vreaddirents as readdirents } from "../../storage/vfs.ts"; +import { isRemotePath, vreaddirents as readdirents } from "../../storage/vfs.ts"; +import { resolveOrganizeRecovery } from "../acquisition-organize/services/organize-retry.ts"; import { FilesError, type FmDeps, isTempName, loadRootDefs, resolveIn, within } from "./common.ts"; -import { kindOfName } from "./list.ts"; +import { isWritableTarget, kindOfName } from "./list.ts"; // G8 目录摘要(树折叠 / 上一级):只列子目录(是否还有下级)+ 本目录递归视频数与已登记数; // G5 整理记录 → 源 FileRef(整理记录"重新整理"按登记目录 hostPath 换算)。 @@ -82,22 +83,9 @@ export async function treeOf(d: FmDeps, q: FilesTreeQuery): Promise { const db = d.ctx.db; - const rec = await db - .selectFrom("organization_record") - .select([ - "id", - "media_entity_id", - "action", - "status", - "source_path", - "source_path_view", - "target_path", - "target_library_root_id", - "plan", - ]) - .where("id", "=", recordId) - .executeTakeFirst(); + const rec = await db.selectFrom("organization_record").selectAll().where("id", "=", recordId).executeTakeFirst(); if (!rec) throw new FilesError("not_found", "整理记录不存在"); + const { recovery } = await resolveOrganizeRecovery(db, rec); const mains = await db .selectFrom("organization_file_result") .select(["source_path", "target_path", "target_library_root_id", "status"]) @@ -129,6 +117,7 @@ export async function recordSources(d: FmDeps, recordId: string): Promise !!p); + if (!mains.length && recovery.request.sourcePaths?.length) sources = recovery.request.sourcePaths; // 部分移动的文件分布在媒体库与下载目录，不能折叠成一个目录丢掉已移动文件。 const dir = partialMove ? null : moved ? targetOf(rec.target_library_root_id, rec.target_path) : rec.source_path; if ( @@ -142,9 +131,9 @@ export async function recordSources(d: FmDeps, recordId: string): Promise true, () => false, ) @@ -152,7 +141,25 @@ export async function recordSources(d: FmDeps, recordId: string): Promise null); + if (target?.lst?.isDirectory() && (await isWritableTarget(target.def, target.rootReal))) targetDir = ref; + } + } + if (recovery.request.targetDir && !targetDir) { + recovery.ready = false; + if (!recovery.missing.includes("options")) recovery.missing.push("options"); + } return { + recovery, + ...(targetDir ? { targetDir } : {}), recordId: rec.id, mediaEntityId: rec.media_entity_id, action: rec.action, diff --git a/apps/server/test/acquisition-organize/organize-identity-recovery.test.ts b/apps/server/test/acquisition-organize/organize-identity-recovery.test.ts index b1b20bbf..827ff566 100644 --- a/apps/server/test/acquisition-organize/organize-identity-recovery.test.ts +++ b/apps/server/test/acquisition-organize/organize-identity-recovery.test.ts @@ -1,3 +1,4 @@ +import { randomUUID } from "node:crypto"; import { symlink, unlink, writeFile } from "node:fs/promises"; import { join } from "node:path"; import { createTempTree, type TempTree } from "@umt/testkit"; @@ -123,7 +124,7 @@ describe.skipIf(!hasPg)("按路径身份补齐后恢复原整理", () => { ).toMatchObject({ retryOfRecordId: plan.recordId }); }); it("身份已找到但重新规划失败，保持原事项待处理", async () => { - const { dec } = await unknown("Blocked.Path", { targetLibraryRootId: "missing-root" }); + const { dec } = await unknown("Blocked.Path", { targetLibraryRootId: randomUUID() }); const entityId = await seedEntity(h.db, { kind: "movie", title: "Blocked Path", year: 2022 }); await expect(assignOrganizeIdentity(h.d, dec.id, { mediaEntityId: entityId }, actor, "标题一致")).rejects.toThrow( /重新整理没能执行/, @@ -165,7 +166,9 @@ describe.skipIf(!hasPg)("按路径身份补齐后恢复原整理", () => { it("旧记录没有请求快照时拒绝猜测一次性参数", async () => { const { plan } = await unknown("Legacy.Path"); await h.db.updateTable("organization_record").set({ plan: "{}" }).where("id", "=", plan.recordId!).execute(); - await expect(retryOrganizeRecord(h.d, plan.recordId!, actor, "test")).rejects.toThrow(/没有可恢复/); + await expect(retryOrganizeRecord(h.d, plan.recordId!, actor, "test")).rejects.toThrow( + /本次整理设置.*已保留可用信息/, + ); }); it("整理文件季集事项默认调查，无写授权降级建议，资源站普通事项不开启全局恢复", async () => { await h.setSettings("agent", { autoInvestigate: false, autoIdentifyOrganize: true }); diff --git a/apps/server/test/file-manager/legacy-organize-recovery.test.ts b/apps/server/test/file-manager/legacy-organize-recovery.test.ts new file mode 100644 index 00000000..1f66ca02 --- /dev/null +++ b/apps/server/test/file-manager/legacy-organize-recovery.test.ts @@ -0,0 +1,350 @@ +import { randomUUID } from "node:crypto"; +import { access, mkdir, rmdir, symlink, unlink, writeFile } from "node:fs/promises"; +import { dirname, join } from "node:path"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { ensureTaskRow } from "../../src/modules/acquisition-organize/services/common.ts"; +import { + executeRecord, + type FrozenPlan, + planImport, +} from "../../src/modules/acquisition-organize/services/organize.ts"; +import { loadRecordsPublic } from "../../src/modules/acquisition-organize/services/organize-records.ts"; +import { + OrganizeRecoveryError, + resolveOrganizeRecovery, + retryOrganizeRecord, +} from "../../src/modules/acquisition-organize/services/organize-retry.ts"; +import { createFmHarness, type FmHarness, key, pgAvailable, seedEntity } from "./harness.ts"; + +describe.skipIf(!(await pgAvailable()))("历史整理恢复沿用已知事实", () => { + let h: FmHarness; + beforeAll(async () => { + h = await createFmHarness("legacy_recovery", {}, { runTransfers: false }); + }); + afterAll(async () => { + await h?.close(); + }); + const actor = { type: "user", id: "legacy-recovery" }; + const record = (id: string) => + h.db.selectFrom("organization_record").selectAll().where("id", "=", id).executeTakeFirstOrThrow(); + async function removeSnapshot(id: string) { + const row = await record(id); + const plan = { ...(row.plan as unknown as Partial) }; + delete plan.request; + await h.db + .updateTable("organization_record") + .set({ plan: JSON.stringify(plan) }) + .where("id", "=", id) + .execute(); + } + async function source(label: string) { + const title = `${label} ${randomUUID().slice(0, 8)}`; + const path = join(h.dl, `${title.replaceAll(" ", ".")}.2024.1080p.mkv`); + await writeFile(path, title.padEnd(128, "x")); + return { title, path }; + } + it("旧任务保存完整请求时恢复Move及一次性设置，完成后幂等重放不读取已移动来源", async () => { + const file = await source("Legacy Task"); + const targetDir = join(h.lib, "task-target"); + await mkdir(targetDir); + const input = { + sourcePaths: [file.path], + action: "move" as const, + targetDir, + namingMode: "keep" as const, + scrapeAfterImport: false, + minSizeBytes: 10, + onExisting: "conflict" as const, + keepBoth: true, + }; + const taskId = await ensureTaskRow(h.db, { kind: "verifyAndImport", idempotencyKey: key(), input }); + const original = await planImport(h.module.deps().ao, { ...input, taskId, trigger: "manual", actorType: "user" }); + expect(original.reason).toBe("identity_unknown"); + await h.db.updateTable("organization_record").set({ plan: "{}" }).where("id", "=", original.recordId!).execute(); + const resolved = await resolveOrganizeRecovery(h.db, await record(original.recordId!)); + expect(resolved.recovery).toMatchObject({ ready: true, source: "task", missing: [], request: input }); + const entity = await seedEntity(h.db, { kind: "movie", title: file.title, year: 2024 }); + const options = { mediaEntityId: entity, idempotencyKey: `legacy:${original.recordId}` }; + const next = await retryOrganizeRecord(h.module.deps().ao, original.recordId!, actor, "test", options); + expect((await record(next.recordId!)).plan).toMatchObject({ request: { ...input, mediaEntityId: entity } }); + expect((await executeRecord(h.module.deps().ao, next.recordId!)).status).toBe("succeeded"); + await expect(access(file.path)).rejects.toThrow(); + expect(await retryOrganizeRecord(h.module.deps().ao, original.recordId!, actor, "test", options)).toMatchObject({ + recordId: next.recordId, + status: "succeeded", + }); + }); + it("完整冻结计划缺原设置时预填事实，向导确认后保留原选项和关联，不从头选择来源", async () => { + const file = await source("Legacy Plan"); + const entity = await seedEntity(h.db, { kind: "movie", title: file.title, year: 2024 }); + const targetDir = join(h.lib, "plan-target"); + await mkdir(targetDir); + const original = await planImport(h.module.deps().ao, { + sourcePaths: [file.path], + mediaEntityId: entity, + action: "copy", + targetDir, + namingMode: "keep", + scrapeAfterImport: false, + minSizeBytes: 0, + trigger: "manual", + actorType: "user", + }); + await removeSnapshot(original.recordId!); + await h.db + .updateTable("organization_record") + .set({ status: "failed" }) + .where("id", "=", original.recordId!) + .execute(); + await h.db + .deleteFrom("organization_file_result") + .where("organization_record_id", "=", original.recordId!) + .execute(); + const resolved = await resolveOrganizeRecovery(h.db, await record(original.recordId!)); + expect(resolved.recovery).toMatchObject({ + ready: false, + source: "plan", + missing: ["options"], + request: { + sourcePaths: [file.path], + mediaEntityId: entity, + targetDir, + namingMode: "keep", + scrapeAfterImport: false, + }, + }); + await expect(retryOrganizeRecord(h.module.deps().ao, original.recordId!, actor, "test")).rejects.toBeInstanceOf( + OrganizeRecoveryError, + ); + const refs = await h.api("GET", `/files/record-sources?recordId=${original.recordId}`); + expect(refs.body.data.items).toEqual([{ root: h.dlRoot, path: file.path.slice(h.dl.length + 1) }]); + expect(refs.body.data.targetDir).toEqual({ root: h.libRoot, path: "plan-target" }); + const row = await record(original.recordId!); + expect((await loadRecordsPublic(h.db, [row], false))[0]).not.toHaveProperty("recovery"); + expect((await loadRecordsPublic(h.db, [row], true))[0]!.recovery).toMatchObject({ source: "plan", ready: false }); + const payload = { + recoveryRecordId: original.recordId, + action: "copy", + onExisting: "conflict", + minSizeBytes: 0, + items: refs.body.data.items.map((item: { root: string; path: string }) => ({ + ...item, + mediaEntityId: entity, + targetDir: refs.body.data.targetDir, + })), + }; + const dry = await h.api("POST", "/files/organize", { ...payload, idempotencyKey: key(), dryRun: true }); + expect(dry.body.data.groups[0].error).toBeNull(); + const exec = await h.api("POST", "/files/organize", { + ...payload, + idempotencyKey: key(), + dryRun: false, + planIds: dry.body.data.groups.map((g: { planId: string }) => g.planId), + }); + expect(exec.body.data.groups[0].error).toBeNull(); + const next = await record(exec.body.data.recordIds[0]); + expect(next.plan).toMatchObject({ + retryOfRecordId: original.recordId, + request: { targetDir, namingMode: "keep", scrapeAfterImport: false }, + }); + }); + it("旧向导任务仅有基础参数不能冒充完整请求，未知来源明确缺项", async () => { + const file = await source("Legacy Basic"); + const input = { sourcePaths: [file.path], action: "copy", planId: randomUUID() }; + const taskId = await ensureTaskRow(h.db, { kind: "verifyAndImport", idempotencyKey: key(), input }); + const original = await planImport(h.module.deps().ao, { + sourcePaths: [file.path], + taskId, + trigger: "manual", + actorType: "user", + }); + await h.db + .updateTable("organization_record") + .set({ plan: "{}", source_path: null }) + .where("id", "=", original.recordId!) + .execute(); + expect((await resolveOrganizeRecovery(h.db, await record(original.recordId!))).recovery).toMatchObject({ + ready: false, + missing: ["options"], + request: { sourcePaths: [file.path] }, + }); + await h.db.deleteFrom("decision").where("subject_id", "=", original.recordId!).execute(); + await h.db.updateTable("organization_record").set({ task_id: null }).where("id", "=", original.recordId!).execute(); + expect((await resolveOrganizeRecovery(h.db, await record(original.recordId!))).recovery.missing).toEqual([ + "options", + "sourcePaths", + ]); + }); + it("恢复向导拒绝非原文件，历史来源后来变为越界符号链接时不预填可操作引用", async () => { + const file = await source("Legacy Guard"); + const entity = await seedEntity(h.db, { kind: "movie", title: file.title, year: 2024 }); + const original = await planImport(h.module.deps().ao, { + sourcePaths: [file.path], + mediaEntityId: entity, + action: "copy", + minSizeBytes: 0, + trigger: "manual", + actorType: "user", + }); + const other = await source("Unrelated"); + const denied = await h.api("POST", "/files/organize", { + recoveryRecordId: original.recordId, + action: "copy", + idempotencyKey: key(), + dryRun: true, + items: [{ root: h.dlRoot, path: other.path.slice(h.dl.length + 1), mediaEntityId: entity }], + }); + expect(denied.status).toBe(400); + const escaped = join(h.outside, "outside.mkv"); + await writeFile(escaped, "outside"); + await unlink(file.path); + await symlink(escaped, file.path); + const refs = await h.api("GET", `/files/record-sources?recordId=${original.recordId}`); + expect(refs.body.data.items).toEqual([]); + expect(refs.body.data.unresolved).toEqual([file.path]); + }); + it("失效的原目标明确缺项，确认自动目标不能继续沿用失效目录", async () => { + const file = await source("Lost Target"); + const entity = await seedEntity(h.db, { kind: "movie", title: file.title, year: 2024 }); + const targetDir = join(h.lib, `lost-${randomUUID()}`); + await mkdir(targetDir); + const original = await planImport(h.module.deps().ao, { + sourcePaths: [file.path], + mediaEntityId: entity, + targetDir, + action: "copy", + minSizeBytes: 0, + trigger: "manual", + actorType: "user", + }); + await rmdir(targetDir); + const refs = await h.api("GET", `/files/record-sources?recordId=${original.recordId}`); + expect(refs.body.data.targetDir).toBeUndefined(); + expect(refs.body.data.recovery).toMatchObject({ ready: false, missing: ["options"] }); + const body = { + recoveryRecordId: original.recordId, + action: "copy", + minSizeBytes: 0, + dryRun: true, + items: refs.body.data.items.map((x: { root: string; path: string }) => ({ ...x, mediaEntityId: entity })), + }; + const denied = await h.api("POST", "/files/organize", { ...body, idempotencyKey: key() }); + expect(denied.body.data.groups[0].error).toBeNull(); + const accepted = await h.api("POST", "/files/organize", { + ...body, + targetLibraryRootId: h.libraryRootId, + idempotencyKey: key(), + }); + expect(accepted.body.data.groups[0].error).toBeNull(); + }); + it("部分移动保留成功目标上的原季集覆写，换作品清除旧归属参数", async () => { + const title = `Partial ${randomUUID().slice(0, 8)}`; + const entity = await seedEntity(h.db, { kind: "series", title, year: 2024 }); + const folder = join(h.dl, title.replaceAll(" ", ".")); + await mkdir(folder); + const paths = [1, 2].map((n) => join(folder, `${title.replaceAll(" ", ".")}.S01E0${n}.mkv`)); + for (const path of paths) await writeFile(path, "video".padEnd(128, "x")); + const original = await planImport(h.module.deps().ao, { + sourcePaths: paths, + mediaEntityId: entity, + action: "move", + minSizeBytes: 0, + keepBoth: true, + scrapeAfterImport: false, + fileOverrides: Object.fromEntries( + paths.map((p, i) => [p, { season: 1, episodes: { start: i + 7, end: i + 7 } }]), + ), + trigger: "manual", + actorType: "user", + }); + const files = await h.db + .selectFrom("organization_file_result") + .selectAll() + .where("organization_record_id", "=", original.recordId!) + .where("role", "=", "main") + .orderBy("ordinal") + .execute(); + expect(files).toHaveLength(2); + const conflict = join(h.lib, files[1]!.target_path!); + await mkdir(dirname(conflict), { recursive: true }); + await writeFile(conflict, "different existing content"); + expect((await executeRecord(h.module.deps().ao, original.recordId!)).status).toBe("partial"); + const target = join(h.lib, files[0]!.target_path!); + await expect(access(paths[0]!)).rejects.toThrow(); + await access(target); + const refs = await h.api("GET", `/files/record-sources?recordId=${original.recordId}`); + expect(refs.body.data.items).toHaveLength(2); + const targetDir = join(h.lib, `retry-${randomUUID()}`); + await mkdir(targetDir); + async function submit(mediaEntityId: string) { + const body = { + recoveryRecordId: original.recordId, + action: "copy", + minSizeBytes: 0, + items: refs.body.data.items.map((x: { root: string; path: string }) => ({ + ...x, + mediaEntityId, + targetDir: { root: h.libRoot, path: targetDir.slice(h.lib.length + 1) }, + })), + }; + const dry = await h.api("POST", "/files/organize", { ...body, idempotencyKey: key(), dryRun: true }); + expect(dry.body.data.groups[0].error).toBeNull(); + const exec = await h.api("POST", "/files/organize", { + ...body, + idempotencyKey: key(), + dryRun: false, + planIds: dry.body.data.groups.map((g: { planId: string }) => g.planId), + }); + expect(exec.body.data.groups[0].error).toBeNull(); + return (await record(exec.body.data.recordIds[0])).plan as unknown as FrozenPlan; + } + const previous = await record(original.recordId!); + const previousPlan = previous.plan as unknown as FrozenPlan; + previousPlan.request = { + ...previousPlan.request!, + sourceConfigId: randomUUID(), + episodeStructure: { groupId: null, layout: null }, + }; + await h.db + .updateTable("organization_record") + .set({ plan: JSON.stringify(previousPlan) }) + .where("id", "=", previous.id) + .execute(); + const same = await submit(entity); + expect(same.request).toMatchObject({ + keepBoth: true, + fileOverrides: { + [target]: { season: 1, episodes: { start: 7, end: 7 } }, + [paths[1]!]: { season: 1, episodes: { start: 8, end: 8 } }, + }, + }); + expect(same.request).not.toHaveProperty("sourceConfigId"); + expect(same.request).not.toHaveProperty("episodeStructure"); + const row = await record(original.recordId!); + const plan = row.plan as unknown as FrozenPlan; + plan.request = { + ...plan.request!, + contentUnitId: randomUUID(), + slotId: randomUUID(), + episodeStructure: { groupId: null, layout: null }, + importMode: "replace", + replacementAssetIds: [randomUUID()], + }; + await h.db + .updateTable("organization_record") + .set({ plan: JSON.stringify(plan) }) + .where("id", "=", row.id) + .execute(); + const other = await seedEntity(h.db, { kind: "series", title: `Changed ${randomUUID()}`, year: 2024 }); + const changed = await submit(other); + for (const field of [ + "contentUnitId", + "slotId", + "episodeStructure", + "fileOverrides", + "importMode", + "replacementAssetIds", + ]) + expect(changed.request).not.toHaveProperty(field); + }); +});