diff --git a/apps/server/src/modules/acquisition-organize/services/organize-finalize.ts b/apps/server/src/modules/acquisition-organize/services/organize-finalize.ts index e4458e5e..32d6c243 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize-finalize.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize-finalize.ts @@ -497,6 +497,30 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise fileKindOf(f.path) === "video").sort((a, b) => b.size - a.size)[0]?.path ?? + null; + const parsed = + takeover?.parsed ?? + (mainFile ? parseReleasePath(segmentsWithin(mainFile, sourceRoot), await wordsOption(db)) : null); const rid = await fail("identity_unknown", takeover ? { outcome: takeover.result.outcome } : {}); await decide_({ kind: "identity_ambiguous", @@ -429,6 +438,18 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise; entityId: string | null; sourceInstanceId: string | null; + request?: import("./organize-types.ts").StoredImportRequest; }, ): Promise { const now = d.ports.now(); @@ -1335,10 +1359,12 @@ export async function writeFailedRecord( media_entity_id: r.entityId, download_task_id: r.downloadTaskId, source_downloader_instance_id: r.sourceInstanceId, + source_path: r.request?.sourcePaths?.[0] ?? null, action: r.action, status: "failed", reason: r.reason, error: jsonb(r.evidence), + ...(r.request ? { plan: jsonb({ request: r.request, retryOfRecordId: r.request.retryOfRecordId }) } : {}), started_at: now, finished_at: now, }) diff --git a/apps/server/src/modules/acquisition-organize/services/organize-records.ts b/apps/server/src/modules/acquisition-organize/services/organize-records.ts index a707c7e2..2850426f 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize-records.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize-records.ts @@ -23,8 +23,9 @@ export function attentionSql(alias = "organization_record") { const r = sql.raw(alias); return sql`(not exists ( select 1 from organization_record r2 - where ${r}.download_task_id is not null and r2.download_task_id = ${r}.download_task_id + where (${r}.download_task_id is not null and r2.download_task_id = ${r}.download_task_id and r2.created_at > ${r}.created_at) + or (r2.plan->>'retryOfRecordId' = ${r}.id::text and r2.status in ('queued', 'running', 'succeeded', 'partial'))) and (exists ( select 1 from decision dd where dd.subject_type = 'organization_record' and dd.subject_id = ${r}.id @@ -71,8 +72,15 @@ async function loadAttention(exec: Exec, rows: OrganizationRecordRow[]) { eb .selectFrom("organization_record as r2") .select("r2.id") - .whereRef("r2.download_task_id", "=", "r.download_task_id") - .whereRef("r2.created_at", ">", "r.created_at"), + .where((eb) => + eb.or([ + eb.and([ + eb("r2.download_task_id", "=", eb.ref("r.download_task_id")), + eb("r2.created_at", ">", eb.ref("r.created_at")), + ]), + sql`r2.plan->>'retryOfRecordId' = r.id::text and r2.status in ('queued', 'running', 'succeeded', 'partial')`, + ]), + ), ), ) .execute(), @@ -102,7 +110,10 @@ async function loadAttention(exec: Exec, rows: OrganizationRecordRow[]) { evidence: ev, } : null, - agentIdentify: !!d && d.dedupe_key.startsWith("organize_identity:") && typeof ev.downloadTaskId === "string", + agentIdentify: + !!d && + d.dedupe_key.startsWith("organize_identity:") && + (typeof ev.downloadTaskId === "string" || Array.isArray(ev.sourcePaths)), }); } return out; diff --git a/apps/server/src/modules/acquisition-organize/services/organize-retry.ts b/apps/server/src/modules/acquisition-organize/services/organize-retry.ts new file mode 100644 index 00000000..82ae363c --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/services/organize-retry.ts @@ -0,0 +1,140 @@ +import { randomUUID } from "node:crypto"; +import { type OrganizationRecordRow, sql } from "@umt/db"; +import type { DecisionHookActor } from "../../decision-hooks.ts"; +import { type AoDeps, ensureTaskRow, markTask } from "./common.ts"; +import { sourceGuard } from "./organize-guard.ts"; +import { planImport } from "./organize-plan.ts"; +import type { ImportRequest, StoredImportRequest } from "./organize-types.ts"; + +/** 只保留业务参数；路径守卫与任务身份在恢复时重新装配。 */ +export function storedImportRequest(req: ImportRequest): StoredImportRequest { + const { + trigger: _trigger, + actorType: _actorType, + actorId: _actorId, + taskId: _taskId, + dryRun: _dryRun, + pathGuard: _pathGuard, + idempotencyKey: _idempotencyKey, + ...business + } = req; + return JSON.parse(JSON.stringify(business)); +} + +export function organizeRequestFromRecord( + record: Pick, +): StoredImportRequest | null { + const request = (record.plan as { request?: StoredImportRequest } | null)?.request; + if (request && (request.downloadTaskId || request.sourcePaths?.length)) return structuredClone(request); + // 老记录没有完整请求，不猜测一次性选项。 + return null; +} + +/** 按原请求重新规划；不复用旧冻结计划，来源路径重新经过当前目录边界。 */ +export async function retryOrganizeRecord( + d: AoDeps, + recordId: string, + actor: DecisionHookActor, + via: string, + overrides: { mediaEntityId?: string; idempotencyKey?: string; skipSuccessfulRecords?: boolean } = {}, +) { + const record = await d.ctx.db + .selectFrom("organization_record") + .selectAll() + .where("id", "=", recordId) + .executeTakeFirst(); + if (!record) throw new Error("整理记录不存在"); + if (overrides.idempotencyKey) { + const existing = await d.ctx.db + .selectFrom("organization_record") + .select(["id", "task_id", "status", "reason"]) + .where(sql`plan->>'idempotencyKey'`, "=", overrides.idempotencyKey) + .where("status", "in", ["queued", "running", "succeeded", "partial"]) + .executeTakeFirst(); + 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("整理记录没有可恢复的来源，重新选择来源后整理"); + if (overrides.skipSuccessfulRecords && request.sourcePaths?.length) { + const done = await d.ctx.db + .selectFrom("organization_file_result") + .select("source_path") + .where("organization_record_id", "=", recordId) + .where((eb) => + eb.or([ + eb("status", "=", "succeeded"), + eb.and([eb("status", "=", "skipped"), eb("skip_reason", "=", "already_in_library")]), + ]), + ) + .execute(); + const completed = new Set(done.map((f) => f.source_path)); + request.sourcePaths = request.sourcePaths.filter((p) => !completed.has(p)); + if (!request.sourcePaths.length) throw new Error("这条记录没有剩余待整理文件"); + } + let guarded: Pick = {}; + if (request.sourcePaths?.length) { + const guard = await sourceGuard(d.ctx.db); + guarded = { + sourcePaths: await Promise.all(request.sourcePaths.map((p) => guard.check(p))), + pathGuard: guard.check, + }; + } + return startManualImport( + d, + { + ...request, + ...guarded, + ...(overrides.mediaEntityId ? { mediaEntityId: overrides.mediaEntityId } : {}), + ...(overrides.idempotencyKey ? { idempotencyKey: overrides.idempotencyKey } : {}), + ...(overrides.skipSuccessfulRecords !== undefined + ? { skipSuccessfulRecords: overrides.skipSuccessfulRecords } + : {}), + retryOfRecordId: recordId, + }, + actor, + via, + ); +} + +export async function startManualImport( + d: AoDeps, + req: Omit, + actor: DecisionHookActor, + via: string, +): Promise<{ recordId: string | null; status: string; reason: string | null; taskId: string }> { + const db = d.ctx.db; + const idem = req.idempotencyKey ?? `${via}:${req.downloadTaskId ?? "x"}:${randomUUID()}`; + const by = actor.id ? `${actor.type}:${actor.id}` : actor.type; + const taskId = await ensureTaskRow(db, { + kind: "verifyAndImport", + idempotencyKey: idem, + input: { ...storedImportRequest({ ...req, trigger: "manual", actorType: "system" }), via }, + subjectType: req.downloadTaskId ? "download_task" : "organize_request", + subjectId: req.downloadTaskId ?? null, + triggeredBy: by, + }); + const actorType = actor.type === "api_key" || actor.type === "agent" || actor.type === "system" ? actor.type : "user"; + const plan = await planImport(d, { + ...req, + trigger: "manual", + actorType, + actorId: by, + taskId, + }); + if (plan.status === "running" || plan.status === "queued") { + const { runId } = await d.ctx.engine.enqueue( + "verifyAndImport", + { recordId: plan.recordId, downloadTaskId: req.downloadTaskId, taskId }, + { idempotencyKey: `verifyAndImport:${idem}`, traceId: taskId }, + ); + await db.updateTable("task").set({ engine_run_id: runId }).where("id", "=", taskId).execute(); + } else if (plan.status === "succeeded" || plan.status === "partial") { + await markTask(db, taskId, "succeeded", { + output: { recordId: plan.recordId, status: plan.status, reason: plan.reason }, + }); + } else { + await markTask(db, taskId, "failed", { error: plan.reason ?? plan.status }); + } + return { recordId: plan.recordId, status: plan.status, reason: plan.reason, taskId }; +} diff --git a/apps/server/src/modules/acquisition-organize/services/organize-types.ts b/apps/server/src/modules/acquisition-organize/services/organize-types.ts index 3ce6be4a..94a4906b 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize-types.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize-types.ts @@ -8,6 +8,8 @@ import type { FrozenReplacement } from "./replacement.ts"; // 资产/位置/路径绑定同事务登记 → Slot 推进 → Move 后移除 qB 任务记录(不删数据)。 export interface ImportRequest { + /** 从这条记录补齐事实后重新规划，用于原记录与后继结果关联。 */ + retryOfRecordId?: string; skipSuccessfulRecords?: boolean | undefined; scrapeAfterImport?: boolean | undefined; importMode?: "normal" | "replace"; @@ -57,6 +59,11 @@ export interface ImportRequest { episodeStructure?: OrganizeEpisodeStructure | undefined; } +export type StoredImportRequest = Omit< + ImportRequest, + "trigger" | "actorType" | "actorId" | "taskId" | "dryRun" | "pathGuard" | "idempotencyKey" +>; + /** 整理时指定的季集结构:groupId null = TMDB 默认季 */ export interface OrganizeEpisodeStructure { groupId: string | null; @@ -100,6 +107,8 @@ export type EpisodeFacts = Pick< /** organization_record.plan 的冻结内容 */ export interface FrozenPlan { + request?: StoredImportRequest; + retryOfRecordId?: string; scrapeAfterImport?: boolean; skippedSuccessful?: OrganizePlan["files"]; replacement?: FrozenReplacement; diff --git a/apps/server/src/modules/acquisition-organize/services/takeover.ts b/apps/server/src/modules/acquisition-organize/services/takeover.ts index 04f72bef..feb4d107 100644 --- a/apps/server/src/modules/acquisition-organize/services/takeover.ts +++ b/apps/server/src/modules/acquisition-organize/services/takeover.ts @@ -1,4 +1,3 @@ -import { randomUUID } from "node:crypto"; import { basename, dirname, extname, join } from "node:path"; import type { DownloadTaskRow } from "@umt/db"; import { @@ -16,8 +15,8 @@ import { addAliasIfAbsent } from "../../sources-identity/service/entities.ts"; import { wordsOption } from "../../sources-identity/service/identify-words-cache.ts"; import type { NameIdentifyResult } from "../ports.ts"; import { jsonb } from "../util.ts"; -import { type AoDeps, type Exec, ensureTaskRow, hostPathOf, loadInstances, markTask } from "./common.ts"; -import { type ImportRequest, planImport } from "./organize.ts"; +import { type AoDeps, type Exec, hostPathOf, loadInstances } from "./common.ts"; +import { retryOrganizeRecord, startManualImport } from "./organize-retry.ts"; // QB-1 接管任务(用户手动加进 qB、贴了管理标签,没有站点资源 / 计划)完成后的作品识别与事项闭环。 // 识别对照 MP transfer:没有下载历史时按主视频文件名(合并上级 / 上上级目录名,即 MetaInfoPath)+ 种子名识别, @@ -214,43 +213,7 @@ export async function keepBothTargets(plan: OrganizePlan): Promise { // ── 手动整理(待处理里选了处理方式):冻结计划 + 入队执行 ── -export async function startManualImport( - d: AoDeps, - req: Omit, - actor: DecisionHookActor, - via: string, -): Promise<{ recordId: string | null; status: string; reason: string | null; taskId: string }> { - const db = d.ctx.db; - const idem = `${via}:${req.downloadTaskId ?? "x"}:${randomUUID()}`; - const by = actor.id ? `${actor.type}:${actor.id}` : actor.type; - const taskId = await ensureTaskRow(db, { - kind: "verifyAndImport", - idempotencyKey: idem, - input: { ...req, via }, - subjectType: req.downloadTaskId ? "download_task" : "organize_request", - subjectId: req.downloadTaskId ?? null, - triggeredBy: by, - }); - const actorType = actor.type === "api_key" || actor.type === "agent" || actor.type === "system" ? actor.type : "user"; - const plan = await planImport(d, { - ...req, - trigger: "manual", - actorType, - actorId: by, - taskId, - }); - if (plan.status === "running" || plan.status === "queued") { - const { runId } = await d.ctx.engine.enqueue( - "verifyAndImport", - { recordId: plan.recordId, downloadTaskId: req.downloadTaskId, taskId }, - { idempotencyKey: `verifyAndImport:${idem}`, traceId: taskId }, - ); - await db.updateTable("task").set({ engine_run_id: runId }).where("id", "=", taskId).execute(); - } else { - await markTask(db, taskId, "failed", { error: plan.reason ?? plan.status }); - } - return { recordId: plan.recordId, status: plan.status, reason: plan.reason, taskId }; -} +export { startManualImport } from "./organize-retry.ts"; // ── 待处理执行钩子:organize_identity / organize_blocked ── @@ -330,9 +293,8 @@ export function organizeDecisionHook(d: AoDeps): DecisionHook { ? await d.ctx.db.selectFrom("download_task").selectAll().where("id", "=", dtId).executeTakeFirst() : undefined; if (row.dedupe_key.startsWith("organize_identity:")) { - if (!dt) throw new HookError("not_found", "关联的下载任务不存在"); const entityId = await chosenEntity(d, body, opt, actor); - return reorganizeAs(d, dt.id, entityId, actor, "decision"); + return reorganizeAs(d, row, entityId, actor, "decision"); } // organize_blocked:目标已存在 → 替换 / 保留两者 / 跳过 const key = body.optionKey; @@ -367,20 +329,23 @@ export function organizeDecisionHook(d: AoDeps): DecisionHook { }; } -/** 回填下载任务的作品后按该作品重新整理(人工指定与智能体识别共用) */ +/** 身份补齐后沿用失败整理请求，不重置一次性选项。 */ async function reorganizeAs( d: AoDeps, - downloadTaskId: string, + row: import("@umt/db").DecisionRow, entityId: string, actor: DecisionHookActor, via: string, ) { - await d.ctx.db - .updateTable("download_task") - .set({ media_entity_id: entityId }) - .where("id", "=", downloadTaskId) - .execute(); - const r = await startManualImport(d, { downloadTaskId, mediaEntityId: entityId }, actor, via); + const ev = (row.evidence ?? {}) as { downloadTaskId?: string | null }; + const recordId = row.subject_type === "organization_record" ? row.subject_id : null; + if (!recordId) throw new HookError("not_found", "这条事项没有关联的整理记录"); + const dtId = ev.downloadTaskId; + const r = await retryOrganizeRecord(d, recordId, actor, via, { mediaEntityId: entityId }); + if (!["running", "queued", "succeeded", "partial"].includes(r.status)) + throw new HookError("conflict", `作品已确认，但重新整理没能执行：${r.reason ?? r.status}`); + if (dtId) + await d.ctx.db.updateTable("download_task").set({ media_entity_id: entityId }).where("id", "=", dtId).execute(); return { kind: "organize", mediaEntityId: entityId, ...r }; } @@ -388,7 +353,7 @@ const OPEN_DECISION = ["open", "investigating", "awaiting_user"]; /** * organize.assign_identity:智能体给「没识别出作品」的整理事项指定作品(本地作品或 TMDB),随即重新整理并关闭事项。 - * 与人工在整理记录里「指定作品」走同一条路(回填下载任务 → 冻结手动整理计划 → 入队执行)。 + * 与人工指定作品共用原记录恢复入口；重新规划失败保留事项，排队成功后解决的是身份问题，入库结果另查整理记录。 */ export async function assignOrganizeIdentity( d: AoDeps, @@ -403,9 +368,6 @@ export async function assignOrganizeIdentity( if (!row.dedupe_key.startsWith("organize_identity:")) throw new HookError("bad_request", "只能用于整理时「没识别出作品」的事项"); if (!OPEN_DECISION.includes(row.status)) throw new HookError("conflict", "这条记录已经处理过了"); - const ev = (row.evidence ?? {}) as { downloadTaskId?: string | null }; - const dtId = ev.downloadTaskId ?? null; - if (!dtId) throw new HookError("not_found", "这条记录没有关联的下载任务，只能在整理向导里手动整理"); const entityId = pick.mediaEntityId ? await chosenEntity( d, @@ -419,7 +381,7 @@ export async function assignOrganizeIdentity( { key: "x", label: "x", input: { requires: "tmdb" } }, actor, ); - const r = await reorganizeAs(d, dtId, entityId, actor, "agent-identify"); + const r = await reorganizeAs(d, row, entityId, actor, "agent-identify"); const by = actor.id ? `${actor.type}:${actor.id}` : actor.type; await db.transaction().execute(async (trx) => { const closed = await trx diff --git a/apps/server/src/modules/agent/investigate.ts b/apps/server/src/modules/agent/investigate.ts index 3a04e9ff..25e0ebd1 100644 --- a/apps/server/src/modules/agent/investigate.ts +++ b/apps/server/src/modules/agent/investigate.ts @@ -191,9 +191,14 @@ export async function autoInvestigate( // Agent 自己引起的事项不再触发 Agent(防止递归自触发) if (d.subject_type === "agent_run") return { runId: null, created: false }; const s = await ctx.settings.get("agent"); - if (!s.autoInvestigate) return { runId: null, created: false }; + const organizeEpisodes = + d.kind === "episode_mapping" && + Array.isArray((d.evidence as { fileSamples?: unknown })?.fileSamples) && + (d.evidence as { fileSamples: unknown[] }).fileSamples.length > 0; + if (organizeEpisodes ? !s.autoIdentifyOrganize : !s.autoInvestigate) return { runId: null, created: false }; - const requested = s.autoInvestigateMode; + // 显式整理的文件季集异常沿用整理识别开关；写规则仍要经过原有持续授权。 + const requested = organizeEpisodes ? "recover" : s.autoInvestigateMode; let mode: "investigate" | "recover" = requested; let degraded: { reason: string } | undefined; let attempt = 0; @@ -328,6 +333,10 @@ function evidenceLine(ev: Record): string { const parts = [ typeof ev.torrentName === "string" ? `种子名「${ev.torrentName}」` : "", typeof ev.mainFile === "string" ? `主视频「${ev.mainFile}」` : "", + typeof ev.sourceReleaseId === "string" + ? `来源资源 ${ev.sourceReleaseId}（可用 release.inspect 查看下载来源详情）` + : "", + Array.isArray(ev.sourcePaths) ? `整理来源 ${ev.sourcePaths.slice(0, 5).join("、")}` : "", parsed.title ? `解析出的标题「${String(parsed.title)}」` : "", parsed.year ? `年份 ${String(parsed.year)}` : "", parsed.season != null ? `第 ${String(parsed.season)} 季` : "", @@ -372,7 +381,7 @@ export async function identifyOrganize( evidenceLine(ev), "用 catalog.lookup 按标题 / 年份查 TMDB，用 inventory.query 看媒体库里有没有这个作品；标题是拼音、缩写或译名对不上时可以用 web.search 查正式名称。", `确认标题、年份、类型（电影 / 剧集）和季集都对得上时，调用一次 organize.assign_identity（decisionId 填 ${d.id}，` + - "本地已有作品填 mediaEntityId，否则填 tmdb 类型和 ID，reasons 写依据），系统会按这个作品重新整理。", + "本地已有作品填 mediaEntityId，否则填 tmdb 类型和 ID，reasons 写依据），系统会保留这次整理的选项按这个作品重新整理。指定作品成功只表示已补齐身份并重新排队，不能据此声称已经入库。", "拿不准就不要指定，只说明查到了什么、为什么认不出。不下载、不删除、不改别的记录。", ].join(""); const { run, created } = await startRun(ctx, { @@ -502,13 +511,34 @@ export async function checkRecovery(ctx: ServerContext, actions: RecoveryAction[ : "pending", ); } else if (a.capabilityId === "organize.assign_identity") { - // 指定作品后事项随即关闭、重新整理已入队;事项关闭即视为恢复(整理结果看整理记录) + // 身份已补齐与入库成功分开核实，以重新整理记录的业务终态为准。 const id = String(a.input.decisionId ?? ""); const d = id - ? await ctx.db.selectFrom("decision").select(["status"]).where("id", "=", id).executeTakeFirst() + ? await ctx.db.selectFrom("decision").select(["status", "resolution"]).where("id", "=", id).executeTakeFirst() : undefined; - facts[`decision:${id}`] = d ? { status: d.status } : null; - states.push(!d ? "failed" : d.status === "resolved" ? "recovered" : "pending"); + const recordId = (d?.resolution as { execution?: { recordId?: string } } | null)?.execution?.recordId; + const record = recordId + ? await ctx.db + .selectFrom("organization_record") + .select(["status", "reason"]) + .where("id", "=", recordId) + .executeTakeFirst() + : null; + facts[`decision:${id}`] = d + ? { + status: d.status, + recordId: recordId ?? null, + organizeStatus: record?.status ?? null, + reason: record?.reason ?? null, + } + : null; + states.push( + !d || !record || record.status === "failed" + ? "failed" + : record.status === "succeeded" + ? "recovered" + : "pending", + ); } else if (a.capabilityId === "episode_mapping.apply_rule") { // 事项内资源都已不是待确认(写进作品识别词后的重算会把事项自动关闭)即视为恢复 const id = String(a.input.decisionId ?? ""); @@ -563,8 +593,13 @@ export async function applyRecoveryOutcome( else if (res.reason === "budget") outcome = "budget_exceeded"; else outcome = "not_attempted"; const what = (done.length > 0 ? done : acts.actions).map((a) => a.capabilityId).join(" + "); - const summary = - outcome === "recovered" + const organizePending = + outcome === "unverified" && + done.some((a) => a.capabilityId === "organize.assign_identity") && + check?.state === "pending"; + const summary = organizePending + ? "作品已确认，重新整理已提交；入库结果以整理记录为准。" + : outcome === "recovered" ? `已恢复（系统核实）：执行 ${what} 后业务状态已恢复。` : outcome === "failed" ? done.length > 0 @@ -584,7 +619,7 @@ export async function applyRecoveryOutcome( denied: acts.denied, facts: check?.facts ?? {}, verifiedAt: new Date().toISOString(), - ...(outcome === "recovered" ? {} : { suggestion: suggestionFor(d, summary.replace(/。$/, "")) }), + ...(outcome === "recovered" || organizePending ? {} : { suggestion: suggestionFor(d, summary.replace(/。$/, "")) }), }; await mergeEvidence(ctx, decisionId, evidence); if (outcome === "recovered") { diff --git a/apps/server/test/acquisition-organize/organize-identity-recovery.test.ts b/apps/server/test/acquisition-organize/organize-identity-recovery.test.ts new file mode 100644 index 00000000..b1b20bbf --- /dev/null +++ b/apps/server/test/acquisition-organize/organize-identity-recovery.test.ts @@ -0,0 +1,206 @@ +import { symlink, unlink, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { createTempTree, type TempTree } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { executeRecord, planImport } from "../../src/modules/acquisition-organize/services/organize.ts"; +import { + retryOrganizeRecord, + storedImportRequest, +} from "../../src/modules/acquisition-organize/services/organize-retry.ts"; +import { + assignOrganizeIdentity, + organizeDecisionHook, +} from "../../src/modules/acquisition-organize/services/takeover.ts"; +import { autoInvestigate, checkRecovery } from "../../src/modules/agent/investigate.ts"; +import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "./harness.ts"; + +const hasPg = await pgAvailable(); +const actor = { type: "user", id: "identity-recovery" }; +it("请求快照保留业务参数，不序列化运行身份与路径守卫", () => { + expect( + storedImportRequest({ + sourcePaths: ["/cache/movie.mkv"], + namingMode: "keep", + action: "move", + trigger: "manual", + actorType: "user", + actorId: "secret", + taskId: "old", + idempotencyKey: "old", + dryRun: true, + pathGuard: async (p) => p, + }), + ).toEqual({ sourcePaths: ["/cache/movie.mkv"], namingMode: "keep", action: "move" }); +}); +describe.skipIf(!hasPg)("按路径身份补齐后恢复原整理", () => { + let h: Harness; + let lib: TempTree; + let cache: TempTree; + let outside: TempTree; + beforeAll(async () => { + lib = await createTempTree({}, { sentinel: true }); + cache = await createTempTree({}); + outside = await createTempTree({}); + h = await createHarness("identity_recovery", {}, { withAgent: true }); + await seedBase(h.db, { libraryPath: lib.root, cachePath: cache.root, qbUrl: "http://127.0.0.1:1" }); + }, 60_000); + afterAll(async () => { + await h?.close(); + await lib?.cleanup(); + await cache?.cleanup(); + await outside?.cleanup(); + }); + async function unknown(name: string, extra = {}) { + const src = join(cache.root, `${name}.2022.1080p.mkv`); + await writeFile(src, Buffer.alloc(300_000, 1)); + const plan = await planImport(h.d, { + sourcePaths: [src], + trigger: "manual", + actorType: "user", + actorId: actor.id, + ...extra, + }); + expect(plan).toMatchObject({ status: "failed", reason: "identity_unknown" }); + const dec = await h.db + .selectFrom("decision") + .selectAll() + .where("subject_id", "=", plan.recordId!) + .executeTakeFirstOrThrow(); + return { src, plan, dec }; + } + it("路径身份事项有文件证据和Agent入口，指定作品保留目录、命名与刮削参数", async () => { + const targetDir = join(lib.root, "custom"); + const { src, plan, dec } = await unknown("Recovery.Path", { + action: "copy", + targetDir, + namingMode: "keep", + scrapeAfterImport: false, + minSizeBytes: 100, + }); + expect(dec.evidence).toMatchObject({ + sourcePaths: [src], + mainFile: src, + parsed: { title: "Recovery Path", year: 2022 }, + }); + expect((await h.api("GET", `/organization-records/${plan.recordId}`)).body.data.attention.agentIdentify).toBe(true); + const entityId = await seedEntity(h.db, { kind: "movie", title: "Recovery Path", year: 2022 }); + const assigned = await assignOrganizeIdentity(h.d, dec.id, { mediaEntityId: entityId }, actor, "标题和年份一致"); + expect(assigned.status).toBe("running"); + const record = await h.db + .selectFrom("organization_record") + .selectAll() + .where("id", "=", assigned.recordId!) + .executeTakeFirstOrThrow(); + expect(record.plan).toMatchObject({ + retryOfRecordId: plan.recordId, + request: { sourcePaths: [src], targetDir, namingMode: "keep", scrapeAfterImport: false, minSizeBytes: 100 }, + naming: { mode: "keep" }, + }); + const action = [{ capabilityId: "organize.assign_identity", input: { decisionId: dec.id }, status: "succeeded" }]; + expect((await checkRecovery(h.ctx, action)).state).toBe("pending"); + expect((await executeRecord(h.d, assigned.recordId!)).status).toBe("succeeded"); + expect((await checkRecovery(h.ctx, action)).state).toBe("recovered"); + const attention = await h.api("GET", "/organization-records?attention=true&organizeOnly=true"); + expect(attention.body.data.items.map((r: { id: string }) => r.id)).not.toContain(plan.recordId); + }); + it("人工指定恢复原路径请求，不要求关联下载任务", async () => { + const { plan, dec } = await unknown("Human.Path", { namingMode: "keep" }); + const entityId = await seedEntity(h.db, { kind: "movie", title: "Human Path", year: 2022 }); + const result = await organizeDecisionHook(h.d).execute( + dec, + { optionKey: "select_entity", identity: { mediaEntityId: entityId } }, + actor, + ); + expect(result).toMatchObject({ status: "running", mediaEntityId: entityId }); + expect( + ( + await h.db + .selectFrom("organization_record") + .select("plan") + .where("id", "=", result.recordId as string) + .executeTakeFirstOrThrow() + ).plan, + ).toMatchObject({ retryOfRecordId: plan.recordId }); + }); + it("身份已找到但重新规划失败，保持原事项待处理", async () => { + const { dec } = await unknown("Blocked.Path", { targetLibraryRootId: "missing-root" }); + 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( + /重新整理没能执行/, + ); + expect( + (await h.db.selectFrom("decision").select("status").where("id", "=", dec.id).executeTakeFirstOrThrow()).status, + ).toBe("open"); + expect( + ( + await checkRecovery(h.ctx, [ + { capabilityId: "organize.assign_identity", input: { decisionId: dec.id }, status: "succeeded" }, + ]) + ).state, + ).toBe("failed"); + }); + it("恢复重新检查路径边界，拒绝后来被换成越界符号链接的源文件", async () => { + const { src, dec } = await unknown("Guarded.Path"); + const entityId = await seedEntity(h.db, { kind: "movie", title: "Guarded Path", year: 2022 }); + const escaped = join(outside.root, "escaped.mkv"); + await writeFile(escaped, Buffer.alloc(300_000)); + await unlink(src); + await symlink(escaped, src); + await expect(assignOrganizeIdentity(h.d, dec.id, { mediaEntityId: entityId }, actor, "标题一致")).rejects.toThrow( + /不在已登记/, + ); + expect( + (await h.db.selectFrom("decision").select("status").where("id", "=", dec.id).executeTakeFirstOrThrow()).status, + ).toBe("open"); + }); + it("Move 恢复完成后稳定幂等键重放沿用结果，不再读取已移动的源文件", async () => { + const { plan } = await unknown("Replay.Path", { action: "move", namingMode: "keep" }); + const entityId = await seedEntity(h.db, { kind: "movie", title: "Replay Path", year: 2022 }); + const options = { mediaEntityId: entityId, idempotencyKey: "recovery:replay-path" }; + const first = await retryOrganizeRecord(h.d, plan.recordId!, actor, "test-replay", options); + expect((await executeRecord(h.d, first.recordId!)).status).toBe("succeeded"); + const replay = await retryOrganizeRecord(h.d, plan.recordId!, actor, "test-replay", options); + expect(replay).toMatchObject({ recordId: first.recordId, taskId: first.taskId, status: "succeeded" }); + }); + 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(/没有可恢复/); + }); + it("整理文件季集事项默认调查，无写授权降级建议，资源站普通事项不开启全局恢复", async () => { + await h.setSettings("agent", { autoInvestigate: false, autoIdentifyOrganize: true }); + const file = await h.db + .insertInto("decision") + .values({ + kind: "episode_mapping", + title: "整理文件季集待确认", + dedupe_key: "episode:test-files", + evidence: JSON.stringify({ fileSamples: [{ recordId: "sample" }] }), + }) + .returning("id") + .executeTakeFirstOrThrow(); + const out = await autoInvestigate(h.ctx, file.id); + expect(out).toMatchObject({ created: true, mode: "investigate" }); + const run = await h.db.selectFrom("agent_run").selectAll().where("id", "=", out.runId!).executeTakeFirstOrThrow(); + expect(run.mode).toBe("plan_only"); + expect((await autoInvestigate(h.ctx, file.id)).created).toBe(false); + const ordinary = await h.db + .insertInto("decision") + .values({ kind: "episode_mapping", title: "资源站季集待确认", dedupe_key: "episode:test-ordinary" }) + .returning("id") + .executeTakeFirstOrThrow(); + expect((await autoInvestigate(h.ctx, ordinary.id)).created).toBe(false); + await h.setSettings("agent", { autoIdentifyOrganize: false }); + const disabled = await h.db + .insertInto("decision") + .values({ + kind: "episode_mapping", + title: "整理关闭", + dedupe_key: "episode:test-disabled", + evidence: JSON.stringify({ fileSamples: [{}] }), + }) + .returning("id") + .executeTakeFirstOrThrow(); + expect((await autoInvestigate(h.ctx, disabled.id)).created).toBe(false); + }); +});