fix(organize): resume original imports after episode corrections

2026-10-13 16:50:51 +08:00
1 parent 679389f854
commit e874247e7c
@@ -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,
@@ -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}` : "")
          );
        }),
@@ -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<boolean>`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<number> {
+  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;
+}
@@ -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<PlanRes
  let sourceRoot: string | null = null;
  const topo = dt ? (await loadInstances(db, { enabledOnly: false, ids: [dt.instance_id] }))[0] : undefined;
  let qbFiles: QbFileLike[] = [];
-  if (dt) {
+  if (dt && !(req.retryOfRecordId && req.sourcePaths?.length)) {
    if (!topo || !dt.save_path) return fail("source_path_unknown");
    const base = hostPathOf(topo, dt.save_path);
    if (!base) return fail("path_mapping_missing", { savePath: dt.save_path });
@@ -352,7 +353,12 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<PlanRes
        files.push({ path: p, size: st.facts.size, device: st.facts.dev });
      }
    }
-    sourceRoot = req.sourcePaths.length === 1 ? req.sourcePaths[0]!.replace(/\/+$/, "") : null;
+    sourceRoot =
+      dt?.content_path && topo
+        ? hostPathOf(topo, dt.content_path)
+        : req.sourcePaths.length === 1
+          ? req.sourcePaths[0]!.replace(/\/+$/, "")
+          : null;
    if (sourceRoot && files.length === 1 && files[0]!.path === sourceRoot) {
      sourceRoot = dirname(sourceRoot);
    }
@@ -748,6 +754,7 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<PlanRes
      const raw = parsed;
      const r = episodes.map(raw, episodes.opts);
      resolution = r.resolution;
+      episodes.samples.push({ path: f.path, resolution: r.resolution, parsed: raw });
      if (r.resolution.method === "unresolved" || r.resolution.special) {
        const unresolved = r.resolution.method === "unresolved";
        mappedOverrides[f.path] = {
@@ -756,7 +763,6 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<PlanRes
            : { reason: "special_extra", detail: `rule:${r.resolution.special}` },
        };
        unresolvedEpisodes[f.path] = episodeFacts(r.resolution);
-        if (unresolved) episodes.unresolved.push({ path: f.path, resolution: r.resolution, parsed: raw });
        continue;
      }
      parsed = r.parsed;
@@ -1183,15 +1189,25 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<PlanRes
    await emitRecord(d, trx, rec.id);
    return { recordId: rec.id, status: rec.status, reason };
  });
-  // 文件集数待确认:并入关联资源的 episode_mapping 待处理(待处理按资源归集;没有关联资源只记整理失败原因)
+  // 文件是独立样本；资源标题已解决也不能覆盖真正失败的文件季集。
  // 临时结构下的待确认是相对指定结构的,不并进作品的集数待确认
-  const pending = episodes?.unresolved[0];
-  if (pending && episodes?.release && !episodes.toEntity) {
-    await settleEpisodeDecision(d.ctx, db, {
-      sourceReleaseId: episodes.release.id,
+  if (result.recordId && episodes?.samples.some((f) => 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: [],
  };
 }

@@ -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")
@@ -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";
@@ -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,
@@ -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;
 }