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