diff --git a/apps/server/src/modules/acquisition-organize/handlers.ts b/apps/server/src/modules/acquisition-organize/handlers.ts index dd5b4705..772b813f 100644 --- a/apps/server/src/modules/acquisition-organize/handlers.ts +++ b/apps/server/src/modules/acquisition-organize/handlers.ts @@ -4,6 +4,7 @@ import { evaluateAcquisition } from "./services/acquisition.ts"; import { type AoDeps, automationBlock, ensureTaskRow, markTask } from "./services/common.ts"; import { cleanupDownloadCache, submitDownload } from "./services/downloads.ts"; import { executeRecord, planImport } from "./services/organize.ts"; +import { resumeEpisodeOrganize } from "./services/organize-episode-recovery.ts"; import { reconcileLibrary } from "./services/reconcile.ts"; import { runScrapeRecord, SCRAPE_TASK } from "./services/scrape.ts"; import { errMessage } from "./util.ts"; @@ -115,9 +116,31 @@ export function registerAoTasks(registry: TaskRegistry, d: AoDeps) { name: "verifyAndImport", version: 1, lane: "file", - inputSchema: z.object({ downloadTaskId: id.optional(), recordId: id.optional(), taskId: id.optional() }), + inputSchema: z.object({ + downloadTaskId: id.optional(), + recordId: id.optional(), + taskId: id.optional(), + mediaEntityId: id.optional(), + }), retryPolicy: FILE_RETRY, handler: async (t) => { + if (t.input.mediaEntityId && !t.input.recordId && !t.input.downloadTaskId) { + const taskId = await taskRow( + d, + "verifyAndImport", + t.input.taskId, + `episodes:${t.input.mediaEntityId}:${t.runId}`, + t.input, + t.runId, + { type: "media_entity", id: t.input.mediaEntityId }, + ); + return tracked( + d, + taskId, + () => t.step.run("resume-episodes", () => resumeEpisodeOrganize(d, t.input.mediaEntityId!)), + (r) => (r.waiting ? "waiting" : "succeeded"), + ); + } const key = t.input.recordId ? `record:${t.input.recordId}` : `import:${t.input.downloadTaskId ?? t.runId}`; const taskId = await taskRow( d, diff --git a/apps/server/src/modules/acquisition-organize/index.ts b/apps/server/src/modules/acquisition-organize/index.ts index 1a6366ce..b6efa558 100644 --- a/apps/server/src/modules/acquisition-organize/index.ts +++ b/apps/server/src/modules/acquisition-organize/index.ts @@ -17,6 +17,7 @@ import { registerOrganizationRoutes } from "./routes/organization.ts"; import { registerRuleRoutes } from "./routes/rules.ts"; import { type AoDeps, automationBlock, enqueueSweep, enqueueTracked, timeSlot } from "./services/common.ts"; import { reportVolumeStates, SWEEP_SLOT_MS, sweepReservedPlans, syncDownloads } from "./services/downloads.ts"; +import { sweepEpisodeOrganize } from "./services/organize-episode-recovery.ts"; import { recoverStalledRecords } from "./services/organize-records.ts"; import { sweepDueSlots, sweepMetadataRefresh, sweepSubscriptionSearch } from "./services/subscriptions.ts"; import { organizeDecisionHook } from "./services/takeover.ts"; @@ -91,6 +92,10 @@ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): Se registerAoCapabilities(registry, () => lazy(ctx)); }, outboxRoutes: () => ({ + [EventType.mediaEpisodeMappingChanged]: (ev) => { + const p = ev.payload as EventPayload<"media.episode_mapping_changed">; + return { task: "verifyAndImport", input: { mediaEntityId: p.mediaEntityId } }; + }, [EventType.releaseResolved]: (ev) => { const p = ev.payload as EventPayload<"release.resolved">; return { @@ -167,10 +172,12 @@ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): Se const submitted = await sweepReservedPlans(d); const imports = await sweepPendingImports(d); const rec = await recoverStalledRecords(d); + const episodes = await sweepEpisodeOrganize(d); return ( `补提交 ${submitted}，补整理 ${imports}` + (rec.requeued ? `，重排整理 ${rec.requeued}` : "") + (rec.failed ? `，整理停止 ${rec.failed}` : "") + + (episodes ? `，季集恢复 ${episodes}` : "") + (rec.removed ? `，移除下载任务 ${rec.removed}` : "") ); }), diff --git a/apps/server/src/modules/acquisition-organize/services/organize-episode-recovery.ts b/apps/server/src/modules/acquisition-organize/services/organize-episode-recovery.ts new file mode 100644 index 00000000..7d9beda1 --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/services/organize-episode-recovery.ts @@ -0,0 +1,108 @@ +import { sql } from "@umt/db"; +import { parseReleasePath } from "@umt/domain"; +import { wordsOption } from "../../sources-identity/service/identify-words-cache.ts"; +import { errMessage, hashJson } from "../util.ts"; +import { type AoDeps, automationBlock, enqueueSweep, timeSlot } from "./common.ts"; +import { SWEEP_SLOT_MS } from "./downloads.ts"; +import { planFileEpisodes, segmentsWithin } from "./organize-plan.ts"; +import { organizeRequestFromRecord, retryOrganizeRecord } from "./organize-retry.ts"; +import type { FrozenPlan } from "./organize-types.ts"; + +/** 自动事实恢复只尝试仍未被后继记录接手的来源，失败后继留给原记录就地处理。 */ +function pendingEpisodeRecoverySql() { + return sql`r.status in ('failed', 'partial') and r.attention_cleared_at is null + and r.plan -> 'unresolvedEpisodes' is not null and r.plan -> 'request' is not null + and not exists (select 1 from organization_record next + where (r.download_task_id is not null and next.download_task_id = r.download_task_id and next.created_at > r.created_at) + or next.plan ->> 'retryOfRecordId' = r.id::text)`; +} + +/** 事实更新后只恢复季集待确认的整理；执行异常仍走原有文件账本恢复。 */ +export async function resumeEpisodeOrganize(d: AoDeps, entityId: string) { + const result = { resumed: 0, waiting: 0 }; + if (await automationBlock(d)) return result; + const entity = await d.ctx.db.selectFrom("media_entity").selectAll().where("id", "=", entityId).executeTakeFirst(); + if (!entity || (entity.kind !== "series" && entity.kind !== "variety")) return result; + const rows = await d.ctx.db + .selectFrom("organization_record as r") + .selectAll("r") + .where("r.media_entity_id", "=", entityId) + .where(pendingEpisodeRecoverySql()) + .orderBy("r.created_at") + .execute(); + const words = await wordsOption(d.ctx.db); + for (const record of rows) { + const plan = record.plan as unknown as FrozenPlan; + const request = organizeRequestFromRecord(record); + // 临时季集结构是本次专属约定，不按作品规则改变它；旧记录缺原请求时交给人工。 + if (!request || plan.episodeStructure?.temporary) continue; + const pending = Object.entries(plan.unresolvedEpisodes ?? {}).filter(([, r]) => r.method === "unresolved"); + if (!pending.length) continue; + const download = record.download_task_id + ? await d.ctx.db + .selectFrom("download_task") + .selectAll() + .where("id", "=", record.download_task_id) + .executeTakeFirst() + : undefined; + const release = download?.source_release_id + ? await d.ctx.db + .selectFrom("source_release") + .selectAll() + .where("id", "=", download.source_release_id) + .executeTakeFirst() + : undefined; + const episodes = await planFileEpisodes(d, entity, release, download, null); + const resolved = pending.map( + ([path]) => + episodes.map(parseReleasePath(segmentsWithin(path, plan.sourceRoot), words), episodes.opts).resolution, + ); + if (resolved.some((r) => !r.canonical || r.method === "unresolved" || r.method === "unverified" || r.special)) { + result.waiting++; + continue; + } + try { + const retry = await retryOrganizeRecord(d, record.id, { type: "system", id: null }, "episode-mapping", { + mediaEntityId: entityId, + skipSuccessfulRecords: true, + sourcePaths: pending.map(([path]) => path), + idempotencyKey: `organize-episodes:${record.id}:${hashJson(resolved)}`, + }); + if (retry.status === "running" || retry.status === "queued") result.resumed++; + else result.waiting++; + } catch (error) { + result.waiting++; + d.ctx.logger.warn({ recordId: record.id, error: errMessage(error) }, "季集映射已更新，整理仍需处理"); + } + } + return result; +} + +/** 复用补整理巡检：暂停期间已处理的映射在恢复自动化后仍有唤醒机会。 */ +export async function sweepEpisodeOrganize(d: AoDeps): Promise { + if (await automationBlock(d)) return 0; + const entities = await d.ctx.db + .selectFrom("organization_record as r") + .select("r.media_entity_id") + .distinct() + .where("r.media_entity_id", "is not", null) + .where(pendingEpisodeRecoverySql()) + .execute(); + let queued = 0; + for (const entity of entities) { + if (!entity.media_entity_id) continue; + const run = await enqueueSweep( + d, + { + kind: "verifyAndImport", + idempotencyKey: `episode-organize:${entity.media_entity_id}`, + input: { mediaEntityId: entity.media_entity_id }, + subjectType: "media_entity", + subjectId: entity.media_entity_id, + }, + timeSlot(d.ports.now(), SWEEP_SLOT_MS), + ); + if (run) queued++; + } + return queued; +} diff --git a/apps/server/src/modules/acquisition-organize/services/organize-plan.ts b/apps/server/src/modules/acquisition-organize/services/organize-plan.ts index a3d31dbd..62655797 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize-plan.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize-plan.ts @@ -41,6 +41,7 @@ import { type FileEpisodeOptions, publishedAtOf, settleEpisodeDecision, + settleOrganizeEpisodeDecisions, } from "../../sources-identity/service/episode-resolution.ts"; import type { Layout } from "../../sources-identity/service/episode-structure.ts"; import { wordsOption } from "../../sources-identity/service/identify-words-cache.ts"; @@ -279,7 +280,7 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise f.resolution.method === "unresolved") && !episodes.toEntity) { + const releaseResolution = episodes.release?.episode_resolution as EpisodeResolution | null; + if (episodes.release && releaseResolution?.method === "unresolved") { + await settleEpisodeDecision(d.ctx, db, { + sourceReleaseId: episodes.release.id, + entityId: entity.id, + resolution: releaseResolution, + parsed: parsedOf(episodes.release), + }); + } + await settleOrganizeEpisodeDecisions(d.ctx, db, { + recordId: result.recordId, entityId: entity.id, - resolution: pending.resolution, - parsed: pending.parsed, + files: episodes.samples.map((file) => ({ + ...file, + publishedAt: episodes.release?.source_published_at?.toISOString() ?? null, + })), }); } return result; @@ -1204,14 +1220,14 @@ interface FileEpisodes { toEntity?: (span: EpisodeSpan) => EpisodeSpan | null; /** 下载关联的资源(已绑定本作品) */ release: SourceReleaseRow | null; - unresolved: { path: string; resolution: EpisodeResolution; parsed: ParsedRelease }[]; + samples: { path: string; resolution: EpisodeResolution; parsed: ParsedRelease }[]; } /** * 整理计划的集数映射装配:作品季表 + 规则加载一次;下载关联的资源(下载任务直接关联,或经下载 / 导入计划) * 提供发布时间,原始季集与文件相同时直接沿用它的规范季集 */ -async function planFileEpisodes( +export async function planFileEpisodes( d: AoDeps, entity: MediaEntityRow, rel: SourceReleaseRow | undefined, @@ -1239,7 +1255,7 @@ async function planFileEpisodes( toEntity: s.toEntity, opts: linked ? { publishedAt: publishedAtOf(linked) } : {}, release: own, - unresolved: [], + samples: [], }; } const map = await createEntityEpisodeMapper(db, entity.id, entity.kind); @@ -1251,7 +1267,7 @@ async function planFileEpisodes( ...(res?.method ? { release: res } : {}), }, release: own, - unresolved: [], + samples: [], }; } diff --git a/apps/server/src/modules/acquisition-organize/services/organize-retry.ts b/apps/server/src/modules/acquisition-organize/services/organize-retry.ts index 82ae363c..9a00448a 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize-retry.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize-retry.ts @@ -30,13 +30,39 @@ export function organizeRequestFromRecord( return null; } -/** 按原请求重新规划；不复用旧冻结计划，来源路径重新经过当前目录边界。 */ +interface RetryOptions { + mediaEntityId?: string; + idempotencyKey?: string; + skipSuccessfulRecords?: boolean; + sourcePaths?: string[]; +} + +/** 同一原记录的重新规划串行；事件、巡检和人工同时到达也不能重复冻结文件计划。 */ export async function retryOrganizeRecord( d: AoDeps, recordId: string, actor: DecisionHookActor, via: string, - overrides: { mediaEntityId?: string; idempotencyKey?: string; skipSuccessfulRecords?: boolean } = {}, + overrides: RetryOptions = {}, +) { + return d.ctx.db.connection().execute(async (connection) => { + const key = `organize-retry:${recordId}`; + await sql`select pg_advisory_lock(hashtextextended(${key}, 0))`.execute(connection); + try { + return await retryLocked({ ...d, ctx: { ...d.ctx, db: connection } }, recordId, actor, via, overrides); + } finally { + await sql`select pg_advisory_unlock(hashtextextended(${key}, 0))`.execute(connection); + } + }); +} + +/** 按原请求重新规划；不复用旧冻结计划，来源路径重新经过当前目录边界。 */ +async function retryLocked( + d: AoDeps, + recordId: string, + actor: DecisionHookActor, + via: string, + overrides: RetryOptions, ) { const record = await d.ctx.db .selectFrom("organization_record") @@ -56,6 +82,17 @@ export async function retryOrganizeRecord( } const request = organizeRequestFromRecord(record); if (!request) throw new Error("整理记录没有可恢复的来源，重新选择来源后整理"); + if (overrides.sourcePaths) { + const files = await d.ctx.db + .selectFrom("organization_file_result") + .select("source_path") + .where("organization_record_id", "=", recordId) + .execute(); + const original = new Set(files.map((f) => f.source_path)); + if (!overrides.sourcePaths.length || overrides.sourcePaths.some((p) => !original.has(p))) + throw new Error("恢复只能选择原整理记录中的文件"); + request.sourcePaths = [...new Set(overrides.sourcePaths)]; + } if (overrides.skipSuccessfulRecords && request.sourcePaths?.length) { const done = await d.ctx.db .selectFrom("organization_file_result") diff --git a/apps/server/src/modules/agent/investigate.ts b/apps/server/src/modules/agent/investigate.ts index 25e0ebd1..277ae7ce 100644 --- a/apps/server/src/modules/agent/investigate.ts +++ b/apps/server/src/modules/agent/investigate.ts @@ -540,7 +540,7 @@ export async function checkRecovery(ctx: ServerContext, actions: RecoveryAction[ : "pending", ); } else if (a.capabilityId === "episode_mapping.apply_rule") { - // 事项内资源都已不是待确认(写进作品识别词后的重算会把事项自动关闭)即视为恢复 + // 资源与真实整理文件都须解决；仅资源列表为空不能代表文件映射已恢复。 const id = String(a.input.decisionId ?? ""); const d = id ? await ctx.db.selectFrom("decision").select(["status", "evidence"]).where("id", "=", id).executeTakeFirst() @@ -554,8 +554,18 @@ export async function checkRecovery(ctx: ServerContext, actions: RecoveryAction[ .execute() : []; const unresolved = rows.filter((r) => r.method === "unresolved").length; - facts[`decision:${id}`] = d ? { status: d.status, releases: ids.length, unresolved } : null; - states.push(!d ? "failed" : d.status === "resolved" || unresolved === 0 ? "recovered" : "pending"); + const fileSamples = + ((d?.evidence ?? {}) as { fileSamples?: { resolution?: { method?: string } }[] }).fileSamples ?? []; + const unresolvedFiles = fileSamples.filter((f) => f.resolution?.method === "unresolved").length; + facts[`decision:${id}`] = d ? { status: d.status, releases: ids.length, unresolved, unresolvedFiles } : null; + states.push( + !d + ? "failed" + : unresolvedFiles === 0 && + (d.status === "resolved" || (ids.length > 0 && rows.length === ids.length && unresolved === 0)) + ? "recovered" + : "pending", + ); } } const state = states.includes("failed") ? "failed" : states.every((x) => x === "recovered") ? "recovered" : "pending"; diff --git a/apps/server/src/modules/sources-identity/service/episode-mapping-admin.ts b/apps/server/src/modules/sources-identity/service/episode-mapping-admin.ts index af6cbd15..f6d41266 100644 --- a/apps/server/src/modules/sources-identity/service/episode-mapping-admin.ts +++ b/apps/server/src/modules/sources-identity/service/episode-mapping-admin.ts @@ -302,6 +302,8 @@ export async function simulateEpisodeRule( } if (regressions.length) failures.push(`会让 ${regressions.length} 条已对上的资源变成待确认（如「${regressions[0]!.title}」）`); + if (fileRegressions.length) + failures.push(`会让 ${fileRegressions.length} 个已对上的文件变成待确认（如「${fileRegressions[0]!.file.path}」）`); const summary: EpisodeSimulationSummary = { samples: ids.size + files.length, resolved: resolved.length + resolvedFiles.length, diff --git a/apps/server/src/modules/sources-identity/service/episode-resolution.ts b/apps/server/src/modules/sources-identity/service/episode-resolution.ts index f0cd7403..2a354e3c 100644 --- a/apps/server/src/modules/sources-identity/service/episode-resolution.ts +++ b/apps/server/src/modules/sources-identity/service/episode-resolution.ts @@ -460,7 +460,16 @@ export async function settleOrganizeEpisodeDecisions( const previous = existing.find((d) => d.dedupe_key === key); const ev = previous?.evidence as EpisodeDecisionEvidence | undefined; const group = groups.get(key); - const files = [...episodeFileSamples(ev).filter((f) => f.recordId !== input.recordId), ...(group?.files ?? [])]; + // 同批已确定的文件作为模拟护栏，不能用修正坏文件的规则破坏它们。 + // 其它范围的未决文件由其自己的事项处理，不要求一条规则解决整批问题。 + const context = group + ? input.files.filter((f) => f.resolution.method !== "unresolved").map((f) => ({ ...f, recordId: input.recordId })) + : []; + const files = [ + ...episodeFileSamples(ev).filter((f) => f.recordId !== input.recordId), + ...(group?.files ?? []), + ...context, + ]; const ids = ev?.sourceReleaseIds ?? []; if (!ids.length && !files.some((f) => f.resolution.method === "unresolved")) { await closeDecisions( @@ -826,6 +835,7 @@ export async function reresolveEntityEpisodes( } } await reresolveFileSamples(ctx, exec, entityId, entity.kind, table, rules); + await ctx.outbox.emit(exec, "media.episode_mapping_changed", { mediaEntityId: entityId }); return out; } diff --git a/apps/server/test/acquisition-organize/organize-episode-recovery.test.ts b/apps/server/test/acquisition-organize/organize-episode-recovery.test.ts new file mode 100644 index 00000000..ede67bf1 --- /dev/null +++ b/apps/server/test/acquisition-organize/organize-episode-recovery.test.ts @@ -0,0 +1,406 @@ +import { access, mkdir, readFile, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + executeRecord, + type FrozenPlan, + planImport, +} from "../../src/modules/acquisition-organize/services/organize.ts"; +import { + resumeEpisodeOrganize, + sweepEpisodeOrganize, +} from "../../src/modules/acquisition-organize/services/organize-episode-recovery.ts"; +import { checkRecovery } from "../../src/modules/agent/investigate.ts"; +import { reresolveEntityEpisodes } from "../../src/modules/sources-identity/service/episode-resolution.ts"; +import { createFmHarness, type FmHarness, pgAvailable, seedDownload, seedEntity } from "../file-manager/harness.ts"; +import { setWordsText } from "../support/identify-words.ts"; +import { seedRelease } from "./harness.ts"; + +describe.skipIf(!(await pgAvailable()))("季集事实补齐后继续原整理", () => { + let h: FmHarness; + beforeAll(async () => { + h = await createFmHarness("episode_resume", {}, { runTransfers: false }); + }); + afterAll(async () => { + await h?.close(); + }); + + async function show(title: string) { + const id = await seedEntity(h.db, { title, kind: "series", year: 2024 }); + await h.db + .updateTable("media_entity") + .set({ metadata: JSON.stringify({ airStatus: { value: "Ended", source: "tmdb" } }) }) + .where("id", "=", id) + .execute(); + await h.db + .insertInto("content_unit") + .values({ + media_entity_id: id, + kind: "season", + season_number: 1, + episode_total: 12, + status: "confirmed", + source: "tmdb", + }) + .execute(); + return id; + } + + it("按路径整理无需订阅也保留全部文件样本，修正规则后恢复，重放不重复排队", async () => { + const entity = await show("Episode Resume"); + const paths = ["Episode.Resume.S02E01.1080p.mkv", "Episode.Resume.S02E02.1080p.mkv"].map((name) => + join(h.dl, name), + ); + for (const path of paths) await writeFile(path, path.padEnd(128, "x")); + const original = await planImport(h.module.deps().ao, { + sourcePaths: paths, + mediaEntityId: entity, + trigger: "manual", + actorType: "user", + action: "copy", + targetDir: join(h.lib, "preferred"), + namingMode: "keep", + scrapeAfterImport: false, + minSizeBytes: 0, + }); + expect(original.status).toBe("failed"); + const decision = await h.db + .selectFrom("decision") + .selectAll() + .where("kind", "=", "episode_mapping") + .where("subject_id", "=", entity) + .executeTakeFirstOrThrow(); + expect((decision.evidence as { fileSamples: unknown[] }).fileSamples).toHaveLength(2); + expect( + ( + await checkRecovery(h.module.deps().ao.ctx, [ + { capabilityId: "episode_mapping.apply_rule", input: { decisionId: decision.id }, status: "succeeded" }, + ]) + ).state, + ).toBe("pending"); + expect(await resumeEpisodeOrganize(h.module.deps().ao, entity)).toEqual({ resumed: 0, waiting: 1 }); + await setWordsText(h.db, entity, "S02 >> S01 EP\n"); + await reresolveEntityEpisodes(h.module.deps().ao.ctx, h.db, entity); + expect( + await h.db.selectFrom("outbox_event").select("id").where("type", "=", "media.episode_mapping_changed").execute(), + ).not.toHaveLength(0); + const concurrent = await Promise.all([ + resumeEpisodeOrganize(h.module.deps().ao, entity), + resumeEpisodeOrganize(h.module.deps().ao, entity), + ]); + expect(concurrent.every((r) => r.waiting === 0)).toBe(true); + expect( + await h.db + .selectFrom("organization_record") + .select("id") + .where("media_entity_id", "=", entity) + .where("id", "!=", original.recordId!) + .execute(), + ).toHaveLength(1); + expect( + h.engine.enqueued.filter( + (e) => + e.task === "verifyAndImport" && + (e.input as { recordId?: string }).recordId && + (e.input as { recordId?: string }).recordId !== original.recordId, + ), + ).toHaveLength(1); + const next = await h.db + .selectFrom("organization_record") + .selectAll() + .where("media_entity_id", "=", entity) + .where("id", "!=", original.recordId!) + .executeTakeFirstOrThrow(); + const plan = next.plan as unknown as FrozenPlan; + expect(plan.request).toMatchObject({ + action: "copy", + targetDir: join(h.lib, "preferred"), + namingMode: "keep", + scrapeAfterImport: false, + }); + expect(plan.retryOfRecordId).toBe(original.recordId); + expect(await resumeEpisodeOrganize(h.module.deps().ao, entity)).toEqual({ resumed: 0, waiting: 0 }); + expect((await executeRecord(h.module.deps().ao, next.id)).status).toBe("succeeded"); + const future = await planImport(h.module.deps().ao, { + sourcePaths: [paths[0]!], + mediaEntityId: entity, + trigger: "manual", + actorType: "user", + dryRun: true, + minSizeBytes: 0, + }); + expect(Object.values(future.plan!.facts)[0]!.episodeResolution).toMatchObject({ + method: "rule", + canonical: { season: 1, episodes: { start: 1, end: 1 } }, + }); + }); + + it("部分移动仅继续剩余文件，已入库内容保持原样", async () => { + const entity = await show("Partial Resume"); + const first = join(h.dl, "Partial.Resume.S01E01.1080p.mkv"); + const second = join(h.dl, "Partial.Resume.S02E02.1080p.mkv"); + await writeFile(first, "first".padEnd(128, "x")); + await writeFile(second, "second".padEnd(128, "y")); + const original = await planImport(h.module.deps().ao, { + sourcePaths: [first, second], + mediaEntityId: entity, + trigger: "manual", + actorType: "user", + action: "move", + scrapeAfterImport: false, + minSizeBytes: 0, + }); + expect((await executeRecord(h.module.deps().ao, original.recordId!)).status).toBe("partial"); + const decision = await h.db + .selectFrom("decision") + .select("evidence") + .where("kind", "=", "episode_mapping") + .where("subject_id", "=", entity) + .executeTakeFirstOrThrow(); + expect((decision.evidence as { fileSamples: { path: string }[] }).fileSamples.map((f) => f.path).sort()).toEqual( + [first, second].sort(), + ); + await expect(access(first)).rejects.toThrow(); + const record = await h.db + .selectFrom("organization_record") + .select("plan") + .where("id", "=", original.recordId!) + .executeTakeFirstOrThrow(); + const target = (record.plan as unknown as FrozenPlan).organize.files.find((f) => f.src === first)!.dst; + const content = await readFile(target, "utf8"); + await setWordsText(h.db, entity, "S02 >> S01 EP\n"); + expect(await resumeEpisodeOrganize(h.module.deps().ao, entity)).toEqual({ resumed: 1, waiting: 0 }); + const next = await h.db + .selectFrom("organization_record") + .selectAll() + .where("media_entity_id", "=", entity) + .where("id", "!=", original.recordId!) + .executeTakeFirstOrThrow(); + expect((next.plan as unknown as FrozenPlan).request?.sourcePaths).toEqual([second]); + expect((await executeRecord(h.module.deps().ao, next.id)).status).toBe("succeeded"); + expect(await readFile(target, "utf8")).toBe(content); + expect( + await h.db + .selectFrom("media_asset") + .select("id") + .where("media_entity_id", "=", entity) + .where("status", "=", "active") + .execute(), + ).toHaveLength(2); + }); + + it("下载任务部分移动后仍可恢复，不重新检查已经移走的源文件", async () => { + const entity = await show("Download Resume"); + const name = "Download.Resume"; + const dir = join(h.dl, name); + await mkdir(dir); + const paths = ["Download.Resume.S01E01.1080p.mkv", "Download.Resume.S02E02.1080p.mkv"].map((f) => join(dir, f)); + for (const path of paths) await writeFile(path, path.padEnd(128, "x")); + const dt = await seedDownload(h.db, h, { name, status: "completed" }); + const site = await h.db + .insertInto("site") + .values({ key: "resume", name: "恢复测试站", adapter: "nexusphp", base_url: "https://pt.example.test" }) + .returning("id") + .executeTakeFirstOrThrow(); + const release = await seedRelease(h.db, { + siteId: site.id, + entityId: entity, + title: name, + size: 256, + hash: "a".repeat(40), + hr: false, + }); + await h.db + .updateTable("download_task") + .set({ + media_entity_id: entity, + source_release_id: release, + file_list: JSON.stringify(paths.map((p) => ({ path: p.slice(h.dl.length + 1), size: 128 }))), + }) + .where("id", "=", dt) + .execute(); + const original = await planImport(h.module.deps().ao, { + downloadTaskId: dt, + mediaEntityId: entity, + trigger: "manual", + actorType: "user", + action: "move", + scrapeAfterImport: false, + minSizeBytes: 0, + }); + expect((await executeRecord(h.module.deps().ao, original.recordId!)).status).toBe("partial"); + await expect(access(paths[0]!)).rejects.toThrow(); + await setWordsText(h.db, entity, "S02 >> S01 EP\n"); + expect(await resumeEpisodeOrganize(h.module.deps().ao, entity)).toEqual({ resumed: 1, waiting: 0 }); + const next = await h.db + .selectFrom("organization_record") + .selectAll() + .where("download_task_id", "=", dt) + .where("id", "!=", original.recordId!) + .executeTakeFirstOrThrow(); + expect((next.plan as unknown as FrozenPlan).request).toMatchObject({ + downloadTaskId: dt, + sourcePaths: [paths[1]!], + }); + expect((await executeRecord(h.module.deps().ao, next.id)).status).toBe("succeeded"); + expect( + await h.db + .selectFrom("media_asset") + .select("id") + .where("media_entity_id", "=", entity) + .where("status", "=", "active") + .execute(), + ).toHaveLength(2); + }); + + it("补齐在播季表后沿用原资源发布时间，恢复正确的绝对集号换算", async () => { + const entity = await show("Dated Resume"); + const name = "Dated.Resume.S02E13.1080p.mkv"; + const path = join(h.dl, name); + await writeFile(path, "dated".padEnd(128, "x")); + const dt = await seedDownload(h.db, h, { name, status: "completed" }); + const site = await h.db + .insertInto("site") + .values({ + key: "dated-resume", + name: "发布时间测试站", + adapter: "nexusphp", + base_url: "https://dated.example.test", + }) + .returning("id") + .executeTakeFirstOrThrow(); + const release = await seedRelease(h.db, { + siteId: site.id, + entityId: entity, + title: name, + size: 128, + hash: "b".repeat(40), + }); + await h.db + .updateTable("source_release") + .set({ source_published_at: new Date("2026-01-08T16:30:00Z") }) + .where("id", "=", release) + .execute(); + await h.db + .updateTable("download_task") + .set({ + media_entity_id: entity, + source_release_id: release, + file_list: JSON.stringify([{ path: name, size: 128 }]), + }) + .where("id", "=", dt) + .execute(); + const original = await planImport(h.module.deps().ao, { + downloadTaskId: dt, + mediaEntityId: entity, + trigger: "manual", + actorType: "user", + action: "copy", + scrapeAfterImport: false, + minSizeBytes: 0, + }); + expect(original.status).toBe("failed"); + await h.db + .updateTable("media_entity") + .set({ metadata: JSON.stringify({ airStatus: "Returning Series" }) }) + .where("id", "=", entity) + .execute(); + const season = await h.db + .insertInto("content_unit") + .values({ + media_entity_id: entity, + kind: "season", + season_number: 2, + episode_total: 12, + status: "confirmed", + source: "tmdb", + }) + .returning("id") + .executeTakeFirstOrThrow(); + await h.db + .insertInto("content_unit") + .values({ + media_entity_id: entity, + kind: "episode", + parent_unit_id: season.id, + season_number: 2, + episode_number: 1, + air_date: "2026-01-08", + status: "confirmed", + source: "tmdb", + }) + .execute(); + expect(await resumeEpisodeOrganize(h.module.deps().ao, entity)).toEqual({ resumed: 1, waiting: 0 }); + const next = await h.db + .selectFrom("organization_record") + .selectAll() + .where("download_task_id", "=", dt) + .where("id", "!=", original.recordId!) + .executeTakeFirstOrThrow(); + expect(Object.values((next.plan as unknown as FrozenPlan).facts)[0]!.episodeResolution).toMatchObject({ + method: "absolute_in_season", + canonical: { season: 2, episodes: { start: 1, end: 1 } }, + }); + expect((await executeRecord(h.module.deps().ao, next.id)).status).toBe("succeeded"); + }); + + it("暂停期间修正规则，启用后只恢复原目录的未解决文件", async () => { + const entity = await show("Directory Resume"); + const dir = join(h.dl, "Directory.Resume"); + await mkdir(dir); + const first = join(dir, "Directory.Resume.S01E01.1080p.mkv"); + const second = join(dir, "Directory.Resume.S02E02.1080p.mkv"); + for (const path of [first, second]) await writeFile(path, path.padEnd(128, "x")); + const original = await planImport(h.module.deps().ao, { + sourcePaths: [dir], + mediaEntityId: entity, + trigger: "manual", + actorType: "user", + action: "copy", + scrapeAfterImport: false, + minSizeBytes: 0, + }); + expect((await executeRecord(h.module.deps().ao, original.recordId!)).status).toBe("partial"); + const added = join(dir, "Directory.Resume.S01E03.1080p.mkv"); + await writeFile(added, "later".padEnd(128, "x")); + await h.db + .insertInto("settings") + .values({ key: "automation", value: JSON.stringify({ enabled: false }) }) + .onConflict((oc) => oc.column("key").doUpdateSet({ value: JSON.stringify({ enabled: false }) })) + .execute(); + await setWordsText(h.db, entity, "S02 >> S01 EP\n"); + await reresolveEntityEpisodes(h.module.deps().ao.ctx, h.db, entity); + expect(await resumeEpisodeOrganize(h.module.deps().ao, entity)).toEqual({ resumed: 0, waiting: 0 }); + expect(await sweepEpisodeOrganize(h.module.deps().ao)).toBe(0); + await h.db + .updateTable("settings") + .set({ value: JSON.stringify({ enabled: true }) }) + .where("key", "=", "automation") + .execute(); + expect(await sweepEpisodeOrganize(h.module.deps().ao)).toBe(1); + expect( + h.engine.enqueued.some( + (e) => e.task === "verifyAndImport" && (e.input as { mediaEntityId?: string }).mediaEntityId === entity, + ), + ).toBe(true); + expect(await resumeEpisodeOrganize(h.module.deps().ao, entity)).toEqual({ resumed: 1, waiting: 0 }); + const next = await h.db + .selectFrom("organization_record") + .selectAll() + .where("media_entity_id", "=", entity) + .where("id", "!=", original.recordId!) + .executeTakeFirstOrThrow(); + expect((next.plan as unknown as FrozenPlan).request?.sourcePaths).toEqual([second]); + expect((await executeRecord(h.module.deps().ao, next.id)).status).toBe("succeeded"); + expect( + await h.db + .selectFrom("media_asset") + .select("id") + .where("media_entity_id", "=", entity) + .where("status", "=", "active") + .execute(), + ).toHaveLength(2); + expect(await readFile(added, "utf8")).toBe("later".padEnd(128, "x")); + expect(await sweepEpisodeOrganize(h.module.deps().ao)).toBe(0); + }); +}); diff --git a/apps/server/test/sources-identity/organize-episode-samples.test.ts b/apps/server/test/sources-identity/organize-episode-samples.test.ts index ca3abb68..2d8b35da 100644 --- a/apps/server/test/sources-identity/organize-episode-samples.test.ts +++ b/apps/server/test/sources-identity/organize-episode-samples.test.ts @@ -264,6 +264,25 @@ describe.skipIf(!pg)("整理真实文件的季集映射闭环", () => { ).rejects.toThrow("已对上的资源"); }); + it("首次混合整理保留已对上的文件，纯文件偏移也要通过回归保护", async () => { + const entityId = await series(); + await settleOrganizeEpisodeDecisions(h.ctx, h.db, { + recordId: randomUUID(), + entityId, + files: [await file(entityId, 1), await file(entityId, 40)], + }); + const d = (await pending(entityId))[0]!; + expect(episodeFileSamples(d.evidence)).toHaveLength(2); + const simulation = await simulateForAgent(h.db, { decisionId: d.id, line: "S01 @ANi >> S02 EP-30" }, null); + expect(simulation.summary).toMatchObject({ samples: 2, resolved: 1, regressions: 1, ok: false }); + await expect( + applyAgentEpisodeRule(h.ctx, { decisionId: d.id, line: simulation.line, reasons: ["同批文件"] }, ACTOR, null), + ).rejects.toThrow("已对上的文件"); + expect( + await h.db.selectFrom("identify_words").select("id").where("media_entity_id", "=", entityId).execute(), + ).toHaveLength(0); + }); + it("保存部分有效的映射保留未决文件与已解决样本，不误关闭事项", async () => { const entityId = await series(); const files = [await file(entityId, 40), await file(entityId, 60)]; diff --git a/apps/web/app/components/settings/agent.tsx b/apps/web/app/components/settings/agent.tsx index 29adea51..177973be 100644 --- a/apps/web/app/components/settings/agent.tsx +++ b/apps/web/app/components/settings/agent.tsx @@ -105,14 +105,14 @@ export const BUDGET_CARD: CardSpec = { export const INVESTIGATE_CARD: CardSpec = { anchor: "investigate", title: "自动处理", - description: "整理时识别不到作品就叫智能体识别；下载失败、集数对不上时自动发起一次调查，受下面的上限约束。", + description: "整理时作品或季集认不出，智能体调查一次；其他下载问题按自动调查设置处理。", fields: [ { name: "agent.autoIdentifyOrganize", - label: "整理识别不到作品时叫智能体识别", + label: "整理识别不到作品或季集时，交给智能体", kind: "switch", wide: true, - hint: "认出来自动重新整理；认不出的留在整理记录「需要处理」里", + hint: "认出来自动重新整理；季集规则有持续授权才自动写入，否则留下建议；拿不准的交给你确认", }, { name: "agent.autoInvestigate", label: "自动调查", kind: "switch", wide: true }, { diff --git a/docs/plan/README.md b/docs/plan/README.md index 2f73f4a1..a4984c64 100644 --- a/docs/plan/README.md +++ b/docs/plan/README.md @@ -17,7 +17,7 @@ ## 当前状态 -- 版本 v0.3.37，本机测试环境已部署。 +- 本机测试环境已部署；当前运行版本以 `/api/v1/health` 为准，发布走 `bun run release`。 - 开发与发布约定见仓库根 [CLAUDE.md](../../CLAUDE.md)。 ## 待办 diff --git a/docs/porting/moviepilot-v3-comparison.md b/docs/porting/moviepilot-v3-comparison.md index 8d7d16e3..03bf209c 100644 --- a/docs/porting/moviepilot-v3-comparison.md +++ b/docs/porting/moviepilot-v3-comparison.md @@ -349,3 +349,17 @@ MP 依据 §1 固定提交 `app/chain/subscribe/*`，旧版逐函数记录见 [ - 第 1、4、12 项：表单收敛为「基础」＝作品与范围、下载、洗版；「进阶」＝过滤（规则组、包含／排除、分辨率、质量、特效、关键词）、搜索；收录版本模板／顺序、优先级、自动补全折叠到「多版本（一般不用改）」。 - 第 7 项：洗版已是 MP 三种（电影洗版、按集洗版、全集洗版），文案改为「拿到规则组最高档即完成」，不再另做。 - 第 5、10、11、13、14 项维持现状；第 15 项不做。 + +## 8 整理闭环对照（2026-10-13，MP v3.1.5） + +本节为上游发布说明与具体提交的源码对照，没有实测 MP 页面。参考版本为 [v3.1.5](https://github.com/jxxghp/MoviePilot/releases/tag/v3.1.5)，北京时间 2026-10-13 08:46 发布。此前章节仍保留各自固定版本。 + +| 上游改动与证据 | 本项目取舍 | +| --- | --- | +| [c670432](https://github.com/jxxghp/MoviePilot/commit/c6704328829c32676674b520aa5f7b7d6a1447aa)：大小策略拒绝覆盖要保留已有成功历史 | 保留每次整理记录和资产事实，不把规则拒绝视作旧文件失效；复用现有替换核验机制 | +| [4357035](https://github.com/jxxghp/MoviePilot/commit/435703573e8811d8a7e43a7356cce69f3e09a5ff)：provider 检查点保存规划时间，区分计划和执行字段不完整 | 已有持久引擎与冻结文件计划，补齐可序列化原请求；执行恢复复用账本，身份纠正重新规划，保留一次性选项 | +| [de7726c](https://github.com/jxxghp/MoviePilot/commit/de7726c6fc54b74592161007b637414ae2c3f38e)：手动下载部分入库也同步订阅进度 | 已有覆盖账本统一重算，不另建手动下载完成口径；保留 partial 的内部防重整约定，修复部分移动的再次整理来源 | +| [6ef8408](https://github.com/jxxghp/MoviePilot/commit/6ef8408f432dde643aee2ba7078f59d2776f84ce)：TMDB 辅助补齐缺失分类字段 | 继续以 TMDB 为主要规范身份，补齐事实不让 Agent 改分类或收录规则 | +| 前版 [bb29451](https://github.com/jxxghp/MoviePilot/commit/bb29451)：账本之外的未知执行异常停止自动重放，转人工复核 | 我们已有有限任务重试和整理巡检，不照搬 MP 租约与人工记账 API；补上重放的主文件／附属文件依赖，并让未知问题留在原整理记录处理 | + +本轮额外闭环针对既有需求：真正失败的文件名参与季集 inspect／simulate，规则修正通过领域事件通知整理；重试入口只继续剩余文件，保持来源边界与原一次性选项。没有添加通用恢复引擎、第二套季集映射或独立处理中心。验证以针对行为回归及合并后的全量 CI 为准，日志放 `.local/organize-closure/`。 diff --git a/docs/runbook/dev.md b/docs/runbook/dev.md index ad006983..e0951dda 100644 --- a/docs/runbook/dev.md +++ b/docs/runbook/dev.md @@ -97,6 +97,10 @@ curl -H "Authorization: Bearer umt-dev-admin-key" "localhost:4390/api/v1/events? 手动下载以默认值直接提交，高级选项仅覆盖本次操作。搜索与工作台复用 `ManualDownloadActions` 的提交反馈、未识别确认和高级选项窗口，资源摘要与 API 参数仍由各入口转换；原站页面保留必须先解析详情的准备步骤。默认下载不先加载目录、作品详情或已有版本，不为三个入口引入通用流程引擎。 +整理恢复区分两种情况：文件执行中断从原冻结计划和持久文件结果继续；作品或季集事实纠正后，通过 `retryOrganizeRecord` 从原业务请求重新规划。同一原记录的重规划串行，避免事件和巡检同时冻结重复计划。原请求保存一次性目标、操作方式和覆写，函数、任务身份与路径守卫不序列化，恢复时重新检查来源边界。已成功文件与资产保持原状，恢复主文件失败状态时也恢复其附属文件依赖。 + +季集计算和识别词归 sources-identity，整理请求、文件操作与入库状态归 acquisition-organize。整理把真实失败文件和同批正常文件样本交给映射事项，模拟与写入共用同一判断；映射重算后以 `media.episode_mapping_changed` 通知整理模块，只有季集已经确定的未完成记录重新规划。恢复预检与规划复用 `planFileEpisodes`，保留原资源的发布时间等解析上下文。Agent 的调查运行、预算和授权沿用现有机制，不另外创建恢复框架；身份补齐、重新排队和实际入库完成分别按业务事实核实。 + 维护时先验证真实的成功、失败和恢复路径,尤其是授权、文件替换、取消与外部副作用；删除空壳或无使用者的机制时同步删除其专属测试与文档。不要为了目录对称增加转发层,也不要把不同阶段的状态合并成一个枚举。 ## 进程角色 diff --git a/docs/runbook/organize.md b/docs/runbook/organize.md index a5fc21a5..0663d87a 100644 --- a/docs/runbook/organize.md +++ b/docs/runbook/organize.md @@ -47,6 +47,16 @@ 不做:切换结构后自动改名已整理的文件(仍用「重新整理」)、剧集组季的独立海报。测试:`apps/server/test/file-manager/organize-episode-structure.test.ts`、`apps/web/app/components/organize-wizard/model.test.ts`「季集结构」。 +## 识别纠正与继续整理 + +整理记录保留本次业务请求。没识别出作品时，下载来源和按路径来源都能由人工或智能体指定作品，再沿用原来的目标目录、复制／移动、命名和文件覆写重新整理。只有重新规划已接受才算身份事项处理成功，实际入库结果仍看后续整理记录；重新规划失败会留下原因。 + +季集待确认按作品、来源季和发布组归集真实文件样本，资源站标题与文件名不同的情况也参与模拟。作品识别词必须同时解决待确认文件且不破坏已对上的资源，写入后同类文件沿用这条规则。映射或季表重算后，季集已确定的整理自动继续剩余文件；本次临时选择的季集结构保持原约定，缺少原请求的旧记录由人工重新选择来源。 + +设置中的整理自动识别控制作品与季集的单次调查。身份有把握时可重新整理；季集规则自动写入仍需要有效持续授权，没有授权则保留模拟建议供确认。无法确认的交还人工，不改变收录分辨率、版本优先级或洗版条件。 + +执行中断继续原文件账本。主视频已经失败时，尚未执行的字幕等附属文件留在源目录。部分移动的重新整理入口按每个文件的实际结果选择源文件或已入库目标，不把成功移动的文件当成丢失。 + ## Alist / OpenList 存储 媒体库目录、整理来源和文件管理都可以用登记在 设置 → 媒体库与目录 →「存储」里的 Alist / OpenList（设计 `docs/design/storage-alist.md`，使用与真机验收 `docs/runbook/storage-alist.md`）。 diff --git a/packages/contracts/src/events.ts b/packages/contracts/src/events.ts index eceb3111..3a745e85 100644 --- a/packages/contracts/src/events.ts +++ b/packages/contracts/src/events.ts @@ -79,6 +79,7 @@ export const EventType = { mediaWatched: "media.watched", mediaMetadataRefreshed: "media.metadata_refreshed", mediaEpisodeStructureChanged: "media.episode_structure_changed", + mediaEpisodeMappingChanged: "media.episode_mapping_changed", siteUnreachable: "site.unreachable", siteRecovered: "site.recovered", } as const; @@ -282,6 +283,8 @@ export const MediaEpisodeStructureChangedPayload = z.object({ fromGroupId: z.string().nullable(), toGroupId: z.string().nullable(), }); +/** 作品季集规则或季表重算完成，整理模块按当前事实恢复尚未解决的文件。 */ +export const MediaEpisodeMappingChangedPayload = z.object({ mediaEntityId: Id }); /** 站点不可达:登录失效、列表抓取连续失败或检测失败;reason 为原因码 */ export const SiteUnreachablePayload = z.object({ siteId: Id, reason: z.string(), message: z.string().nullable() }); @@ -325,6 +328,7 @@ export const EVENT_PAYLOADS = { "media.watched": MediaWatchedPayload, "media.metadata_refreshed": MediaMetadataRefreshedPayload, "media.episode_structure_changed": MediaEpisodeStructureChangedPayload, + "media.episode_mapping_changed": MediaEpisodeMappingChangedPayload, "site.unreachable": SiteUnreachablePayload, "site.recovered": SiteRecoveredPayload, } as const satisfies Record; @@ -408,6 +412,12 @@ export const LIFECYCLE_EVENTS: readonly { type: EventType; label: string; emitte emittedBy: "sources-identity", when: "作品换了季集结构", }, + { + type: "media.episode_mapping_changed", + label: "季集映射已重算", + emittedBy: "sources-identity", + when: "作品识别词或季表重算完成", + }, { type: "site.unreachable", label: "站点不可达", diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index 45b1e512..ee8f7f20 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -212,7 +212,7 @@ export const AgentRuntimeSettings = z.object({ maxTokensPerRun: z.number().int().positive().nullable().optional(), /** 已弃用(2026-10-06 起不限单次费用,旧值兼容读取、不生效) */ maxCostUsdPerRun: z.number().positive().nullable().optional(), - /** 整理时没识别出作品:自动派智能体识别一次,认出来就按该作品重新整理(不需要持续授权) */ + /** 整理作品或季集未知时调查一次；指定作品可恢复，季集规则写入仍需持续授权。 */ autoIdentifyOrganize: z.boolean(), /** 后台自动调查的工具调用上限(比用户委托更小) */ investigateMaxToolCalls: z.number().int().min(1).max(100),