From 36aa99b8aa506330ade2b845e4f58182c9893d40 Mon Sep 17 00:00:00 2001 From: root Date: Sun, 27 Sep 2026 23:53:50 +0800 Subject: [PATCH] feat(files): W5-A2 move/copy via planImport, organize overrides, delete records, download MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - /files/move|copy go through planImport's new `transfer` variant (keep names, directory expanded to per-file plan) → organization_record + verifyAndImport/executeRecord: per-file ledger resumable, fsproxy, cross-volume directory move = per-file copy→verify→delete then remove empty source dirs; shared plan store + plan_stale; asset_location follows per file, path_binding revised when a whole directory moved (finalizeTransfer); request waits up to FILES_TRANSFER_WAIT_SEC, else executed.pending - /files/organize per-item overrides take effect: season/episodes (domain planOrganize overrides), contentUnitId coverage, targetDir (no binding), operation, slotId (satisfied after verify); warnings only for inapplicable overrides - delete/trash/restore write organization_record (action trash|delete|restore, file-level results; migration 0026) - GET /files/download (admin session, nfo/subtitle/image, size cap, streamed, audited) - GET /files/records?ids|planIds for the organize page - fs race errors map to 503/404/403/400 instead of 500 - runbook updated Co-Authored-By: Claude Opus 5.5 --- .../services/organize-transfer.ts | 418 ++++++++++++++++++ .../acquisition-organize/services/organize.ts | 98 +++- .../server/src/modules/file-manager/common.ts | 27 +- .../src/modules/file-manager/download.ts | 82 ++++ .../src/modules/file-manager/organize.ts | 168 ++++++- .../src/modules/file-manager/records.ts | 150 ++----- .../server/src/modules/file-manager/routes.ts | 19 +- .../src/modules/file-manager/transfer.ts | 263 ++++++----- apps/server/src/modules/file-manager/trash.ts | 94 +++- apps/server/test/file-manager/harness.ts | 35 +- apps/server/test/file-manager/w5a2.test.ts | 414 +++++++++++++++++ apps/server/test/file-manager/write.test.ts | 5 +- apps/web/app/lib/labels.ts | 3 + apps/web/app/routes/console/organize.$id.tsx | 3 + apps/web/app/routes/console/organize.tsx | 3 + docs/runbook/files.md | 30 +- packages/contracts/src/api.ts | 16 + packages/contracts/src/files.ts | 67 ++- packages/contracts/src/organization.ts | 16 +- .../0026_organization_file_actions.ts | 30 ++ packages/db/src/types.ts | 20 +- packages/domain/src/organize/index.ts | 2 + packages/domain/src/organize/plan.ts | 32 ++ packages/domain/test/organize.test.ts | 38 ++ packages/mapping/src/organize.ts | 4 + 25 files changed, 1766 insertions(+), 271 deletions(-) create mode 100644 apps/server/src/modules/acquisition-organize/services/organize-transfer.ts create mode 100644 apps/server/src/modules/file-manager/download.ts create mode 100644 apps/server/test/file-manager/w5a2.test.ts create mode 100644 packages/db/migrations/0026_organization_file_actions.ts diff --git a/apps/server/src/modules/acquisition-organize/services/organize-transfer.ts b/apps/server/src/modules/acquisition-organize/services/organize-transfer.ts new file mode 100644 index 00000000..a0c40bd3 --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/services/organize-transfer.ts @@ -0,0 +1,418 @@ +import { randomUUID } from "node:crypto"; +import { readdir, rmdir } from "node:fs/promises"; +import { dirname, join, relative } from "node:path"; +import { checkRoot, DEFAULT_ROOT_SENTINEL, statSafe } from "@umt/adapters"; +import { fileKindOf, type OrganizeFileKind, type OrganizePlan, type PlannedFile } from "@umt/domain"; +import { organizationStatusFromFiles, organizePlanToFileResults } from "@umt/mapping"; +import { hashJson, jsonb } from "../util.ts"; +import type { AoDeps, Exec } from "./common.ts"; +import { type ExecuteResult, emitRecord, type FrozenPlan, type ImportRequest, type PlanResult } from "./organize.ts"; + +// W5-A2 原样搬运(文件管理 移动/复制到目录):与整理共用 planImport 入口、冻结计划、organization_record、 +// verifyAndImport 任务与 executeRecord(逐文件账本可恢复、fsproxy 挂死隔离、跨卷 Copy→核验→删源、EXDEV 不降级)。 +// 与作品整理的差别:保持文件名与目录结构;不识别作品、不登记新资产;move 成功后已登记的 asset_location 跟随、 +// 整个目录搬完时命中的 path_binding 写修订,并删掉源侧留下的空目录。 + +export interface TransferItem { + /** 源(绝对路径,文件或目录) */ + src: string; + /** 目标(绝对路径,不存在) */ + dst: string; + /** 源所在库根与根内相对路径(源为下载根时 null,不做位置跟随) */ + srcLibraryRootId: string | null; + srcRel: string | null; + /** 目标在目标库根内的相对路径 */ + dstRel: string; +} + +export interface TransferSpec { + items: TransferItem[]; + targetLibraryRootId: string; + /** 目标库根 realpath(相对路径以它为基准) */ + targetRootPath: string; + /** 遍历到的每个文件都经它复核(realpath 在允许的根下) */ + guard?: ((p: string) => Promise) | undefined; +} + +export interface FrozenTransfer { + items: (TransferItem & { isDir: boolean; files: number })[]; + targetLibraryRootId: string; +} + +const KIND_OF = (p: string): OrganizeFileKind => { + const k = fileKindOf(p); + return k === "video" || k === "subtitle" || k === "audio" || k === "nfo" ? k : "other"; +}; + +/** 递归列出目录下的普通文件与子目录(不跟随符号链接;符号链接列为 skipped) */ +async function walkTree( + dir: string, + out: { files: string[]; dirs: string[]; links: string[] }, + limit: number, +): Promise { + const names = (await readdir(dir)).sort(); + for (const n of names) { + if (out.files.length >= limit) return true; + const p = join(dir, n); + const st = await statSafe(p); + if (st.kind !== "ok") throw new Error(`源不可访问: ${p} (${st.kind})`); + if (st.facts.isSymbolicLink) out.links.push(p); + else if (st.facts.isDirectory) { + out.dirs.push(p); + if (await walkTree(p, out, limit)) return true; + } else if (st.facts.isFile) out.files.push(p); + } + return false; +} + +/** 原样搬运计划;dryRun 只返回冻结计划,否则写 organization_record(status=running)等 verifyAndImport 执行 */ +export async function planTransfer(d: AoDeps, req: ImportRequest, t: TransferSpec): Promise { + const db = d.ctx.db; + const dry = req.dryRun === true; + const action = req.action ?? "copy"; + const failed = (reason: string): PlanResult => ({ recordId: null, status: "failed", reason }); + const root = await db + .selectFrom("library_root") + .selectAll() + .where("id", "=", t.targetLibraryRootId) + .executeTakeFirst(); + if (root?.management_mode !== "managed") return failed("target_root_not_managed"); + const sentinel = root.sentinel_file ?? DEFAULT_ROOT_SENTINEL; + const health = await checkRoot(t.targetRootPath, sentinel); + if (health.kind === "offline") return failed(`library_root_offline:${health.reason}`); + + const max = d.ports.options.walkMaxFiles; + const files: PlannedFile[] = []; + const excluded: OrganizePlan["excluded"] = []; + const dirs = new Set(); + const items: FrozenTransfer["items"] = []; + let truncated = false; + for (const it of t.items) { + const st = await statSafe(it.src); + if (st.kind !== "ok") return failed(`source_missing:${it.src}`); + const add = async (src: string, dst: string) => { + if (t.guard) await t.guard(src); + const s = await statSafe(src); + if (s.kind !== "ok") throw new Error(`源不可访问: ${src} (${s.kind})`); + files.push({ + src, + dst, + size: s.facts.size, + kind: KIND_OF(src), + role: "main", + action: "transfer", + crossDevice: s.facts.dev !== health.dev, + }); + }; + if (!st.facts.isDirectory) { + await add(it.src, it.dst); + items.push({ ...it, isDir: false, files: 1 }); + continue; + } + const tree = { files: [] as string[], dirs: [] as string[], links: [] as string[] }; + if (await walkTree(it.src, tree, max - files.length + 1)) truncated = true; + const map = (p: string) => join(it.dst, relative(it.src, p)); + dirs.add(it.dst); + for (const sub of tree.dirs) dirs.add(map(sub)); + for (const f of tree.files) await add(f, map(f)); + for (const l of tree.links) + excluded.push({ src: l, kind: KIND_OF(l), reason: "not_imported_kind", detail: "symlink" }); + items.push({ ...it, isDir: true, files: tree.files.length }); + } + if (files.length > max) { + files.length = max; + truncated = true; + } + if (truncated && !dry) return failed("too_many_files"); + + // 目标已存在(或状态读不出):不覆盖,该文件执行前即记失败 + const conflicts: OrganizePlan["conflicts"] = []; + for (const f of files) { + const st = await statSafe(f.dst); + if (st.kind === "missing") continue; + f.action = "hold"; + conflicts.push({ kind: "target_exists", blocking: false, src: f.src, dst: f.dst }); + } + for (const f of files) dirs.add(dirname(f.dst)); + const plan: OrganizePlan = { + operation: action, + crossDevice: files.some((f) => f.crossDevice === true), + files, + excluded, + preconditions: [], + conflicts, + requiresApproval: false, + }; + const frozen: FrozenPlan = { + version: 1, + organize: plan, + entityId: req.mediaEntityId ?? "", + mediaType: "movie", + libraryRootId: root.id, + rootPath: t.targetRootPath, + sentinel, + targetDirs: [...dirs].sort(), + newBindings: [], + sourceRoot: null, + facts: {}, + transfer: { + items, + targetLibraryRootId: t.targetLibraryRootId, + }, + }; + if (dry) return { recordId: null, status: "queued", reason: null, plan: frozen, ...(truncated ? { truncated } : {}) }; + + const drafts = organizePlanToFileResults(plan, { targetLibraryRootId: root.id }); + const now = d.ports.now(); + const rel = (abs: string | null | undefined) => (abs ? relative(t.targetRootPath, abs) : null); + const heldSrc = new Set(files.filter((f) => f.action === "hold").map((f) => f.src)); + return db.transaction().execute(async (trx) => { + const rec = await trx + .insertInto("organization_record") + .values({ + task_id: req.taskId ?? null, + trigger: req.trigger, + actor_type: req.actorType, + actor_id: req.actorId ?? null, + media_entity_id: req.mediaEntityId ?? null, + source_path_view: "local", + source_path: items.length === 1 ? items[0]!.src : dirname(items[0]!.src), + target_library_root_id: root.id, + target_path: items.length === 1 ? items[0]!.dstRel : rel(dirname(items[0]!.dst)), + action, + status: "running", + plan: jsonb(frozen), + plan_fingerprint: hashJson(frozen), + file_total: drafts.length, + file_skipped: drafts.filter((f) => f.status === "skipped" && !heldSrc.has(f.sourcePath)).length, + file_failed: heldSrc.size, + started_at: now, + }) + .returning(["id"]) + .executeTakeFirstOrThrow(); + for (const f of drafts) { + const held = heldSrc.has(f.sourcePath); + await trx + .insertInto("organization_file_result") + .values({ + organization_record_id: rec.id, + ordinal: f.ordinal ?? 0, + role: f.role ?? "main", + mode: f.mode, + source_path: f.sourcePath, + source_size_bytes: f.sourceSizeBytes ?? null, + target_library_root_id: f.targetPath ? root.id : null, + target_path: rel(f.targetPath), + status: held ? "failed" : f.status, + skip_reason: held ? null : f.skipReason, + ...(held ? { error: jsonb({ code: "EEXIST", message: "目标已存在,不覆盖" }), finished_at: now } : {}), + }) + .execute(); + } + await emitRecord(d, trx, rec.id); + return { recordId: rec.id, status: "running" as const, reason: null }; + }); +} + +// ── 位置跟随 ── + +const likePrefix = (rel: string) => `${rel.replace(/[\\%_]/g, (m) => `\\${m}`)}/%`; + +export interface RelocateInput { + fromLib: string; + fromRel: string; + toLib: string; + toRel: string; + isDir: boolean; + actor: string; + opKind: "rename" | "move"; +} + +export interface RelocateResult { + locations: { id: string; assetId: string; from: string; to: string }[]; + bindings: { oldBindingId: string; newBindingId: string }[]; + mediaEntityIds: string[]; +} + +/** 同事务:把 fromLib/fromRel(目录含其下全部)的已登记位置改到 toLib/toRel;目录命中的路径绑定写修订 */ +export async function relocate(trx: Exec, input: RelocateInput): Promise { + const { fromLib, fromRel, toLib, toRel, isDir } = input; + const mapRel = (p: string) => (p === fromRel ? toRel : `${toRel}${p.slice(fromRel.length)}`); + const locs = await trx + .selectFrom("asset_location as l") + .innerJoin("media_asset as a", "a.id", "l.media_asset_id") + .select(["l.id", "l.relative_path", "l.media_asset_id", "a.media_entity_id"]) + .where("l.library_root_id", "=", fromLib) + .where("l.status", "<>", "removed") + .where((eb) => + isDir + ? eb.or([eb("l.relative_path", "=", fromRel), eb("l.relative_path", "like", likePrefix(fromRel))]) + : eb("l.relative_path", "=", fromRel), + ) + .execute(); + const out: RelocateResult = { locations: [], bindings: [], mediaEntityIds: [] }; + const entities = new Set(); + for (const l of locs) { + const to = mapRel(l.relative_path); + await trx + .updateTable("asset_location") + .set({ library_root_id: toLib, relative_path: to, last_seen_at: new Date() }) + .where("id", "=", l.id) + .execute(); + await trx + .insertInto("asset_operation") + .values({ + operation_id: `files:${randomUUID()}`, + phase: "commit", + kind: input.opKind, + status: "succeeded", + intent: jsonb({ + fromLibraryRootId: fromLib, + from: l.relative_path, + toLibraryRootId: toLib, + to, + by: input.actor, + }), + result: jsonb({ locationId: l.id }), + media_asset_id: l.media_asset_id, + started_at: new Date(), + finished_at: new Date(), + }) + .execute(); + out.locations.push({ id: l.id, assetId: l.media_asset_id, from: l.relative_path, to }); + if (l.media_entity_id) entities.add(l.media_entity_id); + } + if (isDir) { + const binds = await trx + .selectFrom("path_binding") + .selectAll() + .where("library_root_id", "=", fromLib) + .where("status", "=", "active") + .where((eb) => eb.or([eb("relative_path", "=", fromRel), eb("relative_path", "like", likePrefix(fromRel))])) + .execute(); + for (const b of binds) { + // 显式路径变更(REQ §7.1.1):旧绑定 superseded,新绑定 active 并互相链接;仅译名刷新永不走到这里 + await trx.updateTable("path_binding").set({ status: "superseded" }).where("id", "=", b.id).execute(); + const nb = await trx + .insertInto("path_binding") + .values({ + media_entity_id: b.media_entity_id, + content_unit_id: b.content_unit_id, + season_number: b.season_number, + binding_kind: b.binding_kind, + library_root_id: toLib, + relative_path: mapRel(b.relative_path), + template_version: b.template_version, + status: "active", + bound_by: input.actor, + note: `files.${input.opKind}: ${b.relative_path} → ${mapRel(b.relative_path)}`, + }) + .returning("id") + .executeTakeFirstOrThrow(); + await trx.updateTable("path_binding").set({ superseded_by_id: nb.id }).where("id", "=", b.id).execute(); + out.bindings.push({ oldBindingId: b.id, newBindingId: nb.id }); + entities.add(b.media_entity_id); + } + } + out.mediaEntityIds = [...entities]; + return out; +} + +/** 目录下的子目录,深的在前(删空目录用) */ +async function dirsDeepFirst(dir: string): Promise { + const out: string[] = []; + const walk = async (p: string) => { + const ents = await readdir(p, { withFileTypes: true }).catch(() => []); + for (const e of ents) if (e.isDirectory()) await walk(join(p, e.name)); + out.push(p); + }; + await walk(dir); + return out; +} + +/** 原样搬运的终态:记录状态 + move 的位置跟随/绑定修订 + 删源侧空目录;行锁 fencing,重复调用幂等 */ +export async function finalizeTransfer(d: AoDeps, recordId: string): Promise { + const db = d.ctx.db; + const pre = await db + .selectFrom("organization_record") + .select(["action", "plan", "status"]) + .where("id", "=", recordId) + .executeTakeFirstOrThrow(); + const frozenPre = pre.plan as unknown as FrozenPlan; + const tr = frozenPre.transfer!; + const filesPre = await db + .selectFrom("organization_file_result") + .select(["source_path", "status"]) + .where("organization_record_id", "=", recordId) + .execute(); + const under = (item: { src: string }, p: string) => p === item.src || p.startsWith(`${item.src}/`); + const allOk = (item: { src: string }) => + filesPre.filter((f) => under(item, f.source_path)).every((f) => f.status === "succeeded"); + // move:整个目录都搬完才删源侧空目录(先于终态提交;失败/中断重跑幂等) + if (pre.action === "move" && (pre.status === "running" || pre.status === "queued")) + for (const it of tr.items.filter((x) => x.isDir && allOk(x))) + for (const dir of await dirsDeepFirst(it.src)) await rmdir(dir).catch(() => undefined); + + return db.transaction().execute(async (trx) => { + const rec = await trx + .selectFrom("organization_record") + .selectAll() + .where("id", "=", recordId) + .forUpdate() + .executeTakeFirstOrThrow(); + if (rec.status !== "running" && rec.status !== "queued") + return { recordId, status: rec.status, succeeded: rec.file_succeeded, failed: rec.file_failed }; + const frozen = rec.plan as unknown as FrozenPlan; + const t = frozen.transfer!; + const files = await trx + .selectFrom("organization_file_result") + .selectAll() + .where("organization_record_id", "=", recordId) + .execute(); + const entities = new Set(); + if (rec.action === "move") { + const actor = `organize:${recordId}`; + for (const it of t.items) { + if (it.srcRel === null || it.srcLibraryRootId === null) continue; + const mine = files.filter((f) => under(it, f.source_path)); + const base = { fromLib: it.srcLibraryRootId, toLib: t.targetLibraryRootId, actor, opKind: "move" as const }; + if (mine.length > 0 && mine.every((f) => f.status === "succeeded")) { + // 整项搬完:按前缀跟随(目录含路径绑定修订) + const r = await relocate(trx, { ...base, fromRel: it.srcRel, toRel: it.dstRel, isDir: it.isDir }); + for (const e of r.mediaEntityIds) entities.add(e); + continue; + } + for (const f of mine.filter((x) => x.status === "succeeded")) { + const sub = f.source_path.slice(it.src.length); + const r = await relocate(trx, { + ...base, + fromRel: `${it.srcRel}${sub}`, + toRel: `${it.dstRel}${sub}`, + isDir: false, + }); + for (const e of r.mediaEntityIds) entities.add(e); + } + } + } + const statuses = files.map((f) => f.status); + const status = organizationStatusFromFiles(statuses); + const counts = { + file_succeeded: statuses.filter((s) => s === "succeeded").length, + file_failed: statuses.filter((s) => s === "failed").length, + file_skipped: statuses.filter((s) => s === "skipped").length, + }; + await trx + .updateTable("organization_record") + .set({ + status, + ...counts, + finished_at: d.ports.now(), + ...(status === "failed" && !rec.reason ? { reason: "transfer_failed" } : {}), + ...(!rec.media_entity_id && entities.size === 1 ? { media_entity_id: [...entities][0]! } : {}), + }) + .where("id", "=", recordId) + .execute(); + await emitRecord(d, trx, recordId); + return { recordId, status, succeeded: counts.file_succeeded, failed: counts.file_failed }; + }); +} diff --git a/apps/server/src/modules/acquisition-organize/services/organize.ts b/apps/server/src/modules/acquisition-organize/services/organize.ts index d468ee4f..cab35328 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize.ts @@ -27,6 +27,7 @@ import { movieCompletion, type NewBinding, type OrganizePlan, + overrideFor, type ParsedRelease, parseReleasePath, planOrganize, @@ -42,6 +43,7 @@ import { publishedAcquisitionPolicies, unitKeysOf } from "./acquisition.ts"; import { type AoDeps, type Exec, hostPathOf, loadInstances, upsertDecision } from "./common.ts"; import { applySlotEvent } from "./downloads.ts"; import { PathGuardError } from "./organize-guard.ts"; +import { type FrozenTransfer, finalizeTransfer, planTransfer, type TransferSpec } from "./organize-transfer.ts"; import { hnrSatisfied } from "./retention.ts"; // 入库:冻结整理计划(organization_record + 文件明细)→ 逐文件可恢复执行(操作账本 fencing)→ @@ -67,6 +69,24 @@ export interface ImportRequest { onExisting?: ExistingPolicy | undefined; /** 已获替换授权(审批/手动确认) */ grantReplace?: boolean | undefined; + /** + * W5-A2 逐文件覆写(文件管理整理向导的人工确认,优先于文件名解析):键为源文件或目录的绝对路径 + * (目录对其下文件生效,最长前缀优先)。season/episodes 决定剧集季目录与命名以及集单元覆盖; + * contentUnitIds 为核验后额外登记的内容单元覆盖。 + */ + fileOverrides?: Record | undefined; + /** W5-A2 目标目录覆写(绝对路径,须在目标库根下):按模板命名放入该目录,不使用也不新建路径绑定 */ + targetDir?: string | undefined; + /** W5-A2 手动整理核验通过后推进的 Slot(须属于该作品) */ + slotId?: string | undefined; + /** W5-A2 原样搬运(文件管理 移动/复制):保持文件名与目录结构,不识别作品、不登记新资产,已登记位置跟随 */ + transfer?: TransferSpec | undefined; +} + +export interface FileOverrideInput { + season?: number; + episodes?: { start: number; end: number }; + contentUnitIds?: string[]; } /** organization_record.plan 的冻结内容 */ @@ -84,9 +104,20 @@ export interface FrozenPlan { /** 主视频 src → 解析出的质量特征与单元键 */ facts: Record< string, - { quality: QualityFacts; unitKeys: string[]; season?: number; episodes?: { start: number; end: number } } + { + quality: QualityFacts; + unitKeys: string[]; + season?: number; + episodes?: { start: number; end: number }; + /** 人工指定的内容单元(W5-A2 覆写) */ + unitIds?: string[]; + } >; downgradedFromMove?: string; + /** W5-A2:手动整理指定的 Slot */ + slotId?: string; + /** W5-A2:原样搬运计划(finalize 走 finalizeTransfer) */ + transfer?: FrozenTransfer; } export interface PlanResult { @@ -171,6 +202,7 @@ async function chooseRoot( /** 生成并冻结整理计划;阻塞冲突 → 记录失败 + 待处理,不执行 */ export async function planImport(d: AoDeps, req: ImportRequest): Promise { + if (req.transfer) return planTransfer(d, req, req.transfer); const db = d.ctx.db; let dt: DownloadTaskRow | undefined; if (req.downloadTaskId) { @@ -334,6 +366,21 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise toDomainPathBinding(toApi(r))); const root = await chooseRoot(db, entity, bindings, req.targetLibraryRootId); if (!root) return fail("no_library_root", {}, entity.id); + const targetDir = req.targetDir?.replace(/\/+$/, ""); + if (targetDir !== undefined) { + const rp = root.local_path.replace(/\/+$/, ""); + if (!(targetDir === rp || targetDir.startsWith(`${rp}/`)) || targetDir.split("/").includes("..")) + return fail("target_dir_outside_root", { targetDir, libraryRootId: root.id }, entity.id); + } + if (req.slotId) { + const slot = await db + .selectFrom("acquisition_slot") + .select(["id", "media_entity_id"]) + .where("id", "=", req.slotId) + .executeTakeFirst(); + if (!slot || slot.media_entity_id !== entity.id) + return fail("slot_entity_mismatch", { slotId: req.slotId }, entity.id); + } const sentinel = root.sentinel_file ?? DEFAULT_ROOT_SENTINEL; const health = await checkRoot(root.local_path, sentinel); if (health.kind === "offline") { @@ -361,6 +408,9 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise [s, targetDir])); + } else if (mediaType === "movie") { const r = await handle(decide()); if (typeof r !== "string") return r; target.movieDir = r; @@ -451,6 +506,7 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise c.blocking); @@ -1137,6 +1194,8 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise oc.doNothing()) + .execute(); + } } assetBySrc.set(f.source_path, { assetId, locationId }); if (isMain) mainAssets.push(assetId); @@ -1424,6 +1496,28 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise = { +const REASON_STATUS: Record< + FilesErrorReason | "bad_request" | "forbidden" | "too_large", + { status: number; code: ApiErrorCode } +> = { root_unavailable: { status: 503, code: "upstream_unavailable" }, path_escape: { status: 400, code: "bad_request" }, bad_request: { status: 400, code: "bad_request" }, @@ -78,6 +87,7 @@ const REASON_STATUS: Record: ` 开头便于前端分支 */ @@ -101,6 +111,21 @@ const OFFLINE_CODES = new Set(["EIO", "ENOTCONN", "ESTALE", "ETIMEDOUT", "EHOSTD export const errnoOf = (e: unknown): string | undefined => (e as { code?: string } | null)?.code; export const isOfflineErr = (e: unknown) => OFFLINE_CODES.has(errnoOf(e) ?? ""); +/** + * 处理器里漏出的文件系统错误(检查与使用之间目录被删、挂载掉线等竞态)→ 业务错误,而不是 500: + * 离线类 → 503 root_unavailable(失败 ≠ 空目录),ENOENT/ENOTDIR → 404,EACCES/EPERM → 403,其余 errno → 400。 + */ +export function fsErrorToFiles(e: unknown): FilesError | null { + if (e instanceof FilesError) return e; + const code = errnoOf(e); + if (typeof code !== "string" || !/^E[A-Z]+$/.test(code)) return null; + const msg = e instanceof Error ? e.message : String(e); + if (OFFLINE_CODES.has(code)) return new FilesError("root_unavailable", `读取失败(${code}),根可能已离线: ${msg}`); + if (code === "ENOENT" || code === "ENOTDIR") return new FilesError("not_found", `路径已不存在(${code})`); + if (code === "EACCES" || code === "EPERM") return new FilesError("forbidden", `无权限访问(${code})`); + return new FilesError("bad_request", `文件系统错误(${code}): ${msg}`); +} + // ── 根 ── export interface RootDef { diff --git a/apps/server/src/modules/file-manager/download.ts b/apps/server/src/modules/file-manager/download.ts new file mode 100644 index 00000000..96ab51b8 --- /dev/null +++ b/apps/server/src/modules/file-manager/download.ts @@ -0,0 +1,82 @@ +import { basename, extname } from "node:path"; +import type { FilesDownloadQuery } from "@umt/contracts"; +import type { Principal } from "../../http/auth.ts"; +import { FilesError, type FmDeps, isAdmin, recheck, resolveIn } from "./common.ts"; +import { audit } from "./ledger.ts"; +import { kindOfName } from "./list.ts"; + +// GET /files/download(W5-A2):仅管理员会话;只下载 NFO/字幕/图片小文件(视频不经浏览器,规格 §1 边界); +// 路径守卫同其它接口(realpath 在根内、符号链接目标也须在根内,打开前再复核一次); +// 打开后按文件句柄再 stat,校验的就是要流出去的那个文件;流式返回,写审计。 + +const ALLOWED = new Set(["nfo", "subtitle", "image"]); + +const MIME: Record = { + jpg: "image/jpeg", + jpeg: "image/jpeg", + png: "image/png", + gif: "image/gif", + webp: "image/webp", + bmp: "image/bmp", + avif: "image/avif", + nfo: "text/plain; charset=utf-8", + xml: "text/plain; charset=utf-8", + srt: "text/plain; charset=utf-8", + ass: "text/plain; charset=utf-8", + ssa: "text/plain; charset=utf-8", + vtt: "text/plain; charset=utf-8", + sub: "text/plain; charset=utf-8", + sup: "application/octet-stream", + idx: "text/plain; charset=utf-8", +}; + +/** RFC 6266 / 5987:ASCII 兜底名 + UTF-8 原名 */ +function disposition(kind: "inline" | "attachment", name: string): string { + const ascii = name.replace(/[^\x20-\x7e]|["\\]/g, "_"); + return `${kind}; filename="${ascii}"; filename*=UTF-8''${encodeURIComponent(name)}`; +} + +export async function downloadFile( + d: FmDeps, + principal: Principal | undefined, + q: FilesDownloadQuery, +): Promise { + if (!isAdmin(principal)) throw new FilesError("forbidden", "下载只允许控制台管理员会话(API Key 不可用)"); + const r = await resolveIn(d, q.root, q.path); + if (r.rel === "" || r.lst?.isDirectory()) throw new FilesError("bad_request", "只能下载文件,不能下载目录"); + const name = basename(r.abs); + const kind = kindOfName(name); + if (!ALLOWED.has(kind)) + throw new FilesError("forbidden", `只允许下载 NFO/字幕/图片(${kind}),视频等大文件不经浏览器下载`); + const real = await recheck(r.abs, r.rootReal); + const file = await Deno.open(real, { read: true }); + let size: number; + try { + const st = await file.stat(); + if (!st.isFile) throw new FilesError("bad_request", "不是普通文件"); + size = st.size; + if (size > d.options.downloadMaxBytes) + throw new FilesError( + "too_large", + `文件 ${size} 字节,超过下载上限 ${d.options.downloadMaxBytes}(FILES_DOWNLOAD_MAX_BYTES)`, + ); + } catch (e) { + file.close(); + throw e; + } + const ext = extname(name).replace(/^\./, "").toLowerCase(); + // SVG 可带脚本:不内联、不给 image/svg+xml + const inline = q.inline && kind === "image" && ext !== "svg"; + await audit(d, principal, "download", { root: q.root, path: r.rel, size, inline }); + return new Response(file.readable, { + status: 200, + headers: { + "content-type": (ext !== "svg" && MIME[ext]) || "application/octet-stream", + "content-length": String(size), + "content-disposition": disposition(inline ? "inline" : "attachment", name), + "x-content-type-options": "nosniff", + "content-security-policy": "default-src 'none'; sandbox", + "cache-control": "private, no-store", + }, + }); +} diff --git a/apps/server/src/modules/file-manager/organize.ts b/apps/server/src/modules/file-manager/organize.ts index e9322e20..0a30986e 100644 --- a/apps/server/src/modules/file-manager/organize.ts +++ b/apps/server/src/modules/file-manager/organize.ts @@ -1,8 +1,9 @@ import { stat } from "node:fs/promises"; +import { join } from "node:path"; import type { FilesOrganizeGroup, FilesOrganizeRequest, FilesOrganizeResult } from "@umt/contracts"; import type { Principal } from "../../http/auth.ts"; import { ensureTaskRow, markTask } from "../acquisition-organize/services/common.ts"; -import { planImport } from "../acquisition-organize/services/organize.ts"; +import { type FileOverrideInput, planImport } from "../acquisition-organize/services/organize.ts"; import { authorizeMove, hnrBlockForMove, @@ -13,9 +14,12 @@ import { } from "../acquisition-organize/services/organize-guard.ts"; import { actorOfP, errMessage, FilesError, type FmDeps, hashJson, isAdmin, resolveIn } from "./common.ts"; import { audit, filesChanged, loadPlan, retirePlan, savePlan } from "./ledger.ts"; +import { isWritableTarget } from "./list.ts"; -// 整理向导落地:按 (mediaEntityId, contentUnitId) 分组,每组走 planImport(与 POST /organization-records 同一实现, -// 同格式整理记录)。dryRun 冻结计划 + 源文件 size/mtime;执行校验 planId、源事实与重算的前后路径一致, +// 整理向导落地:按 (mediaEntityId, contentUnitId, 操作, 目标目录, Slot) 分组,每组走 planImport(与 POST /organization-records +// 同一实现,同格式整理记录)。逐项季/集/单元覆写以 fileOverrides 传入 planImport(计划前后路径即反映覆写), +// 目标目录覆写 = planImport.targetDir(不改路径绑定),Slot 覆写在核验通过后推进该 Slot;无法应用的覆写才进 warnings。 +// dryRun 冻结计划 + 源文件 size/mtime;执行校验 planId、源事实与重算的前后路径一致, // 任一组 plan_stale / 授权 / H&R 失败只影响该组。 type Req = { @@ -35,6 +39,18 @@ interface Frozen { reason: string | null; } +interface Group { + entity: string; + unit: string | null; + action: "copy" | "move"; + targetDir: string | null; + targetLibraryRootId: string | null; + slotId: string | null; + paths: string[]; + overrides: Record; + warnings: string[]; +} + const pairsOf = (plan: Awaited>) => (plan.plan?.organize.files ?? []).map((f) => ({ src: f.src, dst: f.dst ?? null, kind: f.kind, size: f.size })); @@ -47,6 +63,16 @@ async function factsOf(paths: string[]) { ); } +/** 集号数组 → 连续区间;不连续返回 null */ +function episodeRange(eps: number[]): { start: number; end: number } | null { + const u = [...new Set(eps)].sort((a, b) => a - b); + if (u.length === 0) return null; + for (let i = 1; i < u.length; i++) if (u[i] !== u[i - 1]! + 1) return null; + return { start: u[0]!, end: u[u.length - 1]! }; +} + +const BUSY_SLOT = new Set(["reserved", "downloading", "verifying"]); + export async function organizeFiles( d: FmDeps, principal: Principal | undefined, @@ -56,21 +82,119 @@ export async function organizeFiles( const actorType = principal?.kind === "api_key" ? "api_key" : "user"; const actor = actorOfP(principal); + const entityIds = [...new Set(body.items.map((i) => i.mediaEntityId))]; + const entities = new Map( + (await db.selectFrom("media_entity").select(["id", "kind"]).where("id", "in", entityIds).execute()).map((e) => [ + e.id, + e.kind, + ]), + ); + const unitIds = [...new Set(body.items.map((i) => i.contentUnitId).filter((x): x is string => !!x))]; + const units = new Map( + unitIds.length + ? ( + await db + .selectFrom("content_unit") + .select(["id", "media_entity_id", "kind"]) + .where("id", "in", unitIds) + .execute() + ).map((u) => [u.id, u]) + : [], + ); + const slotIds = [...new Set(body.items.map((i) => i.slotId).filter((x): x is string => !!x))]; + const slots = new Map( + slotIds.length + ? ( + await db + .selectFrom("acquisition_slot") + .select(["id", "media_entity_id", "status"]) + .where("id", "in", slotIds) + .execute() + ).map((s) => [s.id, s]) + : [], + ); + // 分组(保持首次出现顺序) - const groups = new Map(); + const groups = new Map(); const roots = new Set(); for (const it of body.items) { const r = await resolveIn(d, it.root, it.path); if (r.rel === "") throw new FilesError("bad_request", "不能整理整个根"); if (r.isSymlink) throw new FilesError("bad_request", `不整理符号链接: ${r.rel}`); - if (body.action === "move" && !r.def.modifiable) + const action = it.operation ?? body.action; + if (action === "move" && !r.def.modifiable) throw new FilesError("not_writable", `源根 ${r.def.name} 为只读登记,不能 Move`); roots.add(it.root); - const key = `${it.mediaEntityId}|${it.contentUnitId ?? ""}`; - const g = groups.get(key) ?? { entity: it.mediaEntityId, unit: it.contentUnitId ?? null, paths: [], warnings: [] }; + const isDir = !!r.lst?.isDirectory(); + const kind = entities.get(it.mediaEntityId); + const isTv = kind !== undefined && kind !== "movie"; + const warnings: string[] = []; + const ov: FileOverrideInput = {}; + + // 目标目录覆写:writable 库根下已存在的目录 + let targetDir: string | null = null; + let targetLib: string | null = body.targetLibraryRootId ?? null; + if (it.targetDir) { + const t = await resolveIn(d, it.targetDir.root, it.targetDir.path); + if (!t.lst?.isDirectory() || t.isSymlink || !t.def.libraryRootId) + throw new FilesError("bad_request", `目标目录必须是库根下已存在的目录: ${it.targetDir.path || "/"}`); + if (!(await isWritableTarget(t.def, t.rootReal))) + throw new FilesError("not_writable", `目标根 ${t.def.name} 不可写`); + targetDir = join(t.def.hostPath, t.rel); + targetLib = t.def.libraryRootId; + } + // 季/集 + if (it.season !== undefined || it.episodes !== undefined) { + if (!isTv) warnings.push(`${r.rel}: 电影不适用季/集覆写,已忽略`); + else { + if (it.season !== undefined) ov.season = it.season; + if (it.episodes?.length) { + const range = episodeRange(it.episodes); + if (!range) warnings.push(`${r.rel}: 集号 ${it.episodes.join(",")} 不连续,未应用(只支持连续区间)`); + else if (isDir) warnings.push(`${r.rel}: 目录项不能指定集号(会套到其下所有视频),未应用`); + else ov.episodes = range; + } + } + } + // 内容单元 + let unit = it.contentUnitId ?? null; + if (unit) { + const u = units.get(unit); + if (!u || u.media_entity_id !== it.mediaEntityId) { + warnings.push(`${r.rel}: 内容单元 ${unit} 不属于该作品,未应用`); + unit = null; + } else if (u.kind === "episode" || u.kind === "part") ov.contentUnitIds = [u.id]; + } + // Slot + let slotId = it.slotId ?? null; + if (slotId) { + const s = slots.get(slotId); + if (!s || s.media_entity_id !== it.mediaEntityId) { + warnings.push(`${r.rel}: Slot ${slotId} 不属于该作品,未应用`); + slotId = null; + } else if (BUSY_SLOT.has(s.status)) { + warnings.push(`${r.rel}: Slot 正在获取中(${s.status}),不改其状态`); + slotId = null; + } + } + + const key = [it.mediaEntityId, unit ?? "", action, targetDir ?? "", targetLib ?? "", slotId ?? ""].join("|"); + const g = + groups.get(key) ?? + ({ + entity: it.mediaEntityId, + unit, + action, + targetDir, + targetLibraryRootId: targetLib, + slotId, + paths: [], + overrides: {}, + warnings: [], + } satisfies Group); g.paths.push(r.abs); - if (it.season !== undefined || it.episodes !== undefined || it.slotId !== undefined) - g.warnings.push(`${r.rel}: 逐项季/集/Slot 覆写暂未生效(按文件名解析),待 planImport 支持 fileOverrides`); + if (Object.keys(ov).length > 0) g.overrides[r.abs] = ov; + g.warnings.push(...warnings); groups.set(key, g); } const guard = await sourceGuard(db); @@ -82,13 +206,29 @@ export async function organizeFiles( sourcePaths: g.paths, mediaEntityId: g.entity, ...(g.unit ? { contentUnitId: g.unit } : {}), - action: body.action, - ...(body.targetLibraryRootId ? { targetLibraryRootId: body.targetLibraryRootId } : {}), + action: g.action, + ...(g.targetLibraryRootId ? { targetLibraryRootId: g.targetLibraryRootId } : {}), + }; + const hasOverrides = Object.keys(g.overrides).length > 0; + const reqHash = hashJson({ + key: requestKey(req), + ...(hasOverrides ? { overrides: g.overrides } : {}), + ...(g.targetDir ? { targetDir: g.targetDir } : {}), + ...(g.slotId ? { slotId: g.slotId } : {}), + }); + const base = { + mediaEntityId: g.entity, + contentUnitId: g.unit, + warnings: g.warnings, + action: g.action, + targetDir: g.targetDir, + slotId: g.slotId, }; - const reqHash = hashJson(requestKey(req)); - const base = { mediaEntityId: g.entity, contentUnitId: g.unit, warnings: g.warnings }; const importReq = (dryRun: boolean, taskId?: string) => ({ ...req, + ...(hasOverrides ? { fileOverrides: g.overrides } : {}), + ...(g.targetDir ? { targetDir: g.targetDir } : {}), + ...(g.slotId ? { slotId: g.slotId } : {}), trigger: "manual" as const, actorType: actorType as "api_key" | "user", actorId: actor, @@ -140,7 +280,7 @@ export async function organizeFiles( throw new FilesError("plan_stale", "重新计算的前后路径与预览不一致(目标目录或绑定已变化),请重新预览"); if (again.status === "failed") throw new FilesError("bad_request", `计划不可执行:${again.reason ?? "blocked"}`); - if (body.action === "move") { + if (g.action === "move") { const managed = await managedTasksUnder(db, req.sourcePaths); if (!isAdmin(principal)) { const auth = await authorizeMove(d.ao, principal, req, g.entity); diff --git a/apps/server/src/modules/file-manager/records.ts b/apps/server/src/modules/file-manager/records.ts index 7ae5d74c..a46cd7b8 100644 --- a/apps/server/src/modules/file-manager/records.ts +++ b/apps/server/src/modules/file-manager/records.ts @@ -1,8 +1,8 @@ -import { randomUUID } from "node:crypto"; -import type { OrganizationAction } from "@umt/contracts"; +import type { FilesRecordsQuery, FilesRecordsResult, OrganizationAction } from "@umt/contracts"; +import { sql } from "@umt/db"; import type { Principal } from "../../http/auth.ts"; import type { Exec } from "../acquisition-organize/services/common.ts"; -import { emitRecord } from "../acquisition-organize/services/organize.ts"; +import { emitRecord, loadRecordsPublic } from "../acquisition-organize/services/organize.ts"; import { actorOfP, type FmDeps, hashJson, jsonb } from "./common.ts"; // 显式路径变更的数据库同步:asset_location 前缀替换、path_binding 修订(旧 superseded → 新 active)、 @@ -10,105 +10,12 @@ import { actorOfP, type FmDeps, hashJson, jsonb } from "./common.ts"; const likePrefix = (rel: string) => `${rel.replace(/[\\%_]/g, (m) => `\\${m}`)}/%`; -export interface RelocateInput { - fromLib: string; - fromRel: string; - toLib: string; - toRel: string; - isDir: boolean; - actor: string; - opKind: "rename" | "move"; -} - -export interface RelocateResult { - locations: { id: string; assetId: string; from: string; to: string }[]; - bindings: { oldBindingId: string; newBindingId: string }[]; - mediaEntityIds: string[]; -} - -/** 同事务:把 fromLib/fromRel(目录含其下全部)改到 toLib/toRel */ -export async function relocate(trx: Exec, input: RelocateInput): Promise { - const { fromLib, fromRel, toLib, toRel, isDir } = input; - const mapRel = (p: string) => (p === fromRel ? toRel : `${toRel}${p.slice(fromRel.length)}`); - const locs = await trx - .selectFrom("asset_location as l") - .innerJoin("media_asset as a", "a.id", "l.media_asset_id") - .select(["l.id", "l.relative_path", "l.media_asset_id", "a.media_entity_id"]) - .where("l.library_root_id", "=", fromLib) - .where("l.status", "<>", "removed") - .where((eb) => - isDir - ? eb.or([eb("l.relative_path", "=", fromRel), eb("l.relative_path", "like", likePrefix(fromRel))]) - : eb("l.relative_path", "=", fromRel), - ) - .execute(); - const out: RelocateResult = { locations: [], bindings: [], mediaEntityIds: [] }; - const entities = new Set(); - for (const l of locs) { - const to = mapRel(l.relative_path); - await trx - .updateTable("asset_location") - .set({ library_root_id: toLib, relative_path: to, last_seen_at: new Date() }) - .where("id", "=", l.id) - .execute(); - await trx - .insertInto("asset_operation") - .values({ - operation_id: `files:${randomUUID()}`, - phase: "commit", - kind: input.opKind, - status: "succeeded", - intent: jsonb({ - fromLibraryRootId: fromLib, - from: l.relative_path, - toLibraryRootId: toLib, - to, - by: input.actor, - }), - result: jsonb({ locationId: l.id }), - media_asset_id: l.media_asset_id, - started_at: new Date(), - finished_at: new Date(), - }) - .execute(); - out.locations.push({ id: l.id, assetId: l.media_asset_id, from: l.relative_path, to }); - if (l.media_entity_id) entities.add(l.media_entity_id); - } - if (isDir) { - const binds = await trx - .selectFrom("path_binding") - .selectAll() - .where("library_root_id", "=", fromLib) - .where("status", "=", "active") - .where((eb) => eb.or([eb("relative_path", "=", fromRel), eb("relative_path", "like", likePrefix(fromRel))])) - .execute(); - for (const b of binds) { - // 显式路径变更(REQ §7.1.1):旧绑定 superseded,新绑定 active 并互相链接;仅译名刷新永不走到这里 - await trx.updateTable("path_binding").set({ status: "superseded" }).where("id", "=", b.id).execute(); - const nb = await trx - .insertInto("path_binding") - .values({ - media_entity_id: b.media_entity_id, - content_unit_id: b.content_unit_id, - season_number: b.season_number, - binding_kind: b.binding_kind, - library_root_id: toLib, - relative_path: mapRel(b.relative_path), - template_version: b.template_version, - status: "active", - bound_by: input.actor, - note: `files.${input.opKind}: ${b.relative_path} → ${mapRel(b.relative_path)}`, - }) - .returning("id") - .executeTakeFirstOrThrow(); - await trx.updateTable("path_binding").set({ superseded_by_id: nb.id }).where("id", "=", b.id).execute(); - out.bindings.push({ oldBindingId: b.id, newBindingId: nb.id }); - entities.add(b.media_entity_id); - } - } - out.mediaEntityIds = [...entities]; - return out; -} +// 位置跟随/绑定修订与原样搬运(organize-transfer)共用同一实现 +export { + type RelocateInput, + type RelocateResult, + relocate, +} from "../acquisition-organize/services/organize-transfer.ts"; /** 预览:受影响的位置数与绑定(不写库) */ export async function relocatePreview(exec: Exec, lib: string, rel: string, isDir: boolean) { @@ -213,3 +120,42 @@ export async function writeRecord( await emitRecord(d.ao, trx, rec.id); return rec.id; } + +/** + * 整理记录回查(整理页 `/console/organize?ids=` 预填):按记录 id,或按 /files 写操作的 planId + * (move/copy/organize 经 task.idempotency_key = files:;删除/改名等直接写的记录看 plan.planId)。 + */ +export async function findRecords(d: FmDeps, q: FilesRecordsQuery): Promise { + const db = d.ctx.db; + const planRecords: Record = {}; + const planIds = q.planIds ?? []; + if (planIds.length > 0) { + const viaTask = await db + .selectFrom("organization_record as r") + .innerJoin("task as t", "t.id", "r.task_id") + .select(["r.id", "t.idempotency_key"]) + .where( + "t.idempotency_key", + "in", + planIds.map((p) => `files:${p}`), + ) + .orderBy("r.created_at", "desc") + .execute(); + for (const r of viaTask) planRecords[r.idempotency_key.slice("files:".length)] ??= r.id; + const direct = await sql<{ id: string; plan_id: string }>` + select id, plan->>'planId' as plan_id from organization_record + where plan->>'planId' = any(${planIds}::text[]) order by created_at desc`.execute(db); + for (const r of direct.rows) planRecords[r.plan_id] ??= r.id; + } + const ids = [...new Set([...(q.ids ?? []), ...Object.values(planRecords)])]; + const rows = ids.length + ? await db + .selectFrom("organization_record") + .selectAll() + .where("id", "in", ids) + .orderBy("created_at", "desc") + .execute() + : []; + const items = (await loadRecordsPublic(db, rows, q.includeFiles)) as FilesRecordsResult["items"]; + return { items, planRecords }; +} diff --git a/apps/server/src/modules/file-manager/routes.ts b/apps/server/src/modules/file-manager/routes.ts index 6def4126..8afcbc08 100644 --- a/apps/server/src/modules/file-manager/routes.ts +++ b/apps/server/src/modules/file-manager/routes.ts @@ -4,14 +4,17 @@ import type { Context } from "hono"; import type { z } from "zod"; import { type AuthEnv, requireScope } from "../../http/auth.ts"; import { errorBody } from "../../http/errors.ts"; +import { PathGuardError } from "../acquisition-organize/services/organize-guard.ts"; import type { ModuleApp } from "../types.ts"; -import { FilesError, type FmDeps, isAdmin } from "./common.ts"; +import { FilesError, type FmDeps, fsErrorToFiles, isAdmin } from "./common.ts"; +import { downloadFile } from "./download.ts"; import { identify } from "./identify.ts"; import { idempotent } from "./ledger.ts"; import { listDir, listRoots, statEntry } from "./list.ts"; import { mkdirIn, renameEntry } from "./mutate.ts"; import { organizeFiles } from "./organize.ts"; import { probeFiles, scrape } from "./probe.ts"; +import { findRecords } from "./records.ts"; import { transfer } from "./transfer.ts"; import { deleteFiles, listTrash, purgeTrash, restoreTrash } from "./trash.ts"; @@ -22,7 +25,7 @@ import { deleteFiles, listTrash, purgeTrash, restoreTrash } from "./trash.ts"; type Input = { query: any; body: any }; type Handler = (c: Context, input: Input) => Promise; -function mount(app: ModuleApp, def: EndpointDef, handler: Handler) { +function mount(app: ModuleApp, def: EndpointDef, handler: Handler, raw = false) { const route = createRoute({ method: def.method, path: def.path, @@ -34,7 +37,9 @@ function mount(app: ModuleApp, def: EndpointDef, handler: Handler) { ...(def.body ? { body: { content: { "application/json": { schema: def.body } }, required: true } } : {}), }, responses: { - 200: { description: "成功", content: { "application/json": { schema: apiResponse(def.response) } } }, + 200: raw + ? { description: "文件流", content: { "application/octet-stream": { schema: def.response } } } + : { description: "成功", content: { "application/json": { schema: apiResponse(def.response) } } }, }, }); app.openapi(route, (async (c: Context) => { @@ -44,10 +49,12 @@ function mount(app: ModuleApp, def: EndpointDef, handler: Handler) { query: def.query ? v.valid("query") : {}, body: def.body ? v.valid("json") : undefined, }); + if (raw) return data as Response; return c.json({ ok: true as const, data }, 200); } catch (err) { - if (err instanceof FilesError) - return c.json(errorBody(err.code, `${err.reason}: ${err.message}`), err.status as 400); + // 检查与使用之间的文件系统竞态(目录被删、挂载掉线)也按业务错误回,不给 500 + const fe = err instanceof PathGuardError ? new FilesError("path_escape", err.message) : fsErrorToFiles(err); + if (fe) return c.json(errorBody(fe.code, `${fe.reason}: ${fe.message}`), fe.status as 400); throw err; } }) as never); @@ -98,4 +105,6 @@ export function registerFileRoutes(app: ModuleApp, d: FmDeps) { mount(app, F.identify, async (c, { body }) => identify(d, c.get("principal"), body)); mount(app, F.scrape, async (c, { body }) => idem(d, c, "scrape", body, () => scrape(d, c.get("principal"), body))); mount(app, F.probe, async (_c, { body }) => probeFiles(d, body)); + mount(app, F.records, async (_c, { query }) => findRecords(d, query)); + mount(app, F.download, async (c, { query }) => downloadFile(d, c.get("principal"), query), true); } diff --git a/apps/server/src/modules/file-manager/transfer.ts b/apps/server/src/modules/file-manager/transfer.ts index 4cd0f958..567c75d7 100644 --- a/apps/server/src/modules/file-manager/transfer.ts +++ b/apps/server/src/modules/file-manager/transfer.ts @@ -1,29 +1,32 @@ -import { lstat, mkdir, readdir, stat } from "node:fs/promises"; +import { lstat, readdir, stat } from "node:fs/promises"; import { basename, dirname, join } from "node:path"; -import { copyWithVerify, crossDeviceMove, moveSafe } from "@umt/adapters"; -import type { FilePlanPair, FilesMoveRequest, FilesTransferResult } from "@umt/contracts"; +import type { FilePlanPair, FilesMoveRequest, FilesTransferResult, OrganizationStatus } from "@umt/contracts"; import type { Principal } from "../../http/auth.ts"; -import { authorizeMove } from "../acquisition-organize/services/organize-guard.ts"; +import { ensureTaskRow, markTask } from "../acquisition-organize/services/common.ts"; +import { planImport } from "../acquisition-organize/services/organize.ts"; +import { authorizeMove, PathGuardError, sourceGuard } from "../acquisition-organize/services/organize-guard.ts"; +import type { TransferSpec } from "../acquisition-organize/services/organize-transfer.ts"; import { actorOfP, - errMessage, FilesError, type FmDeps, hashJson, isAdmin, type Resolved, - recheck, relOf, resolveIn, within, } from "./common.ts"; import { audit, filesChanged, loadPlan, retirePlan, savePlan } from "./ledger.ts"; import { downloadsFor, isWritableTarget, managedDownloads } from "./list.ts"; -import { type RecordFile, relocate, relocatePreview, writeRecord } from "./records.ts"; +import { relocatePreview } from "./records.ts"; -// 移动/复制到目录:dryRun 冻结源事实(文件 size/mtime/inode;目录递归文件清单指纹)→ planId; -// 执行校验 planId(未过期、请求一致)与源事实未变,否则 409 plan_stale。 -// 搬运原语与整理流程相同(adapters moveSafe / crossDeviceMove / copyWithVerify:no-clobber、跨卷 Copy→核验→删源、EXDEV 不降级)。 +// 移动/复制到目录(W5-A2):与整理共用 planImport(原样搬运变体 transfer)、冻结计划存储与 plan_stale 校验。 +// dryRun:冻结源事实(文件 size/mtime/inode;目录递归文件清单指纹)+ planImport 逐文件前后路径 → planId; +// 执行:校验 planId(未过期、请求一致)、源事实与重算的逐文件路径未变,否则 409 plan_stale; +// 然后写 organization_record 并入队 verifyAndImport(executeRecord:逐文件账本可恢复、fsproxy 挂死隔离、 +// 跨卷 Copy→核验→删源、EXDEV 不降级),请求内最多等 FILES_TRANSFER_WAIT_SEC,超时返回 pending。 +// move 完成后已登记位置跟随、整目录搬完时路径绑定写修订、源侧空目录删除(finalizeTransfer)。 interface SourceSnap { root: string; @@ -81,6 +84,8 @@ async function snapshot(d: FmDeps, r: Resolved, rootId: string): Promise(["succeeded", "partial", "failed", "cancelled"]); + export async function transfer( d: FmDeps, principal: Principal | undefined, @@ -89,7 +94,7 @@ export async function transfer( const mode = body.mode; const t = await resolveIn(d, body.target.root, body.target.path); if (!t.lst?.isDirectory() || t.isSymlink) throw new FilesError("bad_request", "目标必须是已存在的目录"); - if (!(await isWritableTarget(t.def, t.rootReal))) + if (!t.def.libraryRootId || !(await isWritableTarget(t.def, t.rootReal))) throw new FilesError("not_writable", `目标根 ${t.def.name} 不可写(仅 managed 库根可作为目标)`); const sources: Resolved[] = []; @@ -115,60 +120,94 @@ export async function transfer( hnrBlocks.push({ taskId: h.taskId, path: s.rel, hnr: h.hnr, status: h.status }); const targetDev = Number((await stat(t.abs)).dev); + const spec: TransferSpec = { + items: sources.map((s, i) => { + const dst = join(t.abs, names[i]!); + return { + src: s.abs, + dst, + srcLibraryRootId: s.def.libraryRootId, + srcRel: s.def.libraryRootId ? s.rel : null, + dstRel: relOf(t.rootReal, dst), + }; + }), + targetLibraryRootId: t.def.libraryRootId, + targetRootPath: t.rootReal, + }; + const actorType = principal?.kind === "api_key" ? ("api_key" as const) : ("user" as const); + const importReq = (dryRun: boolean, extra: { taskId?: string; guard?: (p: string) => Promise } = {}) => ({ + transfer: { ...spec, ...(extra.guard ? { guard: extra.guard } : {}) }, + action: mode, + trigger: "manual" as const, + actorType, + actorId: actorOfP(principal), + dryRun, + ...(extra.taskId ? { taskId: extra.taskId } : {}), + }); + const planned = await planImport(d.ao, importReq(true)); + if (!planned.plan) { + const reason = planned.reason ?? "plan_failed"; + throw new FilesError( + reason.startsWith("library_root_offline") ? "root_unavailable" : "bad_request", + `无法生成搬运计划:${reason}`, + ); + } + const planFiles = planned.plan.organize.files; + const filePairs = planFiles.map((f) => ({ src: f.src, dst: f.dst })); + const conflictSrc = new Set(planFiles.filter((f) => f.action === "hold").map((f) => f.src)); + const under = (item: string, p: string) => p === item || p.startsWith(`${item}/`); + const pairs: FilePlanPair[] = []; let needBytes = 0; for (const [i, s] of sources.entries()) { - const dst = join(t.abs, names[i]!); - const conflict = !!(await lstat(dst).catch(() => null)); + const dst = spec.items[i]!.dst; const registered = s.def.libraryRootId ? (await relocatePreview(d.ctx.db, s.def.libraryRootId, s.rel, snaps[i]!.isDir)).locations > 0 : false; + const crossDevice = Number((await lstat(s.abs)).dev) !== targetDev; pairs.push({ source: { root: body.items[i]!.root, path: s.rel }, target: { root: body.target.root, path: relOf(t.rootReal, dst) }, size: snaps[i]!.size, - conflict, + conflict: !!(await lstat(dst).catch(() => null)) || [...conflictSrc].some((p) => under(s.abs, p)), registered, + files: snaps[i]!.files.length, + crossDevice, }); - const sameDev = Number((await lstat(s.abs)).dev) === targetDev; - if (!(mode === "move" && sameDev)) needBytes += snaps[i]!.size; + if (!(mode === "move" && !crossDevice)) needBytes += snaps[i]!.size; } const du = await d.ao.ports.diskUsage(t.abs).catch(() => null); const freeBytes = du?.status === "ok" ? du.availableBytes : null; const spaceOk = freeBytes === null ? true : needBytes <= freeBytes; - const truncated = snaps.some((s) => s.truncated); + const truncated = snaps.some((s) => s.truncated) || planned.truncated === true; const totalBytes = snaps.reduce((a, s) => a + s.size, 0); const requestHash = hashJson({ items: body.items, target: body.target, mode }); const request = { sourcePaths: sources.map((s) => s.abs), action: mode, - targetLibraryRootId: t.def.libraryRootId ?? undefined, + targetLibraryRootId: t.def.libraryRootId, }; + const allowCross = (await d.ctx.settings.get("organize")).allowCrossVolumeMove; + const crossBlocked = mode === "move" && !allowCross && pairs.some((p) => p.crossDevice); + const warnings: string[] = []; + if (crossBlocked) warnings.push("跨卷移动未开启(设置 → 整理 → 允许跨卷 Move),执行会被拒绝;可改用复制"); + if (conflictSrc.size > 0) warnings.push(`${conflictSrc.size} 个文件在目标已存在,不覆盖,执行时记为失败`); + if (truncated) warnings.push("源文件数超过上限,请分批操作"); + const base = { mode, pairs, totalBytes, freeBytes, spaceOk, hnrBlocks, truncated, warnings }; if (body.dryRun) { const plan = await savePlan(d, "transfer", requestHash, principal, { request, sources: snaps.map((s) => ({ abs: s.abs, fingerprint: s.fingerprint })), + files: filePairs, pairs, }); - return { - dryRun: true, - planId: plan.id, - planExpiresAt: plan.expiresAt.toISOString(), - mode, - pairs, - totalBytes, - freeBytes, - spaceOk, - hnrBlocks, - truncated, - executed: null, - }; + return { dryRun: true, planId: plan.id, planExpiresAt: plan.expiresAt.toISOString(), ...base, executed: null }; } // ── 执行 ── if (!body.planId) throw new FilesError("bad_request", "执行需要 dryRun 返回的 planId"); - const plan = await loadPlan<{ sources: { abs: string; fingerprint: string }[] }>( + const plan = await loadPlan<{ sources: { abs: string; fingerprint: string }[]; files?: typeof filePairs }>( d, body.planId, "transfer", @@ -179,8 +218,15 @@ export async function transfer( if (!frozen || frozen.abs !== s.abs || frozen.fingerprint !== s.fingerprint) throw new FilesError("plan_stale", `源在预览后发生变化(大小/修改时间/文件清单): ${s.rel}`); } + if (plan.content.files && hashJson(plan.content.files) !== hashJson(filePairs)) + throw new FilesError("plan_stale", "重新计算的前后路径与预览不一致(目标已变化),请重新预览"); if (truncated) throw new FilesError("bad_request", "源目录文件数超过上限,请分批操作"); if (!spaceOk) throw new FilesError("bad_request", `目标卷空间不足:需要 ${needBytes},可用 ${freeBytes}`); + if (crossBlocked) + throw new FilesError( + "bad_request", + "跨卷移动未开启(settings.organize.allowCrossVolumeMove),请改用复制或开启后重试", + ); if (mode === "move") { if (hnrBlocks.length > 0) throw new FilesError( @@ -204,100 +250,89 @@ export async function transfer( } await retirePlan(d, plan.id); - const hash = d.ao.ports.options.verifyHash; + const db = d.ctx.db; + const taskId = await ensureTaskRow(db, { + kind: "verifyAndImport", + idempotencyKey: `files:${plan.id}`, + input: { planId: plan.id, op: "transfer", mode, target: body.target }, + subjectType: "organize_request", + triggeredBy: actorOfP(principal), + }); + const guard = await sourceGuard(db); + let rec: Awaited>; + try { + rec = await planImport(d.ao, importReq(false, { taskId, guard: guard.check })); + } catch (e) { + await markTask(db, taskId, "failed", { error: e instanceof Error ? e.message : String(e) }); + if (e instanceof PathGuardError) throw new FilesError("path_escape", e.message); + throw e; + } + if (!rec.recordId || rec.status === "failed") { + await markTask(db, taskId, "failed", { error: rec.reason ?? rec.status }); + throw new FilesError("bad_request", `搬运计划不可执行:${rec.reason ?? rec.status}`); + } + const recordId = rec.recordId; + await db.updateTable("organization_record").set({ task_id: taskId }).where("id", "=", recordId).execute(); + const { runId } = await d.ctx.engine.enqueue( + "verifyAndImport", + { recordId, taskId }, + { idempotencyKey: `verifyAndImport:files:${plan.id}`, traceId: taskId }, + ); + await db.updateTable("task").set({ engine_run_id: runId }).where("id", "=", taskId).execute(); + + // 请求内等待执行完成(小操作即时出结果);超时返回 pending,后台继续,可按 recordId 追踪 + const deadline = Date.now() + d.options.transferWaitMs; + let status: OrganizationStatus = "running"; + for (;;) { + const r = await db + .selectFrom("organization_record") + .select("status") + .where("id", "=", recordId) + .executeTakeFirstOrThrow(); + status = r.status; + if (TERMINAL.has(status) || Date.now() >= deadline) break; + await new Promise((res) => setTimeout(res, 150)); + } + const pending = !TERMINAL.has(status); const succeeded: string[] = []; const failed: { path: string; code: string; message: string }[] = []; - const recFiles: RecordFile[] = []; - const moved: { s: Resolved; dstRel: string; isDir: boolean }[] = []; - for (const [i, s] of sources.entries()) { - const pair = pairs[i]!; - const dst = join(t.abs, names[i]!); - const fail = (code: string, message: string) => { - failed.push({ path: s.rel, code, message }); - recFiles.push({ src: s.abs, dst: pair.target.path, size: pair.size, status: "failed", error: message }); - }; - try { - await recheck(s.abs, s.rootReal); - if (await lstat(dst).catch(() => null)) { - fail("exists", "目标已存在,不覆盖"); + if (!pending) { + const results = await db + .selectFrom("organization_file_result") + .select(["source_path", "status", "error"]) + .where("organization_record_id", "=", recordId) + .execute(); + for (const s of sources) { + const mine = results.filter((f) => under(s.abs, f.source_path) && f.status !== "skipped"); + const bad = mine.find((f) => f.status !== "succeeded"); + if (!bad) { + succeeded.push(s.rel); continue; } - const isDir = snaps[i]!.isDir; - if (mode === "move") { - const r = await moveSafe(s.abs, dst); - if (r.status === "exists") { - fail("exists", "目标已存在,不覆盖"); - continue; - } - if (r.status === "exdev") { - if (isDir) { - fail("exdev_dir", "跨卷移动目录暂不支持(W5),请先复制再删除,或逐个文件移动"); - continue; - } - const x = await crossDeviceMove(s.abs, dst, { hash }); - if (x.status !== "completed") { - fail(x.status, `跨卷移动未完成(${x.status}):${JSON.stringify(x.steps.at(-1) ?? {})}`); - continue; - } - } - moved.push({ s, dstRel: pair.target.path, isDir }); - } else if (isDir) { - for (const f of snaps[i]!.files) { - await copyWithVerify(join(s.abs, f.rel), join(dst, f.rel), { hash, createParents: true }); - } - await mkdir(dst, { recursive: true }); - } else { - await copyWithVerify(s.abs, dst, { hash }); - } - succeeded.push(s.rel); - recFiles.push({ src: s.abs, dst: pair.target.path, size: pair.size, status: "succeeded" }); - } catch (e) { - fail(e instanceof FilesError ? e.reason : "io_error", errMessage(e)); + const err = (bad.error ?? {}) as { code?: string; message?: string }; + failed.push({ + path: s.rel, + code: err.code === "EEXIST" ? "exists" : (err.code ?? bad.status), + message: `${err.message ?? bad.status}${mine.length > 1 ? `(${mine.filter((f) => f.status === "succeeded").length}/${mine.length} 个文件已完成)` : ""}`, + }); } } - - const recordId = await d.ctx.db.transaction().execute(async (trx) => { - const entities = new Set(); - if (mode === "move" && t.def.libraryRootId) { - for (const m of moved) { - if (!m.s.def.libraryRootId) continue; - const rr = await relocate(trx, { - fromLib: m.s.def.libraryRootId, - fromRel: m.s.rel, - toLib: t.def.libraryRootId, - toRel: m.dstRel, - isDir: m.isDir, - actor: actorOfP(principal), - opKind: "move", - }); - for (const e of rr.mediaEntityIds) entities.add(e); - } - } - return writeRecord(d, trx, { - principal, - action: mode === "move" ? "relocate" : "copy", - mediaEntityId: entities.size === 1 ? [...entities][0]! : null, - sourcePath: sources.length === 1 ? sources[0]!.abs : dirname(sources[0]!.abs), - targetLibraryRootId: t.def.libraryRootId, - targetPath: t.rel, - plan: { op: "transfer", mode, planId: plan.id, pairs }, - files: recFiles, - }); + await audit(d, principal, mode, { + planId: plan.id, + succeeded, + failed, + recordId, + taskId, + pending, + target: body.target, }); - await audit(d, principal, mode, { planId: plan.id, succeeded, failed, recordId, target: body.target }); for (const it of body.items) filesChanged(d, it.root, [it.path.includes("/") ? dirname(it.path) : ""]); filesChanged(d, body.target.root, [t.rel]); return { dryRun: false, planId: plan.id, planExpiresAt: plan.expiresAt.toISOString(), - mode, - pairs, - totalBytes, - freeBytes, - spaceOk, - hnrBlocks, - truncated, - executed: { succeeded, failed, recordId }, + ...base, + executed: { succeeded, failed, recordId, taskId, status, pending }, }; } diff --git a/apps/server/src/modules/file-manager/trash.ts b/apps/server/src/modules/file-manager/trash.ts index b3c7f2bc..6710a148 100644 --- a/apps/server/src/modules/file-manager/trash.ts +++ b/apps/server/src/modules/file-manager/trash.ts @@ -34,6 +34,7 @@ import { } from "./common.ts"; import { audit, consumeApproval, filesChanged, loadPlan, retirePlan, savePlan } from "./ledger.ts"; import { bindingsAt, downloadsFor, isDownloading, managedDownloads } from "./list.ts"; +import { type RecordFile, writeRecord } from "./records.ts"; // 删除(原则 5):默认进回收站(同卷 <根>/.umt-trash///<原相对路径>,O(1) rename; // 不可写或跨卷时复制到 DATA_DIR/trash,超过上限拒绝)。引用检查:纳管做种 / 已登记资产 / 路径绑定 / 根本身。 @@ -172,7 +173,7 @@ export async function deleteFiles( if (!isAdmin(principal)) throw new FilesError("forbidden", "永久删除只允许控制台管理员会话"); } const downloads = await managedDownloads(d); - const results = []; + const results: Awaited>[] = []; for (const it of body.items) results.push(await checkItem(d, body, it, downloads)); const checks = results.map((r) => r.check); const requestHash = hashJson({ @@ -221,30 +222,38 @@ export async function deleteFiles( const deleted: string[] = []; const failed: FilesDeleteResult["failed"] = []; const done: typeof results = []; + // 整理记录的文件级结果(删除/回收站也进整理记录页可追溯) + const recFiles: RecordFile[] = []; for (const res of results) { const { check, ctx } = res; const abs = ctx.abs!; + const inner = check.isDir ? await filesUnder(abs) : null; try { await recheck(abs, ctx.rootReal!); + let where: string | null = null; if (body.mode === "permanent") { await rm(abs, { recursive: true, force: false }); deleted.push(check.path); } else { const row = await moveToTrash(d, principal, check, ctx, expiresAt); trashed.push(toTrashItem(row)); + where = row.trash_path; } + recFiles.push(...recordFilesOf(abs, where, check.size, inner, "succeeded")); done.push(res); } catch (e) { - failed.push({ + const f = { path: check.path, code: e instanceof FilesError ? e.reason : (errnoOf(e) ?? "io_error"), message: errMessage(e), - }); + }; + failed.push(f); + recFiles.push({ src: abs, dst: null, size: check.size, status: "failed", error: `${f.code}: ${f.message}` }); } } - // 资产退役 / 解除绑定(只对已实际删除/移入回收站的项) - await d.ctx.db.transaction().execute(async (trx) => { + // 资产退役 / 解除绑定(只对已实际删除/移入回收站的项)+ 整理记录 + const recordId = await d.ctx.db.transaction().execute(async (trx) => { const locIds = done.flatMap((r) => r.check.assets.map((a) => a.locationId)); if (locIds.length > 0) { await trx.updateTable("asset_location").set({ status: "removed" }).where("id", "in", locIds).execute(); @@ -281,6 +290,28 @@ export async function deleteFiles( .set({ status: "superseded", note: `files.delete by ${actorOfP(principal)}` }) .where("id", "in", bindIds) .execute(); + if (recFiles.length === 0) return null; + const libs = [...new Set(results.map((r) => r.ctx.libraryRootId))]; + const entities = [...new Set(done.flatMap((r) => r.check.assets.map((a) => a.mediaEntityId)))]; + return writeRecord(d, trx, { + principal, + action: body.mode === "permanent" ? "delete" : "trash", + mediaEntityId: entities.length === 1 ? entities[0]! : null, + sourcePath: results.length === 1 ? results[0]!.ctx.abs! : dirname(results[0]!.ctx.abs!), + targetLibraryRootId: null, + targetPath: null, + plan: { + op: "delete", + mode: body.mode, + planId, + libraryRootId: libs.length === 1 ? libs[0] : null, + items: checks.map((c) => ({ root: c.root, path: c.path, isDir: c.isDir, size: c.size })), + trashed: trashed.map((t) => ({ id: t.id, trashPath: t.trashPath })), + retiredLocations: done.flatMap((r) => r.check.assets.map((a) => a.locationId)), + unboundBindings: bindIds, + }, + files: recFiles, + }); }); await audit(d, principal, body.mode === "permanent" ? "delete_permanent" : "delete", { @@ -289,6 +320,7 @@ export async function deleteFiles( trashed: trashed.map((t) => t.id), deleted, failed, + recordId, }); for (const c of checks) filesChanged(d, c.root, [c.path.includes("/") ? dirname(c.path) : ""]); return { @@ -300,10 +332,28 @@ export async function deleteFiles( trashed, deleted, failed, - recordId: null, + recordId, }; } +/** 目录下的文件(相对路径);超过上限返回 null(整理记录只记目录本身一行) */ +const RECORD_FILES_MAX = 2000; +async function filesUnder(abs: string): Promise { + const { files } = await sizeOf(abs, RECORD_FILES_MAX + 1).catch(() => ({ files: [] as string[] })); + return files.length > RECORD_FILES_MAX ? null : files; +} + +function recordFilesOf( + abs: string, + where: string | null, + size: number, + inner: string[] | null, + status: RecordFile["status"], +): RecordFile[] { + if (!inner || inner.length === 0) return [{ src: abs, dst: where, size, status }]; + return inner.map((f) => ({ src: join(abs, f), dst: where ? join(where, f) : null, size: 0, status })); +} + async function copyTree(src: string, dst: string, files: string[], hash: "size" | "sha256" | "sample") { const st = await lstat(src); if (!st.isDirectory()) { @@ -423,6 +473,8 @@ export async function restoreTrash( ): Promise { const restored: FilesTrashRestoreResult["restored"] = []; const failed: FilesTrashRestoreResult["failed"] = []; + const recFiles: RecordFile[] = []; + const libs = new Set(); const target = body.target ? await resolveIn(d, body.target.root, body.target.path) : null; if (target && (!target.lst?.isDirectory() || target.isSymlink)) throw new FilesError("bad_request", "还原目标必须是已存在的目录"); @@ -465,18 +517,34 @@ export async function restoreTrash( } await d.ctx.db.deleteFrom("trash_item").where("id", "=", id).execute(); const rel = dst === rootReal ? "" : dst.slice(rootReal.length + 1); + libs.add(target ? target.def.libraryRootId : row.library_root_id); + recFiles.push({ src: row.trash_path, dst: rel, size: Number(row.size_bytes), status: "succeeded" }); restored.push({ id, root: rootId as FilesTrashRestoreResult["restored"][number]["root"], path: rel }); filesChanged(d, rootId, [rel.includes("/") ? dirname(rel) : ""]); } catch (e) { - failed.push({ - id, - code: e instanceof FilesError ? e.reason : (errnoOf(e) ?? "io_error"), - message: errMessage(e), - }); + const f = { id, code: e instanceof FilesError ? e.reason : (errnoOf(e) ?? "io_error"), message: errMessage(e) }; + failed.push(f); + recFiles.push({ src: `trash:${id}`, dst: null, size: 0, status: "failed", error: `${f.code}: ${f.message}` }); } } - await audit(d, principal, "trash_restore", { restored, failed }); - return { restored, failed }; + const lib = libs.size === 1 ? [...libs][0]! : null; + const recordId = + restored.length > 0 + ? await d.ctx.db.transaction().execute((trx) => + writeRecord(d, trx, { + principal, + action: "restore", + mediaEntityId: null, + sourcePath: recFiles.find((f) => f.status === "succeeded")!.src, + targetLibraryRootId: lib, + targetPath: restored.length === 1 ? restored[0]!.path : null, + plan: { op: "trash_restore", ids: body.ids, target: body.target ?? null, restored }, + files: recFiles, + }), + ) + : null; + await audit(d, principal, "trash_restore", { restored, failed, recordId }); + return { restored, failed, recordId }; } /** 清除:先删行(复核仍存在)再删文件;ids 缺省 = 只清过期项 */ diff --git a/apps/server/test/file-manager/harness.ts b/apps/server/test/file-manager/harness.ts index 772bf5bd..0bcd8093 100644 --- a/apps/server/test/file-manager/harness.ts +++ b/apps/server/test/file-manager/harness.ts @@ -14,6 +14,7 @@ import { hashApiKey } from "../../src/http/auth.ts"; import { EventBus } from "../../src/http/sse.ts"; import { createLogger } from "../../src/logger.ts"; import { fakeEngine } from "../../src/modules/acquisition-organize/__tests__/harness.ts"; +import { executeRecord, type FrozenPlan } from "../../src/modules/acquisition-organize/services/organize.ts"; import { createCapabilityRegistry } from "../../src/modules/agent/capability-registry.ts"; import { createFileManagerModule, type FmOverrides } from "../../src/modules/file-manager/index.ts"; import type { ServerContext } from "../../src/modules/types.ts"; @@ -47,6 +48,7 @@ type Loose = any; export interface FmHarness { db: Kysely; engine: ReturnType; + module: ReturnType; events: { type: string; payload: Loose }[]; base: string; lib: string; @@ -57,13 +59,16 @@ export interface FmHarness { libraryRootId: string; instanceId: string; api(method: string, path: string, body?: unknown, key?: string): Promise<{ status: number; body: Loose }>; + /** 原始响应(非 JSON 接口,如 /files/download) */ + raw(method: string, path: string, key?: string): Promise; close(): Promise; } export async function createFmHarness( name: string, overrides: FmOverrides = {}, - opts: { catalog?: boolean } = {}, + /** runTransfers(默认 true):假引擎收到 /files 移动/复制的 verifyAndImport 时就地执行(模拟 worker) */ + opts: { catalog?: boolean; runTransfers?: boolean } = {}, ): Promise { const schema = `fm_test_${name}_${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`; const admin = new pg.Client({ connectionString: DB_URL }); @@ -144,8 +149,30 @@ export async function createFmHarness( ...overrides.ao, options: { backgroundLoops: false, minVideoBytes: 10, verifyHash: "sha256", ...(overrides.ao?.options ?? {}) }, }, - options: { backgroundLoops: false, dataDir: join(base, "data"), ...overrides.options }, + options: { backgroundLoops: false, dataDir: join(base, "data"), transferWaitMs: 10_000, ...overrides.options }, }); + if (opts.runTransfers !== false) { + const enqueue = engine.enqueue.bind(engine); + engine.enqueue = async (task, input, o) => { + const r = await enqueue(task, input, o); + const recordId = (input as { recordId?: string }).recordId; + if (task === "verifyAndImport" && recordId) + setTimeout(async () => { + try { + const rec = await db + .selectFrom("organization_record") + .select("plan") + .where("id", "=", recordId) + .executeTakeFirst(); + if ((rec?.plan as unknown as FrozenPlan | undefined)?.transfer) + await executeRecord(module.deps().ao, recordId); + } catch { + // 模拟 worker:失败留给测试断言(记录保持 running) + } + }, 0); + return r; + }; + } const app = createApp({ config: { NODE_ENV: "development", DEV_ADMIN_KEY: DEV_KEY }, logger, @@ -178,6 +205,7 @@ export async function createFmHarness( return { db, engine, + module, events, base, lib, @@ -198,6 +226,9 @@ export async function createFmHarness( }); return { status: res.status, body: await res.json() }; }, + raw(method, path, key = DEV_KEY) { + return Promise.resolve(app.request(`/api/v1${path}`, { method, headers: { authorization: `Bearer ${key}` } })); + }, async close() { await db.destroy(); const c = new pg.Client({ connectionString: DB_URL }); diff --git a/apps/server/test/file-manager/w5a2.test.ts b/apps/server/test/file-manager/w5a2.test.ts new file mode 100644 index 00000000..46be3f88 --- /dev/null +++ b/apps/server/test/file-manager/w5a2.test.ts @@ -0,0 +1,414 @@ +import { randomUUID } from "node:crypto"; +import { access, copyFile, mkdir, mkdtemp, readFile, rm, symlink, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { findCrossDeviceBase } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { executeRecord } from "../../src/modules/acquisition-organize/services/organize.ts"; +import { fsErrorToFiles } from "../../src/modules/file-manager/common.ts"; +import { + createFmHarness, + EXEC_KEY, + type FmHarness, + key, + pgAvailable, + READ_KEY, + seedAsset, + seedBinding, + seedEntity, + sql, +} from "./harness.ts"; + +// W5-A2:移动/复制走 planImport(目录跨卷逐文件可恢复)、整理逐项覆写生效、删除/还原写整理记录、 +// 下载守卫、记录回查、文件系统竞态不 500。 + +const skip = !(await pgAvailable()); +const xdev = findCrossDeviceBase(); +const exists = (p: string) => + access(p).then( + () => true, + () => false, + ); + +describe.skipIf(skip)("files W5-A2", () => { + let h: FmHarness; + beforeAll(async () => { + h = await createFmHarness( + "w5a2", + { options: { allowPermanentDelete: true, downloadMaxBytes: 100, transferWaitMs: 200 } }, + { runTransfers: false }, + ); + }); + afterAll(async () => h?.close()); + + it.skipIf(!xdev)( + "目录跨卷移动:逐文件 Copy→核验→删源;中途 kill 后续跑 resumed 对账;位置跟随、绑定修订、删空目录", + async () => { + // 第二个库根放在另一设备上 + const xlib = await mkdtemp(join(xdev!, "fm-xlib-")); + try { + await writeFile(join(xlib, ".uhd-root"), "sentinel\n"); + await mkdir(join(xlib, "tv")); + const x = await h.db + .insertInto("library_root") + .values({ name: "跨卷库", media_kind: "mixed", local_path: xlib, management_mode: "managed" }) + .returning("id") + .executeTakeFirstOrThrow(); + const show = await seedEntity(h.db, { kind: "series", title: "Show", year: 2020 }); + await mkdir(join(h.lib, "Show", "extras"), { recursive: true }); + await writeFile(join(h.lib, "Show", "Show.S01E01.mkv"), "episode-one"); + await writeFile(join(h.lib, "Show", "Show.S01E02.mkv"), "episode-two"); + await writeFile(join(h.lib, "Show", "extras", "bonus.nfo"), ""); + const a = await seedAsset(h.db, h, { entityId: show, rel: "Show/Show.S01E01.mkv" }); + const bind = await seedBinding(h.db, h, { entityId: show, rel: "Show" }); + + const body = { + items: [{ root: h.libRoot, path: "Show" }], + target: { root: `lib:${x.id}`, path: "tv" }, + mode: "move", + }; + // 跨卷 Move 未开启:预览给 warning,执行 400(不产生记录) + await sql`insert into settings (key, value) values ('organize', ${JSON.stringify({ allowCrossVolumeMove: false })}::jsonb) + on conflict (key) do update set value = excluded.value`.execute(h.db); + const off = await h.api("POST", "/files/move", { ...body, idempotencyKey: key(), dryRun: true }); + expect(off.body.data.warnings.join()).toMatch(/跨卷移动未开启/); + const offExec = await h.api("POST", "/files/move", { + ...body, + idempotencyKey: key(), + dryRun: false, + planId: off.body.data.planId, + }); + expect(offExec.status).toBe(400); + await sql`insert into settings (key, value) values ('organize', ${JSON.stringify({ allowCrossVolumeMove: true })}::jsonb) + on conflict (key) do update set value = excluded.value`.execute(h.db); + + const dry = await h.api("POST", "/files/move", { ...body, idempotencyKey: key(), dryRun: true }); + expect(dry.status).toBe(200); + expect(dry.body.data.pairs[0]).toMatchObject({ + files: 3, + crossDevice: true, + registered: true, + conflict: false, + }); + expect(dry.body.data.warnings).toEqual([]); + + // 执行:假引擎不跑 worker → 等待窗口内未完成,返回 pending + recordId/taskId + const exec = await h.api("POST", "/files/move", { + ...body, + idempotencyKey: key(), + dryRun: false, + planId: dry.body.data.planId, + }); + expect(exec.status).toBe(200); + const ex = exec.body.data.executed; + expect(ex).toMatchObject({ pending: true, succeeded: [], failed: [], status: "running" }); + expect(ex.taskId).toBeTruthy(); + const recordId = ex.recordId as string; + expect(h.engine.enqueued.at(-1)).toMatchObject({ task: "verifyAndImport", input: { recordId } }); + + // 模拟上一次 attempt:第一个文件已复制到目标,删源前 worker 被 kill(账本停在 started) + const first = await h.db + .selectFrom("organization_file_result") + .select(["ordinal", "source_path", "target_path"]) + .where("organization_record_id", "=", recordId) + .where("source_path", "=", join(h.lib, "Show", "Show.S01E01.mkv")) + .executeTakeFirstOrThrow(); + expect(first.target_path).toBe("tv/Show/Show.S01E01.mkv"); + await mkdir(join(xlib, "tv", "Show"), { recursive: true }); + await copyFile(first.source_path, join(xlib, first.target_path!)); + await sql`insert into tasks_op_ledger (op_key, attempt_token, attempts, status) + values (${`org:${recordId}:${first.ordinal}`}, ${randomUUID()}, 1, 'started')`.execute(h.db); + + const r = await executeRecord(h.module.deps().ao, recordId); + expect(r.status).toBe("succeeded"); + expect(await readFile(join(xlib, "tv", "Show", "Show.S01E01.mkv"), "utf8")).toBe("episode-one"); + expect(await readFile(join(xlib, "tv", "Show", "Show.S01E02.mkv"), "utf8")).toBe("episode-two"); + expect(await readFile(join(xlib, "tv", "Show", "extras", "bonus.nfo"), "utf8")).toBe(""); + // 源目录(含空子目录)已删 + expect(await exists(join(h.lib, "Show"))).toBe(false); + const op = await h.db + .selectFrom("asset_operation") + .select("result") + .where("operation_id", "=", `org:${recordId}:${first.ordinal}`) + .executeTakeFirstOrThrow(); + const res = op.result as { resumed?: boolean; steps?: { step: string; status: string }[] }; + expect(res.resumed).toBe(true); + expect(res.steps?.find((s) => s.step === "copy")?.status).toBe("skipped"); + // 位置跟随到新库根;绑定写修订 + const loc = await h.db + .selectFrom("asset_location") + .select(["library_root_id", "relative_path"]) + .where("id", "=", a.locationId) + .executeTakeFirstOrThrow(); + expect(loc).toEqual({ library_root_id: x.id, relative_path: "tv/Show/Show.S01E01.mkv" }); + const old = await h.db + .selectFrom("path_binding") + .select(["status", "superseded_by_id"]) + .where("id", "=", bind) + .executeTakeFirstOrThrow(); + expect(old.status).toBe("superseded"); + const nb = await h.db + .selectFrom("path_binding") + .select(["library_root_id", "relative_path", "status"]) + .where("id", "=", old.superseded_by_id!) + .executeTakeFirstOrThrow(); + expect(nb).toEqual({ library_root_id: x.id, relative_path: "tv/Show", status: "active" }); + const rec = await h.db + .selectFrom("organization_record") + .select(["action", "status", "file_succeeded", "media_entity_id"]) + .where("id", "=", recordId) + .executeTakeFirstOrThrow(); + expect(rec).toEqual({ action: "move", status: "succeeded", file_succeeded: 3, media_entity_id: show }); + // 重复 finalize 幂等 + expect((await executeRecord(h.module.deps().ao, recordId)).status).toBe("succeeded"); + + // planId 回查整理记录(整理页预填) + const q = await h.api( + "GET", + `/files/records?planIds=${dry.body.data.planId}&includeFiles=true`, + undefined, + READ_KEY, + ); + expect(q.status).toBe(200); + expect(q.body.data.planRecords[dry.body.data.planId]).toBe(recordId); + expect(q.body.data.items[0].files).toHaveLength(3); + } finally { + await sql`delete from settings where key = 'organize'`.execute(h.db); + await rm(xlib, { recursive: true, force: true }); + } + }, + 60_000, + ); + + it("复制目录走同一计划:目标已存在的文件不覆盖、记失败(partial),其余照常", async () => { + await mkdir(join(h.dl, "Pack", "sub"), { recursive: true }); + await writeFile(join(h.dl, "Pack", "a.mkv"), "aaa"); + await writeFile(join(h.dl, "Pack", "sub", "b.srt"), "bbb"); + await mkdir(join(h.lib, "dest", "Pack"), { recursive: true }); + await writeFile(join(h.lib, "dest", "Pack", "a.mkv"), "old"); + const body = { items: [{ root: h.dlRoot, path: "Pack" }], target: { root: h.libRoot, path: "dest" } }; + const dry = await h.api("POST", "/files/copy", { ...body, idempotencyKey: key(), dryRun: true }); + expect(dry.status).toBe(200); + expect(dry.body.data.pairs[0]).toMatchObject({ files: 2, conflict: true }); + expect(dry.body.data.warnings.join()).toMatch(/已存在/); + const exec = await h.api("POST", "/files/copy", { + ...body, + idempotencyKey: key(), + dryRun: false, + planId: dry.body.data.planId, + }); + const recordId = exec.body.data.executed.recordId; + const r = await executeRecord(h.module.deps().ao, recordId); + expect(r.status).toBe("partial"); + expect(await readFile(join(h.lib, "dest", "Pack", "a.mkv"), "utf8")).toBe("old"); + expect(await readFile(join(h.lib, "dest", "Pack", "sub", "b.srt"), "utf8")).toBe("bbb"); + expect(await exists(join(h.dl, "Pack", "a.mkv"))).toBe(true); + }); + + it("整理逐项覆写:季/集改变计划路径;目标目录与操作覆写分组;无法应用的才进 warnings;Slot 覆写在核验后生效", async () => { + const series = await seedEntity(h.db, { kind: "series", title: "剧", year: 2020 }); + const film = await seedEntity(h.db, { kind: "movie", title: "Film", year: 2019 }); + await writeFile(join(h.dl, "weird-a.mkv"), "x".repeat(64)); + await writeFile(join(h.dl, "weird-b.mkv"), "y".repeat(64)); + await writeFile(join(h.dl, "Film.2019.1080p.mkv"), "z".repeat(64)); + await mkdir(join(h.lib, "picked"), { recursive: true }); + const slot = await h.db + .insertInto("acquisition_slot") + .values({ media_entity_id: film, name: "Film 主版本", status: "empty" } as never) + .returning("id") + .executeTakeFirstOrThrow(); + const dry = await h.api("POST", "/files/organize", { + idempotencyKey: key(), + action: "copy", + dryRun: true, + items: [ + { root: h.dlRoot, path: "weird-a.mkv", mediaEntityId: series, season: 2, episodes: [3, 4] }, + { root: h.dlRoot, path: "weird-b.mkv", mediaEntityId: series, season: 2, episodes: [5, 7] }, + { + root: h.dlRoot, + path: "Film.2019.1080p.mkv", + mediaEntityId: film, + season: 1, + slotId: slot.id, + targetDir: { root: h.libRoot, path: "picked" }, + operation: "copy", + }, + ], + }); + expect(dry.status).toBe(200); + const groups = dry.body.data.groups; + expect(groups).toHaveLength(2); + const tv = groups.find((g: { mediaEntityId: string }) => g.mediaEntityId === series); + const mv = groups.find((g: { mediaEntityId: string }) => g.mediaEntityId === film); + const tvTargets = tv.files.map((f: { target: string | null }) => f.target); + expect(tvTargets.some((t: string | null) => t?.endsWith("Season 2/剧 - S02E03-E04.mkv"))).toBe(true); + // 非连续集号不应用(只按季覆写,文件名无集号 → 计划排除),其余覆写不进 warnings + expect(tv.warnings).toEqual([expect.stringMatching(/weird-b\.mkv: 集号 5,7 不连续/)]); + expect(mv.warnings).toEqual([expect.stringMatching(/电影不适用季\/集覆写/)]); + expect(mv).toMatchObject({ action: "copy", slotId: slot.id, targetDir: join(h.lib, "picked") }); + expect(mv.files[0].target).toBe(join(h.lib, "picked", "Film (2019) - 1080p.mkv")); + + const exec = await h.api("POST", "/files/organize", { + idempotencyKey: key(), + action: "copy", + dryRun: false, + planIds: groups.map((g: { planId: string }) => g.planId), + items: [ + { + root: h.dlRoot, + path: "Film.2019.1080p.mkv", + mediaEntityId: film, + season: 1, + slotId: slot.id, + targetDir: { root: h.libRoot, path: "picked" }, + operation: "copy", + }, + ], + }); + expect(exec.status).toBe(200); + const g = exec.body.data.groups[0]; + expect(g.error).toBeNull(); + await executeRecord(h.module.deps().ao, g.recordId); + expect(await readFile(join(h.lib, "picked", "Film (2019) - 1080p.mkv"), "utf8")).toBe("z".repeat(64)); + const s = await h.db + .selectFrom("acquisition_slot") + .select(["status", "current_asset_id"]) + .where("id", "=", slot.id) + .executeTakeFirstOrThrow(); + expect(s.current_asset_id).not.toBeNull(); + expect(s.status).toBe("satisfied"); + // 目标目录覆写不新建路径绑定 + const binds = await h.db.selectFrom("path_binding").select("id").where("media_entity_id", "=", film).execute(); + expect(binds).toEqual([]); + }); + + it("删除/永久删除/还原都写整理记录(文件级结果)", async () => { + await mkdir(join(h.lib, "gone", "in"), { recursive: true }); + await writeFile(join(h.lib, "gone", "x.srt"), "1"); + await writeFile(join(h.lib, "gone", "in", "y.srt"), "2"); + const del = await h.api("POST", "/files/delete", { + idempotencyKey: key(), + items: [{ root: h.libRoot, path: "gone" }], + dryRun: false, + }); + expect(del.status).toBe(200); + const recId = del.body.data.recordId; + expect(recId).toBeTruthy(); + const rec = await h.db + .selectFrom("organization_record") + .select(["action", "status", "file_total", "file_succeeded"]) + .where("id", "=", recId) + .executeTakeFirstOrThrow(); + expect(rec).toEqual({ action: "trash", status: "succeeded", file_total: 2, file_succeeded: 2 }); + const files = await h.db + .selectFrom("organization_file_result") + .select(["source_path", "target_path", "mode"]) + .where("organization_record_id", "=", recId) + .orderBy("ordinal") + .execute(); + expect(files.map((f) => f.source_path).sort()).toEqual( + [join(h.lib, "gone", "in", "y.srt"), join(h.lib, "gone", "x.srt")].sort(), + ); + expect(files.every((f) => f.mode === "trash" && f.target_path?.includes(".umt-trash"))).toBe(true); + + const rs = await h.api("POST", "/files/trash/restore", { + idempotencyKey: key(), + ids: [del.body.data.trashed[0].id], + }); + expect(rs.status).toBe(200); + expect(rs.body.data.restored).toHaveLength(1); + const rrec = await h.db + .selectFrom("organization_record") + .select(["action", "status", "target_library_root_id", "target_path"]) + .where("id", "=", rs.body.data.recordId) + .executeTakeFirstOrThrow(); + expect(rrec).toEqual({ + action: "restore", + status: "succeeded", + target_library_root_id: h.libraryRootId, + target_path: "gone", + }); + + const perm = await h.api("POST", "/files/delete", { + idempotencyKey: key(), + items: [{ root: h.libRoot, path: "gone/x.srt" }], + mode: "permanent", + confirmName: "x.srt", + dryRun: false, + }); + expect(perm.status).toBe(200); + const prec = await h.db + .selectFrom("organization_record") + .select(["action", "file_total"]) + .where("id", "=", perm.body.data.recordId) + .executeTakeFirstOrThrow(); + expect(prec).toEqual({ action: "delete", file_total: 1 }); + // 按 planId/记录 id 回查 + const q = await h.api("GET", `/files/records?ids=${recId},${perm.body.data.recordId}`); + expect(q.body.data.items.map((i: { action: string }) => i.action).sort()).toEqual(["delete", "trash"]); + }); + + it("下载:仅管理员会话;只允许 nfo/字幕/图片且不超上限;路径守卫;流式 + 审计", async () => { + await mkdir(join(h.lib, "dl-test"), { recursive: true }); + await writeFile(join(h.lib, "dl-test", "海报.jpg"), "JPEGDATA"); + await writeFile(join(h.lib, "dl-test", "movie.nfo"), ""); + await writeFile(join(h.lib, "dl-test", "big.srt"), "x".repeat(200)); + await writeFile(join(h.lib, "dl-test", "v.mkv"), "video"); + await symlink(join(h.outside, "secret.txt"), join(h.lib, "dl-test", "leak.nfo")); + const url = (path: string, extra = "") => + `/files/download?root=${encodeURIComponent(h.libRoot)}&path=${encodeURIComponent(path)}${extra}`; + + const ok = await h.raw("GET", url("dl-test/海报.jpg", "&inline=true")); + expect(ok.status).toBe(200); + expect(await ok.text()).toBe("JPEGDATA"); + expect(ok.headers.get("content-type")).toBe("image/jpeg"); + expect(ok.headers.get("content-disposition")).toMatch(/^inline; .*filename\*=UTF-8''%E6%B5%B7%E6%8A%A5\.jpg/); + expect(ok.headers.get("x-content-type-options")).toBe("nosniff"); + const nfo = await h.raw("GET", url("dl-test/movie.nfo", "&inline=true")); + expect(nfo.headers.get("content-disposition")).toMatch(/^attachment;/); + expect(await nfo.text()).toBe(""); + + expect((await h.api("GET", url("dl-test/movie.nfo"), undefined, READ_KEY)).status).toBe(403); + expect((await h.api("GET", url("dl-test/movie.nfo"), undefined, EXEC_KEY)).status).toBe(403); + const video = await h.api("GET", url("dl-test/v.mkv")); + expect(video.status).toBe(403); + const big = await h.api("GET", url("dl-test/big.srt")); + expect(big.status).toBe(413); + expect(big.body.error.message).toMatch(/^too_large:/); + const leak = await h.api("GET", url("dl-test/leak.nfo")); + expect(leak.status).toBe(400); + expect(leak.body.error.message).toMatch(/^path_escape:/); + expect((await h.api("GET", url("dl-test"))).status).toBe(400); + expect((await h.api("GET", url("dl-test/none.nfo"))).status).toBe(404); + expect((await h.api("GET", url("../outside/secret.txt"))).status).toBeGreaterThanOrEqual(400); + + const audits = await h.db + .selectFrom("audit_event") + .select(["action"]) + .where("action", "=", "files.download") + .execute(); + expect(audits.length).toBe(2); + }); + + it("文件系统竞态/离线不 500:errno 映射;下载根目录整个不见 → 503", async () => { + const e = (code: string) => Object.assign(new Error(code), { code }); + expect(fsErrorToFiles(e("ENOTCONN"))?.status).toBe(503); + expect(fsErrorToFiles(e("EIO"))?.reason).toBe("root_unavailable"); + expect(fsErrorToFiles(e("ENOENT"))?.status).toBe(404); + expect(fsErrorToFiles(e("EACCES"))?.status).toBe(403); + expect(fsErrorToFiles(e("ELOOP"))?.status).toBe(400); + expect(fsErrorToFiles(new Error("x"))).toBeNull(); + + const hidden = `${h.dl}.away`; + await rm(hidden, { recursive: true, force: true }); + await import("node:fs/promises").then((fs) => fs.rename(h.dl, hidden)); + try { + const list = await h.api("GET", `/files?root=${encodeURIComponent(h.dlRoot)}&path=`); + expect(list.status).toBe(503); + expect(list.body.error.message).toMatch(/^root_unavailable:/); + const st = await h.api("GET", `/files/stat?root=${encodeURIComponent(h.dlRoot)}&path=x`); + expect(st.status).toBe(503); + } finally { + await import("node:fs/promises").then((fs) => fs.rename(hidden, h.dl)); + } + }); +}); diff --git a/apps/server/test/file-manager/write.test.ts b/apps/server/test/file-manager/write.test.ts index 2f69d38f..c45ec07e 100644 --- a/apps/server/test/file-manager/write.test.ts +++ b/apps/server/test/file-manager/write.test.ts @@ -210,6 +210,8 @@ describe.skipIf(skip)("files 写操作", () => { size: 8, conflict: false, registered: false, + files: 1, + crossDevice: false, }, ]); const planId = dry.body.data.planId; @@ -271,7 +273,8 @@ describe.skipIf(skip)("files 写操作", () => { .select(["action", "status"]) .where("id", "=", exec.body.data.executed.recordId) .executeTakeFirstOrThrow(); - expect(rec).toEqual({ action: "relocate", status: "succeeded" }); + // W5-A2:搬运走 planImport → verifyAndImport,记录 action 与整理一致(move/copy) + expect(rec).toEqual({ action: "move", status: "succeeded" }); }); it("TOCTOU:预览后把源换成指向根外的符号链接,执行被拒绝且根外文件不动", async () => { diff --git a/apps/web/app/lib/labels.ts b/apps/web/app/lib/labels.ts index 7e6239fd..feca5986 100644 --- a/apps/web/app/lib/labels.ts +++ b/apps/web/app/lib/labels.ts @@ -65,6 +65,9 @@ const ORG_ACTION: Record = { rename: "重命名", relocate: "目录调整", hardlink: "硬链接", + trash: "移入回收站", + delete: "永久删除", + restore: "从回收站还原", }; export const organizationActionLabel = (v: string) => ORG_ACTION[v] ?? v; diff --git a/apps/web/app/routes/console/organize.$id.tsx b/apps/web/app/routes/console/organize.$id.tsx index 4646a941..52b3cbce 100644 --- a/apps/web/app/routes/console/organize.$id.tsx +++ b/apps/web/app/routes/console/organize.$id.tsx @@ -17,6 +17,9 @@ const ACTION_LABEL = { rename: "重命名", relocate: "目录调整", hardlink: "硬链接", + trash: "移入回收站", + delete: "永久删除", + restore: "从回收站还原", } as const; const VIEW_LABEL = { downloader: "下载器视角", local: "执行进程本地", library: "库根相对" } as const; const FILE_STATUS: Record = { diff --git a/apps/web/app/routes/console/organize.tsx b/apps/web/app/routes/console/organize.tsx index f957057a..314556aa 100644 --- a/apps/web/app/routes/console/organize.tsx +++ b/apps/web/app/routes/console/organize.tsx @@ -20,6 +20,9 @@ const ACTION_LABEL = { rename: "重命名", relocate: "目录调整", hardlink: "硬链接", + trash: "移入回收站", + delete: "永久删除", + restore: "从回收站还原", } as const; function Counts({ r }: { r: PublicOrganizationRecord }) { diff --git a/docs/runbook/files.md b/docs/runbook/files.md index 0a9b2d65..40961898 100644 --- a/docs/runbook/files.md +++ b/docs/runbook/files.md @@ -1,7 +1,7 @@ # 文件管理(`/files`,W5-A) 实现:`apps/server/src/modules/file-manager/`(路由 `routes.ts`,服务按操作分文件),契约 `packages/contracts/src/files.ts`(端点登记在 `api.ts` 的 `Api.files`),迁移 `0024_trash`。规格见 `docs/plan/06-file-manager-spec.md`。 -复用 acquisition-organize 的服务导出(`planImport`、`sourceGuard`、`authorizeMove`、`hnrBlockForMove`、`managedTasksUnder`、`emitRecord`、H&R 判定),不改它的文件。 +复用 acquisition-organize 的服务导出(`planImport`、`sourceGuard`、`authorizeMove`、`hnrBlockForMove`、`managedTasksUnder`、`emitRecord`、H&R 判定)。W5-A2 给 `planImport` 加了可选参数(`fileOverrides` / `targetDir` / `slotId` / `transfer`,老调用不变),原样搬运的计划与终态在 `acquisition-organize/services/organize-transfer.ts`(`planTransfer` / `finalizeTransfer` / `relocate`)。 ## 根与路径 @@ -19,27 +19,30 @@ - move / copy / organize / delete:先 `dryRun:true` 拿 `planId`(存 `artifact kind=plan, title=files_plan`,默认 30 分钟有效,`FILES_PLAN_TTL_SEC`),执行时回传。 - 执行校验:planId 存在、未过期、未用过,且与本次请求体一致(否则 409 `plan_stale`);源文件事实未变——文件看 size/mtime/inode,目录看递归文件清单(路径+size+mtime)指纹;organize 另外重算一次 dryRun,前后路径必须与冻结计划一致。 - 幂等:写请求带 `idempotencyKey`,按「调用者 + 操作 + 键」唯一(`artifact title=files_idem`,唯一索引 `artifact_files_idem_key`)。同键重放返回首次结果;同键不同请求体或首个请求仍在执行 → 409。执行失败不占键。dryRun 不占键。 -- 所有执行都写 `audit_event`(`files.`)并发 SSE `files.changed {rootId, dirs[]}`;路径变更同时写 `organization_record`(rename → `rename`,移动 → `relocate`,复制 → `copy`,整理 → planImport 原记录),外部程序按 changeSeq 增量感知。 +- 所有执行都写 `audit_event`(`files.`)并发 SSE `files.changed {rootId, dirs[]}`;路径变更同时写 `organization_record`(rename → `rename`,移动/复制 → planImport 记录 `move`/`copy`,整理 → planImport 原记录,删除 → `trash`/`delete`,回收站还原 → `restore`),外部程序按 changeSeq 增量感知。 +- 整理页回查:`GET /files/records?ids=` 或 `?planIds=`(move/copy/organize 经 `task.idempotency_key = files:`,删除等看 `plan.planId`),返回公开整理记录 + `planRecords{planId→recordId}`;`includeFiles=true` 带逐文件结果。 ## 各操作要点 | 操作 | 要点 | |---|---| | rename | 同目录;`sanitizeSegment` 不一致 400;目标存在 409 且原文件不动(moveSafe:link+unlink / lstat+rename,事后 inode 比对);视频同主干 sidecar 一起改;已登记位置同事务改 `asset_location.relative_path` + `asset_operation` 账本;目录命中 `path_binding`(含子绑定):旧行 `superseded` 并 `superseded_by_id` 指向新 active 行;纳管下载内容 409;DB 失败把文件改回原名 | -| move/copy | 目标为 writable 库根下已存在目录;源≠目标、目标不在源内;不覆盖;跨卷文件 Copy→核验→删源(`crossDeviceMove`),跨卷目录移动暂不支持(该项失败,W5);move 触及纳管下载任务 → 409(请用整理入库,Move 会按 H&R 移交);API Key 的 move 需 `authorizeMove`(持续授权或一次性审批,plan_type `organize_plan`/`files_plan` 绑定 planId+内容哈希);已登记资产位置跟随(跨库根会改 `library_root_id`),绑定目录写修订 | -| organize | 按 `(mediaEntityId, contentUnitId)` 分组,每组一次 `planImport`,与 `POST /organization-records` 同实现;执行需 `planIds`(按请求匹配各组);单组失败只影响该组(`groups[].error`);成功组入队 `verifyAndImport`,返回 `recordIds` | -| delete | 默认回收站:`<根>/.umt-trash///<原相对路径>`(同卷 rename);根不可写/跨卷 → `DATA_DIR/trash/...` 复制(上限 `FILES_TRASH_CROSS_MAX_BYTES`,默认 2 GiB,超出请永久删除)。引用检查:纳管任务未完成或 H&R 未满足 409;已登记资产需 `retireAssets`(location→removed、asset 无其他位置→retired、满足的 Slot→attention + 待处理);绑定目录需 `unbind`(→superseded);根本身/只读根拒绝;离线 503。永不连带下载器任务 | +| move/copy | 目标为 writable 库根下已存在目录;源≠目标、目标不在源内;不覆盖。**走 `planImport` 的原样搬运变体**(`transfer`):目录递归成逐文件计划(保持文件名与结构,空子目录照建,符号链接不搬),dryRun 冻结源指纹 + 逐文件前后路径,执行时两者任一变化 → 409 `plan_stale`;执行写 `organization_record(move/copy)` + 入队 `verifyAndImport`(`executeRecord`:逐文件账本 `org::` 可恢复、fsproxy 挂死隔离、跨卷 Copy→核验→删源、EXDEV 不降级)。跨卷 Move 需 `settings.organize.allowCrossVolumeMove`(关闭时预览 `warnings` 提示、执行 400)。目标已存在的文件记失败、不覆盖。请求内最多等 `FILES_TRANSFER_WAIT_SEC`(默认 20 s):完成则 `executed.succeeded/failed` 为最终结果,否则 `executed.pending=true`,后台继续,按 `executed.recordId/taskId` 追踪。move 完成后(`finalizeTransfer`,记录行锁):已登记 `asset_location` 逐文件跟随(跨库根改 `library_root_id`),整项搬完时目录命中的 `path_binding` 写修订,源侧空目录删除。move 触及纳管下载任务 → 409(请用整理入库);API Key 的 move 需 `authorizeMove` | +| organize | 按 `(mediaEntityId, contentUnitId, operation, targetDir, slotId)` 分组,每组一次 `planImport`,与 `POST /organization-records` 同实现;执行需 `planIds`(按请求匹配各组);单组失败只影响该组(`groups[].error`);成功组入队 `verifyAndImport`,返回 `recordIds`。**逐项覆写生效**:`season`/`episodes`(连续区间)→ `planImport.fileOverrides`,决定季目录、文件命名与集单元覆盖;`contentUnitId`(集/分段单元)→ 额外覆盖;`targetDir`(writable 库根下已存在目录)→ 文件按模板命名直接放进去,不用也不建路径绑定;`operation` → 该项的 copy/move;`slotId` → 核验通过后以新资产满足该 Slot(获取中的 Slot 不动)。`groups[].warnings` 只列无法应用的覆写(电影的季/集、非连续集号、目录项指定集号、不属于该作品的单元/Slot、获取中的 Slot) | +| delete | 默认回收站:`<根>/.umt-trash///<原相对路径>`(同卷 rename);根不可写/跨卷 → `DATA_DIR/trash/...` 复制(上限 `FILES_TRASH_CROSS_MAX_BYTES`,默认 2 GiB,超出请永久删除)。引用检查:纳管任务未完成或 H&R 未满足 409;已登记资产需 `retireAssets`(location→removed、asset 无其他位置→retired、满足的 Slot→attention + 待处理);绑定目录需 `unbind`(→superseded);根本身/只读根拒绝;离线 503。永不连带下载器任务。执行写 `organization_record(action=trash|delete)`,文件级结果(目录展开到 2000 个文件,超过只记目录一行),返回 `recordId` | | 永久删除 | `FILES_ALLOW_PERMANENT_DELETE=true` 才可用,且仅管理员会话;单项需 `confirmName` = 文件名 | -| 回收站 | `GET /files/trash`;`restore`(原位被占 → 该项 `exists`,可带 `target` 还原到别处;还原不恢复资产状态,需要时重新整理登记);`purge` 仅管理员会话,缺省只清过期项。持久任务 `purgeTrash` 每日入队(`FILES_TRASH_RETENTION_DAYS`,默认 7);清除前复核行仍存在,且路径必须在某根的 `.umt-trash` 或 `DATA_DIR/trash` 下 | +| 回收站 | `GET /files/trash`;`restore`(原位被占 → 该项 `exists`,可带 `target` 还原到别处;还原不恢复资产状态,需要时重新整理登记;有成功项时写 `organization_record(action=restore)`,返回 `recordId`);`purge` 仅管理员会话,缺省只清过期项。持久任务 `purgeTrash` 每日入队(`FILES_TRASH_RETENTION_DAYS`,默认 7);清除前复核行仍存在,且路径必须在某根的 `.umt-trash` 或 `DATA_DIR/trash` 下 | | identify | 只读:`parseReleasePath`(文件 + 父/祖父目录)→ 本地召回(别名 pg_trgm + 外部身份)→ `catalog.lookup` 能力取远端候选(失败只给 warning)→ `rankCandidates` 前 5;目录展开视频上限 200;不建作品、不进待识别池 | | probe | ≤20 项;已登记资产写回 `media_asset.media_info.probe`;未登记缓存 24 h(`DATA_DIR/cache/probe`) | | scrape | 仅有 active 资产的作品;入队 `refreshMetadata`(字段锁优先,不动路径),返回 `lockedFields`;`writeNfo/images` 目前返回 `not_implemented` | +| download | `GET /files/download?root=&path=&inline=`:仅管理员会话(API Key 含 admin scope 也 403);只允许 NFO/字幕/图片(视频 403),≤ `FILES_DOWNLOAD_MAX_BYTES`(默认 20 MiB,超出 413 `too_large:`);路径守卫同其它接口,打开前再 realpath 复核、按文件句柄 stat;流式返回(非 JSON 信封),`inline=true` 仅对图片生效(SVG 一律附件),带 `nosniff` + `CSP sandbox`;写审计 `files.download` | ## 授权矩阵(实现) | 操作 | 管理员会话 / DEV_ADMIN_KEY | API Key | |---|---|---| -| roots / list / stat / identify / probe / trash 列表 | ✓ | `organize.read` | +| roots / list / stat / identify / probe / trash 列表 / records | ✓ | `organize.read` | +| download | ✓(仅会话) | ✗ 403 | | mkdir / rename / copy / organize(copy) / restore / scrape | ✓ | `organize.execute` | | move / organize(move) | ✓ | `organize.execute` + 持续授权或一次性审批 | | delete(回收站) | ✓ | `organize.execute` + 每次一次性审批(dryRun 的 planId) | @@ -47,11 +50,14 @@ ## 环境变量 -`FILES_LIST_MAX_ENTRIES`(20000)、`FILES_TRASH_RETENTION_DAYS`(7)、`FILES_TRASH_CROSS_MAX_BYTES`(2147483648)、`FILES_PLAN_TTL_SEC`(1800)、`FILES_ALLOW_PERMANENT_DELETE`(false)。 +`FILES_LIST_MAX_ENTRIES`(20000)、`FILES_TRASH_RETENTION_DAYS`(7)、`FILES_TRASH_CROSS_MAX_BYTES`(2147483648)、`FILES_PLAN_TTL_SEC`(1800)、`FILES_ALLOW_PERMANENT_DELETE`(false)、`FILES_TRANSFER_WAIT_SEC`(20,移动/复制请求内等待执行完成的秒数)、`FILES_DOWNLOAD_MAX_BYTES`(20971520)。 -## 已知缺口(W5) +## 错误兜底 -- `GET /files/download`(小文件下载/图片预览)未实现。 -- organize 的逐项 `season/episodes/slotId` 覆写暂不生效(返回 `groups[].warnings`),需 `planImport` 支持 `fileOverrides`;移动/复制未走 `planImport` 的"原样搬运"变体(`targetDir/keepNames`),而是用同一批 adapters 搬运原语 + 本模块冻结计划。 -- 删除不写 `organization_record`(`OrganizationAction` 无 `delete`),只写审计;跨卷目录移动、NFO/图片投影、硬链接/空目录清理均为 W5。 +处理器里漏出的文件系统错误(检查与使用之间目录被删、挂载掉线等竞态)不再 500:离线类 errno(EIO/ENOTCONN/ESTALE/ETIMEDOUT/EHOSTDOWN/ENODEV)→ 503 `root_unavailable`,ENOENT/ENOTDIR → 404,EACCES/EPERM → 403,其余 errno → 400(`fsErrorToFiles`)。 + +## 已知缺口 + +- NFO/图片投影(scrape 的 `writeNfo/images`)、硬链接、蓝光原盘整目录操作仍未做。 +- 搬运 move 在终态提交前删源侧空目录;若进程恰在两者之间退出,重跑 finalize 幂等,但极端情况下可能留下空目录(无数据风险)。 - 路径守卫未下沉到 `packages/adapters/src/fs/guard.ts`(本模块自带按根解析的守卫,整理仍用 `sourceGuard`)。 diff --git a/packages/contracts/src/api.ts b/packages/contracts/src/api.ts index be074f74..c2b7f4d3 100644 --- a/packages/contracts/src/api.ts +++ b/packages/contracts/src/api.ts @@ -1177,6 +1177,22 @@ export const Api = { body: files.FilesProbeRequest, response: files.FilesProbeResult, }), + download: ep({ + method: "get", + path: "/files/download", + scope: "admin", + summary: "下载小文件(nfo/字幕/图片,≤20 MiB;流式;服务端另要求 admin 控制台会话;审计)", + query: files.FilesDownloadQuery, + response: files.FilesDownloadResponse, + }), + records: ep({ + method: "get", + path: "/files/records", + scope: "organize.read", + summary: "按整理记录 id 或 planId 回查整理记录(整理页预填)", + query: files.FilesRecordsQuery, + response: files.FilesRecordsResult, + }), }, // ── 能力 ── diff --git a/packages/contracts/src/files.ts b/packages/contracts/src/files.ts index 1551a081..c93d30e0 100644 --- a/packages/contracts/src/files.ts +++ b/packages/contracts/src/files.ts @@ -3,6 +3,7 @@ import { Id, JsonObject, NonNegInt, Timestamp } from "./common.ts"; import { MediaKind } from "./content.ts"; import { DownloadStatus } from "./execution.ts"; import { InstanceHealth, LibraryMediaKind } from "./instances.ts"; +import { OrganizationStatus, PublicOrganizationRecord } from "./organization.ts"; import { SlotStatus } from "./policy.ts"; // 文件管理(`/files`,docs/plan/06-file-manager-spec.md)。 @@ -285,6 +286,10 @@ export const FilePlanPair = z.object({ conflict: z.boolean(), /** 已登记资产(执行后同步 asset_location) */ registered: z.boolean(), + /** 该项包含的文件数(目录递归;文件为 1)。W5-A2 */ + files: NonNegInt.optional(), + /** 与目标不在同一设备:move 逐文件 Copy→核验→删源(需 settings.organize.allowCrossVolumeMove)。W5-A2 */ + crossDevice: z.boolean().optional(), }); export type FilePlanPair = z.infer; @@ -310,11 +315,20 @@ export const FilesTransferResult = z.object({ /** 执行结果(dryRun=false) */ executed: z .object({ + /** 全部文件已完成的项(源相对路径);pending=true 时为空 */ succeeded: z.array(z.string()), failed: z.array(FileItemFailure), recordId: Id.nullable(), + /** W5-A2:搬运走 planImport → verifyAndImport 任务(逐文件可恢复);任务 id 供任务页追踪 */ + taskId: Id.nullable().optional(), + /** 整理记录状态(等待窗口内的最新值) */ + status: OrganizationStatus.optional(), + /** 等待窗口(FILES_TRANSFER_WAIT_SEC)内未完成:后台继续执行,按 recordId 查整理记录或等 SSE files.changed */ + pending: z.boolean().optional(), }) .nullable(), + /** W5-A2:预览提示(跨卷 Move 未开启、目标已存在的文件等) */ + warnings: z.array(z.string()).optional(), }); export type FilesTransferResult = z.infer; @@ -326,6 +340,10 @@ export const FilesOrganizeItem = FileRef.extend({ season: z.number().int().min(0).max(9999).optional(), episodes: z.array(z.number().int().min(0).max(99999)).max(50).optional(), slotId: Id.optional(), + /** W5-A2:逐项操作方式(缺省取请求的 action);不同操作分到不同组 */ + operation: FilesTransferMode.optional(), + /** W5-A2:目标目录覆写(writable 库根下已存在目录;不建路径绑定,文件按模板命名放入该目录) */ + targetDir: FileRef.optional(), }); export type FilesOrganizeItem = z.infer; @@ -350,8 +368,14 @@ export const FilesOrganizeGroup = z.object({ truncated: z.boolean(), recordId: Id.nullable(), taskId: Id.nullable(), - /** 暂未生效的逐项覆写(season/episodes/slotId 等待 planImport 扩 fileOverrides) */ + /** 无法应用的逐项覆写(如电影项的季/集、非连续集号、不属于该作品的 Slot/单元);能应用的覆写已反映在 files 里 */ warnings: z.array(z.string()), + /** W5-A2:该组的操作方式(逐项 operation 覆写后) */ + action: FilesTransferMode.optional(), + /** W5-A2:该组的目标目录覆写(本机绝对路径) */ + targetDir: z.string().nullable().optional(), + /** W5-A2:该组核验通过后推进的 Slot */ + slotId: Id.nullable().optional(), /** 该组失败原因(plan_stale / 授权 / H&R 等),其他组不受影响 */ error: z.object({ code: z.string(), message: z.string() }).nullable(), }); @@ -469,6 +493,8 @@ export type FilesTrashRestoreRequest = z.infer; export const FilesTrashRestoreResult = z.object({ restored: z.array(z.object({ id: Id, root: FileRootId, path: z.string() })), failed: z.array(z.object({ id: Id, code: z.string(), message: z.string() })), + /** W5-A2:还原写的整理记录(action=restore);无成功项为 null */ + recordId: Id.nullable().optional(), }); export type FilesTrashRestoreResult = z.infer; @@ -583,6 +609,45 @@ export type FileProbeItem = z.infer; export const FilesProbeResult = z.object({ items: z.array(FileProbeItem) }); export type FilesProbeResult = z.infer; +// 下载(W5-A2):GET /files/download 直接返回文件流(非 JSON 信封);仅管理员会话; +// 仅 nfo/字幕/图片且 ≤ FILES_DOWNLOAD_MAX_BYTES(默认 20 MiB);inline=true 仅对图片生效(预览)。 +export const FilesDownloadQuery = z.object({ root: FileRootId, path: FileRelPath, inline: boolQuery.default(false) }); +export type FilesDownloadQuery = z.infer; +/** OpenAPI 描述用:实际响应为二进制流 */ +export const FilesDownloadResponse = z.string().describe("文件内容(application/octet-stream 等)"); + +// 整理记录回查(W5-A2,只读):整理页 `/console/organize?ids=` 预填;planIds 为 /files/move|copy|organize 的 planId + +const csvIds = z + .string() + .max(8000) + .transform((s, ctx) => { + const parts = s + .split(",") + .map((x) => x.trim()) + .filter(Boolean); + for (const p of parts) + if (!Id.safeParse(p).success) { + ctx.addIssue({ code: "custom", message: `非法 id: ${p}` }); + return z.NEVER; + } + return parts.slice(0, 200); + }); +export const FilesRecordsQuery = z.object({ + /** 逗号分隔的整理记录 id */ + ids: csvIds.optional(), + /** 逗号分隔的 planId(执行后映射到其整理记录) */ + planIds: csvIds.optional(), + includeFiles: boolQuery.default(false), +}); +export type FilesRecordsQuery = z.infer; +export const FilesRecordsResult = z.object({ + items: z.array(PublicOrganizationRecord), + /** planId → recordId(未执行或已过期的 planId 不在其中) */ + planRecords: z.record(z.string(), z.string()), +}); +export type FilesRecordsResult = z.infer; + /** SSE `files.changed` 负载 */ export const FilesChangedEvent = z.object({ rootId: FileRootId, dirs: z.array(z.string()) }); export type FilesChangedEvent = z.infer; diff --git a/packages/contracts/src/organization.ts b/packages/contracts/src/organization.ts index ac04b605..73b249e5 100644 --- a/packages/contracts/src/organization.ts +++ b/packages/contracts/src/organization.ts @@ -14,8 +14,20 @@ export const OrganizationTrigger = z.enum(["auto", "manual", "agent"]); export type OrganizationTrigger = z.infer; export const OrganizationStatus = z.enum(["queued", "running", "succeeded", "partial", "failed", "cancelled"]); export type OrganizationStatus = z.infer; -/** Copy / Move / 重命名 / 目录调整 / 硬链接(仅兼容既有) */ -export const OrganizationAction = z.enum(["copy", "move", "rename", "relocate", "hardlink"]); +/** + * Copy / Move / 重命名 / 目录调整 / 硬链接(仅兼容既有)/ 移入回收站 / 永久删除 / 从回收站还原 + * (后三者只由文件管理的显式删除/还原产生,记录已结束,不经 planImport) + */ +export const OrganizationAction = z.enum([ + "copy", + "move", + "rename", + "relocate", + "hardlink", + "trash", + "delete", + "restore", +]); export type OrganizationAction = z.infer; /** 源路径所在视图:下载器视角 / 执行进程本地 / 库根相对 */ export const SourcePathView = z.enum(["downloader", "local", "library"]); diff --git a/packages/db/migrations/0026_organization_file_actions.ts b/packages/db/migrations/0026_organization_file_actions.ts new file mode 100644 index 00000000..8e3b4b21 --- /dev/null +++ b/packages/db/migrations/0026_organization_file_actions.ts @@ -0,0 +1,30 @@ +import type { Kysely } from "kysely"; +import { execDdl, oneOf } from "../src/migration-helpers.ts"; + +// W5-A2 文件管理:删除 / 回收站 / 还原也写整理记录(organization_record + 逐文件结果),整理记录页可追溯。 +// action / mode 取值新增 trash(移入回收站)、delete(永久删除)、restore(从回收站还原)。 + +const OLD = ["copy", "move", "rename", "relocate", "hardlink"] as const; +const NEW = [...OLD, "trash", "delete", "restore"] as const; + +const swap = (values: readonly string[]) => ` +alter table organization_record drop constraint organization_record_action_check; +alter table organization_record add constraint organization_record_action_check ${oneOf("action", values)}; +alter table organization_file_result drop constraint organization_file_result_mode_check; +alter table organization_file_result add constraint organization_file_result_mode_check ${oneOf("mode", values)}; +`; + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl(db, swap(NEW)); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl( + db, + `delete from organization_file_result where mode in ('trash', 'delete', 'restore'); +delete from organization_record where action in ('trash', 'delete', 'restore'); +${swap(OLD)}`, + ); +} diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index 7edecf62..5a491115 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -154,12 +154,28 @@ export type MediaServerInstanceHealthStatus = "unknown" | "healthy" | "degraded" export type MetadataFieldSource = "tmdb" | "user" | "agent" | "nfo" | "site" | "import"; export type ObservationStatus = "new" | "linked" | "ignored" | "error"; export type OrganizationFileResultRole = "main" | "subtitle" | "audio" | "bluray_dir" | "nfo" | "image" | "other"; -export type OrganizationFileResultMode = "copy" | "move" | "rename" | "relocate" | "hardlink"; +export type OrganizationFileResultMode = + | "copy" + | "move" + | "rename" + | "relocate" + | "hardlink" + | "trash" + | "delete" + | "restore"; export type OrganizationFileResultStatus = "queued" | "running" | "succeeded" | "failed" | "skipped" | "cancelled"; export type OrganizationRecordTrigger = "auto" | "manual" | "agent"; export type OrganizationRecordActorType = "user" | "agent" | "system" | "api_key"; export type OrganizationRecordSourcePathView = "downloader" | "local" | "library"; -export type OrganizationRecordAction = "copy" | "move" | "rename" | "relocate" | "hardlink"; +export type OrganizationRecordAction = + | "copy" + | "move" + | "rename" + | "relocate" + | "hardlink" + | "trash" + | "delete" + | "restore"; export type OrganizationRecordStatus = "queued" | "running" | "succeeded" | "partial" | "failed" | "cancelled"; export type OutboxEventStatus = "pending" | "dispatched" | "failed" | "dead"; export type PathBindingBindingKind = "entity" | "season"; diff --git a/packages/domain/src/organize/index.ts b/packages/domain/src/organize/index.ts index d6ac2d35..30e3b324 100644 --- a/packages/domain/src/organize/index.ts +++ b/packages/domain/src/organize/index.ts @@ -2,11 +2,13 @@ export { type ExcludedFile, type ExistingPolicy, type ExistingTarget, + type FileOverride, type OrganizeConflict, type OrganizeFileKind, type OrganizeInput, type OrganizeOperation, type OrganizePlan, + overrideFor, type PlannedAction, type PlannedFile, type Precondition, diff --git a/packages/domain/src/organize/plan.ts b/packages/domain/src/organize/plan.ts index 8f628197..bbe8aede 100644 --- a/packages/domain/src/organize/plan.ts +++ b/packages/domain/src/organize/plan.ts @@ -54,6 +54,32 @@ export interface OrganizeInput { parse?: ParseOptions; /** 随行字幕里此语言加 .default 标记(对照 MP DEFAULT_SUB);不设不加 */ defaultSubtitleLang?: DefaultSubtitleLang; + /** + * 逐文件季/集覆写(人工确认,优先于文件名解析):键为源文件路径或其所在目录(最长前缀生效)。 + * 只影响剧集的季目录选择与命名;电影忽略。 + */ + overrides?: Readonly>; +} + +export interface FileOverride { + season?: number; + episodes?: { start: number; end: number }; +} + +/** 源路径的覆写:精确命中优先,否则取最长的目录前缀 */ +export function overrideFor( + overrides: Readonly> | undefined, + path: string, +): T | undefined { + if (!overrides) return undefined; + const exact = overrides[path]; + if (exact) return exact; + let best: { len: number; o: T } | undefined; + for (const [k, o] of Object.entries(overrides)) { + const dir = k.replace(/\/+$/, ""); + if (path.startsWith(`${dir}/`) && (!best || dir.length > best.len)) best = { len: dir.length, o }; + } + return best?.o; } export type PlannedAction = "transfer" | "replace" | "skip" | "hold"; @@ -195,6 +221,12 @@ export function planOrganize(input: OrganizeInput): OrganizePlan { continue; } const parsed = parseReleasePath(segmentsFor(f.path, input.sourceRoot), input.parse ?? {}); + const ov = input.mediaType === "tv" ? overrideFor(input.overrides, f.path) : undefined; + if (ov?.season !== undefined) parsed.season = ov.season; + if (ov?.episodes) { + parsed.episodes = ov.episodes; + delete parsed.airDate; + } if (parsed.contentType === "extra" && parsed.evidence.includes("dir:extras")) { excluded.push({ src: f.path, kind, reason: "extra_content" }); continue; diff --git a/packages/domain/test/organize.test.ts b/packages/domain/test/organize.test.ts index 7fc20870..54f9b94e 100644 --- a/packages/domain/test/organize.test.ts +++ b/packages/domain/test/organize.test.ts @@ -270,6 +270,44 @@ describe("W4-E4:字幕 .default 与 latest 旧版本判定口径(对照 MP)", () expect(subs.some((n) => n?.endsWith(".eng.srt"))).toBe(true); }); + it("逐文件季/集覆写:优先于文件名解析,目录前缀对其下文件生效,电影忽略", () => { + const D = "/dl/Weird.Name"; + const plan = planOrganize({ + operation: "copy", + mediaType: "tv", + sourceRoot: D, + files: [ + { path: `${D}/weird-a.mkv`, size: GiB }, + { path: `${D}/weird-a.chs.srt`, size: 1000 }, + { path: `${D}/Show.S01E05.mkv`, size: GiB }, + { path: `${D}/sub/Show.S01E07.mkv`, size: GiB }, + ], + entity: { title: "剧", year: 2020 }, + target: { seasonDirs: { 1: "/lib/剧/Season 1", 2: "/lib/剧/Season 2" } }, + overrides: { + [`${D}/weird-a.mkv`]: { season: 2, episodes: { start: 3, end: 4 } }, + [`${D}/Show.S01E05.mkv`]: { episodes: { start: 6, end: 6 } }, + [`${D}/sub`]: { season: 2 }, + }, + }); + expect(plan.files.map((f) => f.dst)).toEqual([ + "/lib/剧/Season 2/剧 - S02E03-E04.mkv", + "/lib/剧/Season 1/剧 - S01E06.mkv", + "/lib/剧/Season 2/剧 - S02E07.mkv", + "/lib/剧/Season 2/剧 - S02E03-E04.chi.zh-cn.srt", + ]); + expect(plan.files[0]).toMatchObject({ season: 2, episodes: { start: 3, end: 4 } }); + const movie = planOrganize({ + operation: "copy", + mediaType: "movie", + files: [{ path: "/dl/M.2020.mkv", size: GiB }], + entity: { title: "M", year: 2020 }, + target: { movieDir: "/lib/M (2020)" }, + overrides: { "/dl/M.2020.mkv": { season: 3 } }, + }); + expect(movie.files[0]!.dst).toBe("/lib/M (2020)/M (2020).mkv"); + }); + it("findOtherVersions:同季同集同 Part、只看视频扩展名", () => { const sib = [ "/lib/S01/Show - S01E01 - 1080p.mkv", diff --git a/packages/mapping/src/organize.ts b/packages/mapping/src/organize.ts index 2e3c0125..5ddcb2b7 100644 --- a/packages/mapping/src/organize.ts +++ b/packages/mapping/src/organize.ts @@ -29,6 +29,10 @@ const ACTION_BACK: Record = { // domain 不生成目录调整/硬链接计划 relocate: null, hardlink: null, + // 文件管理的删除/还原记录不对应整理计划 + trash: null, + delete: null, + restore: null, }; export const DOMAIN_ORGANIZE_OPERATIONS = Object.keys(OPERATION) as OrganizeOperation[];