feat(identity): content unit data hygiene — specials, provisional promotion, duplicate merge (DH-1)

- metadata.includeSpecials (default true): full refresh and the language
  backfill fetch TMDB season 0 (specials) when present, like MP; season 0
  stays out of the episode-total ledger
- refreshMetadata promotes provisional season/episode units that match the
  TMDB season/episode layout to confirmed (source=tmdb), keeping ids and all
  references; non-matching units stay provisional
- content_unit.source (migration 0053): release / file / tmdb / user
- duplicate units (same entity/kind/season/episode/content type/segment) are
  merged as a refresh step and via `scripts/identity-cleanup.ts
  --merge-duplicates` (dry-run by default); keeper = most referenced, then
  confirmed, then earliest; references auto-discovered via FKs and repointed
- ensureUnits takes a per-unit advisory xact lock to stop concurrent lazy
  creation from producing duplicates
- runbooks updated; user DB read-only dry-run: 征途 S01E09/E10 4 units each
  -> 1 each (6 removed), not applied

Co-Authored-By: Claude Opus 5.5 <[email protected]>

apps/server/src/modules/acquisition-organize/services/organize.ts

@@ -1712,6 +1712,7 @@ async function ensureEpisodeUnits(
         season_number: season,
         episode_number: ep,
         status: "provisional",
+        source: "file",
         display_order: ep ?? season,
       })
       .returning("id")

apps/server/src/modules/sources-identity/routes/identity.ts

@@ -326,6 +326,7 @@ export function identityRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
       status: "status",
       numberingLocked: "numbering_locked",
       fieldLocks: "field_locks",
+      source: "source",
     };
     for (const [k, col] of Object.entries(map)) {
       const v = (body as Record<string, unknown>)[k];
@@ -334,7 +335,10 @@ export function identityRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
     if (body.metadata !== undefined) patch.metadata = jsonb(body.metadata);
     // 人工改编号即确认并锁定编号(REQ §5.4:编号可修订,人工确认后刷新不得改)
     if (touchesNumbering && body.numberingLocked === undefined) patch.numbering_locked = true;
-    if (touchesNumbering && body.status === undefined && unit.status === "provisional") patch.status = "confirmed";
+    if (touchesNumbering && body.status === undefined && unit.status === "provisional") {
+      patch.status = "confirmed";
+      if (body.source === undefined) patch.source = "user";
+    }
     const row = await db()
       .updateTable("content_unit")
       .set(patch)

apps/server/src/modules/sources-identity/service/entities.ts

@@ -6,6 +6,7 @@ import { toContractMediaKind } from "@umt/mapping";
 import { jsonb } from "../rows.ts";
 import type { Actor, Exec } from "./common.ts";
 import { project, writeTmdbFields } from "./metadata.ts";
+import { lockUnitKey } from "./unit-merge.ts";
 
 // 本地作品 / 内容单元 / 外部身份 / 别名的写入辅助。所有写入在调用方事务内完成。
 
@@ -228,6 +229,8 @@ async function findOrCreateUnit(
   episode: number | null,
   parentId: string | null,
 ): Promise<string> {
+  // 并发识别同一集时串行化,避免建出重复单元(DH-1:征途 E09/E10 各 4 条即此竞态)
+  await lockUnitKey(exec, entityId, kind, season, episode);
   let q = exec
     .selectFrom("content_unit")
     .select(["id"])
@@ -247,6 +250,7 @@ async function findOrCreateUnit(
       season_number: season,
       episode_number: episode,
       status: "provisional",
+      source: "release",
       display_order: episode ?? season,
     })
     .returning(["id"])

apps/server/src/modules/sources-identity/service/metadata-rich.ts

@@ -19,7 +19,7 @@ import type { ServerContext } from "../../types.ts";
 import { jsonb } from "../rows.ts";
 import type { SourcesRuntime } from "../runtime.ts";
 import { type Actor, audit, type Exec } from "./common.ts";
-import { ensureSeason } from "./metadata-totals.ts";
+import { ensureSeason, promoteUnits } from "./metadata-totals.ts";
 
 // MD-1 语言回退链 + 每集元数据(REQ §7.3 字段级):
 // - 作品:TMDB 富详情(fetchRichDetail)按设置 metadata.languages 取标题 / 简介 / 标语,另存中文名 titleZh、类型、评分、
@@ -28,6 +28,8 @@ import { ensureSeason } from "./metadata-totals.ts";
 // - 季 / 集:fetchRichSeason 的每集 name / overview / airDate / runtime / stillPath / vote 写 metadata_field(content_unit_id),
 //   投影到 content_unit.metadata(字段级,可锁),name → title 列、airDate → air_date 列;只写变化的字段,
 //   不改编号、不碰路径绑定与资产覆盖。集单元不存在时按 TMDB 建(confirmed,挂季单元)。
+// - DH-1:季 0(特别篇)按设置 metadata.includeSpecials(缺省开)在 TMDB 有季 0 时一并拉取与建单元(对照 MP);
+//   与 TMDB 季集结构对上的临时单元转正式(confirmed、来源 tmdb,保留 id 与所有引用),对不上的保持临时。
 
 type FieldValue = { value: unknown; language: string | null };
 
@@ -38,6 +40,12 @@ export async function metadataLanguages(ctx: ServerContext): Promise<string[]> {
   return langs?.length ? langs : [...DEFAULT_METADATA_LANGUAGES];
 }
 
+/** 设置 metadata.includeSpecials:full 刷新 / 回刷是否包含季 0(特别篇);缺省 true */
+export async function metadataIncludeSpecials(ctx: ServerContext): Promise<boolean> {
+  const m = await ctx.settings.get("metadata").catch(() => null);
+  return (m as { includeSpecials?: boolean } | null)?.includeSpecials !== false;
+}
+
 /** 富详情 → 作品字段(值 + 语言);海报 / 背景图没有时不写(保留旧值) */
 export function richEntityFields(r: TmdbRichDetail): Partial<Record<EntityMetadataField, FieldValue>> {
   const v = (value: unknown, language: string | null = null): FieldValue => ({ value, language });
@@ -296,12 +304,14 @@ export interface SeasonMetadataResult {
   created: number;
   /** 字段有变化(投影改写)的内容单元数(含季单元) */
   changedUnits: number;
+  /** DH-1:临时转正式的集单元数(季单元转正不计) */
+  promoted: number;
 }
 
 /**
  * 季详情 → 季单元与集单元的字段级元数据。集单元按(作品, 季, 集号)找最早建的未退役单元,没有时:
  * create = true(full 刷新 / 追踪中的季)建 confirmed 单元;否则只更新已有单元(集单元仍由发布 / 文件懒建)。
- * 只写变化的来源字段,只投影有变化的单元。
+ * 对上 TMDB 集的临时单元转正式;季 0(特别篇)同样处理。只写变化的来源字段,只投影有变化的单元。
  */
 export async function applySeasonMetadata(
   exec: Exec,
@@ -312,26 +322,28 @@ export async function applySeasonMetadata(
   opts: { create?: boolean } = {},
 ): Promise<SeasonMetadataResult> {
   const create = opts.create ?? true;
-  const out: SeasonMetadataResult = { season: sd.seasonNumber, episodes: 0, created: 0, changedUnits: 0 };
-  if (sd.seasonNumber < 1) return out;
+  const out: SeasonMetadataResult = { season: sd.seasonNumber, episodes: 0, created: 0, changedUnits: 0, promoted: 0 };
+  if (sd.seasonNumber < 0) return out;
   const seasonUnit = await ensureSeason(exec, entityId, sd.seasonNumber, sd.name);
   const units = await exec
     .selectFrom("content_unit")
-    .select(["id", "episode_number", "metadata"])
+    .select(["id", "episode_number", "metadata", "status"])
     .where("media_entity_id", "=", entityId)
     .where("kind", "=", "episode")
     .where("season_number", "=", sd.seasonNumber)
     .where("status", "!=", "retired")
     .orderBy("created_at")
     .execute();
-  const unitOf = new Map<number, { id: string; metadata: unknown }>();
+  const unitOf = new Map<number, { id: string; metadata: unknown; status?: string }>();
   for (const u of units)
     if (u.episode_number !== null && !unitOf.has(u.episode_number)) unitOf.set(u.episode_number, u);
   const eps = sd.episodes.filter((e) => e.episodeNumber >= 1).slice(0, 500);
   const targets: { unitId: string; fields: Record<string, FieldValue>; fresh: boolean }[] = [];
+  const promote: string[] = [];
   for (const e of eps) {
     let hit = unitOf.get(e.episodeNumber);
     let fresh = false;
+    if (hit?.status === "provisional") promote.push(hit.id);
     if (!hit && !create) continue;
     if (!hit) {
       const row = await exec
@@ -343,6 +355,7 @@ export async function applySeasonMetadata(
           season_number: sd.seasonNumber,
           episode_number: e.episodeNumber,
           status: "confirmed",
+          source: "tmdb",
           display_order: e.episodeNumber,
         })
         .returning(["id"])
@@ -361,6 +374,7 @@ export async function applySeasonMetadata(
     ...(sd.posterPath ? { posterPath: { value: sd.posterPath, language: null } } : {}),
   };
   targets.push({ unitId: seasonUnit, fields: seasonFields, fresh: false });
+  out.promoted = await promoteUnits(exec, promote);
   const ids = targets.map((t) => t.unitId);
   const existing = ids.length
     ? await exec
@@ -382,11 +396,11 @@ export async function applySeasonMetadata(
   return out;
 }
 
-/** 该作品已有集单元(或整季资产覆盖)的季号:默认刷新时这些季也拉每集信息 */
+/** 该作品已有集单元(或整季资产覆盖)的季号(含季 0):默认刷新时这些季也拉每集信息 */
 export async function seasonsInLibrary(exec: Exec, entityId: string): Promise<number[]> {
   const r = await sql<{ season_number: number }>`
     select distinct season_number from content_unit
-     where media_entity_id = ${entityId} and kind = 'episode' and status <> 'retired' and season_number >= 1
+     where media_entity_id = ${entityId} and kind = 'episode' and status <> 'retired' and season_number >= 0
      order by season_number`.execute(exec);
   return r.rows.map((x) => x.season_number);
 }
@@ -394,12 +408,14 @@ export async function seasonsInLibrary(exec: Exec, entityId: string): Promise<nu
 /**
  * 本次刷新要拉每集信息的季:full = 全部正片季(最多 60);否则 = 追踪中的季 ∪ 库里已有集单元的季,取最近 8 季
  * (没有追踪订阅、库里也没有集单元时不拉季详情,定时刷新不额外请求 TMDB)。
+ * specials = true(设置 metadata.includeSpecials,缺省开)且 TMDB 有季 0 时,季 0(特别篇)算在内(对照 MP)。
  */
 export function seasonsToRefresh(
   known: readonly number[],
-  opts: { full: boolean; tracked: readonly number[]; inLibrary: readonly number[] },
+  opts: { full: boolean; tracked: readonly number[]; inLibrary: readonly number[]; specials?: boolean },
 ): number[] {
-  const all = [...new Set(known.filter((n) => n >= 1))].sort((a, b) => a - b);
+  const min = opts.specials ? 0 : 1;
+  const all = [...new Set(known.filter((n) => n >= min))].sort((a, b) => a - b);
   if (opts.full) return all.slice(0, 60);
   const want = new Set<number>([...opts.tracked, ...opts.inLibrary].filter((n) => all.includes(n)));
   return [...want].sort((a, b) => a - b).slice(-8);

apps/server/src/modules/sources-identity/service/metadata-totals.ts

@@ -14,27 +14,52 @@ export interface SeasonTotalsResult {
   activated: number;
 }
 
-/** 季单元:按(作品, 季号)找最早建的未退役季单元,没有就建(confirmed) */
+/**
+ * 季单元:按(作品, 季号)找最早建的未退役季单元,没有就建(confirmed,来源 tmdb)。
+ * 调用方拿的是 TMDB 季结构里的季:已有的临时季单元就此转正式(DH-1,保留 id 与引用)。
+ */
 export async function ensureSeason(exec: Exec, entityId: string, season: number, name: string | null): Promise<string> {
   const hit = await exec
     .selectFrom("content_unit")
-    .select("id")
+    .select(["id", "status"])
     .where("media_entity_id", "=", entityId)
     .where("kind", "=", "season")
     .where("season_number", "=", season)
     .where("status", "!=", "retired")
     .orderBy("created_at")
     .executeTakeFirst();
-  if (hit) return hit.id;
+  if (hit) {
+    if (hit.status === "provisional") await promoteUnits(exec, [hit.id]);
+    return hit.id;
+  }
   return (
     await exec
       .insertInto("content_unit")
-      .values({ media_entity_id: entityId, kind: "season", season_number: season, status: "confirmed", title: name })
+      .values({
+        media_entity_id: entityId,
+        kind: "season",
+        season_number: season,
+        status: "confirmed",
+        source: "tmdb",
+        title: name,
+      })
       .returning("id")
       .executeTakeFirstOrThrow()
   ).id;
 }
 
+/** 临时单元转正式(DH-1):与 TMDB 季集结构对上的单元标 confirmed、来源 tmdb;id 不变,引用不动。返回转正数 */
+export async function promoteUnits(exec: Exec, ids: readonly string[]): Promise<number> {
+  if (!ids.length) return 0;
+  const r = await exec
+    .updateTable("content_unit")
+    .set({ status: "confirmed", source: "tmdb" })
+    .where("id", "in", [...ids])
+    .where("status", "=", "provisional")
+    .executeTakeFirst();
+  return Number(r.numUpdatedRows ?? 0);
+}
+
 export async function applySeasonTotals(
   exec: Exec,
   entityId: string,
@@ -127,7 +152,7 @@ export async function applyEpisodeAirDates(exec: Exec, entityId: string, detail:
   const seasonUnit = await ensureSeason(exec, entityId, season, detail.name);
   const units = await exec
     .selectFrom("content_unit")
-    .select(["id", "episode_number", sql<string | null>`to_char(air_date, 'YYYY-MM-DD')`.as("air"), "title"])
+    .select(["id", "episode_number", "status", sql<string | null>`to_char(air_date, 'YYYY-MM-DD')`.as("air"), "title"])
     .where("media_entity_id", "=", entityId)
     .where("kind", "=", "episode")
     .where("season_number", "=", season)
@@ -140,6 +165,7 @@ export async function applyEpisodeAirDates(exec: Exec, entityId: string, detail:
     const air = e.airDate && /^\d{4}-\d{2}-\d{2}/.test(e.airDate) ? e.airDate.slice(0, 10) : null;
     const hit = units.find((u) => u.episode_number === e.episodeNumber);
     if (hit) {
+      if (hit.status === "provisional") await promoteUnits(exec, [hit.id]);
       const patch = {
         ...(air && air !== hit.air ? { air_date: air } : {}),
         ...(!hit.title && e.name ? { title: e.name } : {}),
@@ -159,6 +185,7 @@ export async function applyEpisodeAirDates(exec: Exec, entityId: string, detail:
         season_number: season,
         episode_number: e.episodeNumber,
         status: "confirmed",
+        source: "tmdb",
         display_order: e.episodeNumber,
         title: e.name,
         air_date: air,

apps/server/src/modules/sources-identity/service/metadata.ts

@@ -23,6 +23,7 @@ import { type Actor, audit, type Exec } from "./common.ts";
 import { resolveEntityId, seedAliases } from "./entities.ts";
 import {
   applySeasonMetadata,
+  metadataIncludeSpecials,
   metadataLanguages,
   type SeasonMetadataResult,
   seasonsInLibrary,
@@ -36,6 +37,7 @@ import {
   type SeasonTotalsResult,
   trackedSeasonsForAirDates,
 } from "./metadata-totals.ts";
+import { mergeDuplicateUnits } from "./unit-merge.ts";
 
 // refreshMetadata:字段级刷新(REQ §7.3)。来源值写 metadata_field;最终值 = 覆盖(锁)> 已锁字段保持 > 来源值。
 // 只更新 media_entity 的展示字段与别名,不触碰 path_binding(路径绑定不随元数据变化)。
@@ -125,10 +127,17 @@ export interface RefreshResult {
   episodes?: SeasonMetadataResult[];
   /** MD-1:回退链取到的标题 / 中文名(语言) */
   titles?: { title: string; language: string | null; titleZh: string | null };
+  /** DH-1:本次合并掉的重复内容单元数 */
+  mergedUnits?: number;
+  /** DH-1:本次临时转正式的集单元数 */
+  promotedUnits?: number;
 }
 
 export interface RefreshOptions {
-  /** true = 拉全部正片季的每集信息(POST /media-entities/{id}/refresh?full=1);默认只拉追踪中 / 库里已有 / 最新的季 */
+  /**
+   * true = 拉全部季的每集信息(POST /media-entities/{id}/refresh?full=1;设置 metadata.includeSpecials 开时含季 0 特别篇);
+   * 默认只拉追踪中 / 库里已有的季
+   */
   full?: boolean;
   /** 指定 TMDB 客户端(批量回刷用限速客户端);缺省 rt.tmdb() */
   tmdb?: TmdbClient;
@@ -199,6 +208,7 @@ export async function refreshMetadata(
       full: !!opts.full,
       tracked,
       inLibrary: await seasonsInLibrary(ctx.db, id),
+      specials: await metadataIncludeSpecials(ctx),
     });
     creatable = new Set(opts.full ? want : tracked);
     for (const n of want) {
@@ -240,6 +250,10 @@ export async function refreshMetadata(
     await seedAliases(trx, id, detail);
     if (richDetail) await tagTitleAliases(trx, id, richDetail);
     const r = await project(trx, id, now);
+    // DH-1:先合并重复单元(并发懒建留下的同季同集多条),再按 TMDB 季集结构写入 / 转正
+    const merged = await mergeDuplicateUnits(trx, id);
+    if (merged.merged)
+      ctx.logger.info({ entityId: id, merged: merged.merged, repointed: merged.repointed }, "合并重复内容单元");
     // 季总集数账本:只升;回退不低于已覆盖最高集;锁定不改(订阅完成判定据此)
     const totals =
       ident.entity_type === "tv" && detail.seasons?.length
@@ -250,9 +264,12 @@ export async function refreshMetadata(
     for (const sd of richSeasons)
       episodes.push(await applySeasonMetadata(trx, id, sd, rev, now, { create: creatable.has(sd.seasonNumber) }));
     await trx.updateTable("media_entity").set({ last_metadata_refresh_at: now }).where("id", "=", id).execute();
+    const promoted = episodes.reduce((n, x) => n + x.promoted, 0);
     return {
       status: "refreshed" as const,
       ...r,
+      ...(merged.merged ? { mergedUnits: merged.merged } : {}),
+      ...(promoted ? { promotedUnits: promoted } : {}),
       ...(totals ? { seasons: totals.seasons } : {}),
       ...(richSeasons.length ? { episodes } : {}),
       ...(richDetail

apps/server/src/modules/sources-identity/service/unit-merge.ts

@@ -0,0 +1,238 @@
+import { type DB, type Kysely, sql, type Transaction } from "@umt/db";
+
+// DH-1 重复内容单元合并:同作品、同类型、同季同集号、同内容类型与分段的多条未退役单元(并发懒建留下的重复)合并为一条。
+// 保留被引用最多的那条(同数时已确认优先,再取最早建的);其余单元的所有引用(资源覆盖、资产覆盖、下载计划、订阅覆盖、
+// 字段来源 / 覆盖、子单元…,按外键自动发现)改指向保留的那条,改指向会撞唯一约束的行(保留的单元上已有同一行)删除,
+// 然后删掉重复单元。保留的单元:组内有已确认的即转已确认;标题 / 播出日期 / 总集数 / 元数据为空时取重复单元的值。
+// 调用方负责事务(refreshMetadata 的事务,或 scripts/identity-cleanup.ts --merge-duplicates --apply 的事务)。
+// 本文件只依赖 @umt/db,脚本直接引用。
+
+type Exec = Kysely<DB> | Transaction<DB>;
+
+export interface MergeUnit {
+  id: string;
+  status: string;
+  source: string | null;
+  refs: number;
+  createdAt: Date;
+}
+
+export interface UnitMergeGroup {
+  mediaEntityId: string;
+  kind: string;
+  seasonNumber: number | null;
+  episodeNumber: number | null;
+  contentType: string;
+  segment: string;
+  keep: MergeUnit;
+  drop: MergeUnit[];
+}
+
+export interface UnitMergeResult {
+  groups: number;
+  /** 删除的重复单元数 */
+  merged: number;
+  /** 改指向保留单元的引用行数 */
+  repointed: number;
+  /** 撞唯一约束而删除的引用行数 */
+  droppedRows: number;
+}
+
+interface FkRef {
+  table: string;
+  column: string;
+}
+
+/** 只是单元自身的元数据,不算"被引用" */
+const META_TABLES = new Set(["metadata_field", "field_override", "title_alias"]);
+
+const quoteIdent = (s: string) =>
+  s
+    .split(".")
+    .map((p) => `"${p.replace(/"/g, '""')}"`)
+    .join(".");
+const bare = (t: string) => t.replace(/^.*\./, "").replace(/"/g, "");
+
+/** 引用 content_unit 的全部外键列(含自引用 parent_unit_id) */
+export async function contentUnitRefs(exec: Exec): Promise<FkRef[]> {
+  const r = await sql<FkRef>`
+    select c.conrelid::regclass::text as table, a.attname as column
+      from pg_constraint c
+      join pg_attribute a on a.attrelid = c.conrelid and a.attnum = any(c.conkey)
+     where c.contype = 'f' and c.confrelid = 'content_unit'::regclass
+     order by 1, 2`.execute(exec);
+  return r.rows;
+}
+
+/** 各单元被"使用"的引用数(不含单元自身的元数据表与子单元) */
+async function refCounts(exec: Exec, fks: readonly FkRef[], ids: readonly string[]): Promise<Map<string, number>> {
+  const out = new Map<string, number>(ids.map((id) => [id, 0]));
+  if (!ids.length) return out;
+  for (const f of fks) {
+    if (META_TABLES.has(bare(f.table)) || bare(f.table) === "content_unit") continue;
+    const rows = await sql<{ id: string; n: number }>`
+      select ${sql.raw(quoteIdent(f.column))} as id, count(*)::int as n from ${sql.raw(quoteIdent(f.table))}
+       where ${sql.raw(quoteIdent(f.column))} = any(${[...ids]}::uuid[]) group by 1`.execute(exec);
+    for (const r of rows.rows) out.set(r.id, (out.get(r.id) ?? 0) + r.n);
+  }
+  return out;
+}
+
+/** 找出重复单元组并选出保留的那条;只读 */
+export async function planUnitMerges(exec: Exec, opts: { mediaEntityId?: string } = {}): Promise<UnitMergeGroup[]> {
+  const only = opts.mediaEntityId ?? null;
+  const rows = await sql<{
+    id: string;
+    media_entity_id: string;
+    kind: string;
+    season_number: number | null;
+    episode_number: number | null;
+    content_type: string;
+    segment: string;
+    status: string;
+    source: string | null;
+    created_at: Date;
+  }>`
+    select u.id, u.media_entity_id, u.kind, u.season_number, u.episode_number, u.content_type, u.segment,
+           u.status, to_jsonb(u)->>'source' as source, u.created_at
+      from content_unit u
+      join (select media_entity_id, kind, season_number, episode_number, content_type, segment
+              from content_unit
+             where status <> 'retired' and (${only}::uuid is null or media_entity_id = ${only}::uuid)
+             group by 1, 2, 3, 4, 5, 6 having count(*) > 1) g
+        on g.media_entity_id = u.media_entity_id and g.kind = u.kind
+       and g.season_number is not distinct from u.season_number
+       and g.episode_number is not distinct from u.episode_number
+       and g.content_type = u.content_type and g.segment = u.segment
+     where u.status <> 'retired'
+     order by u.media_entity_id, u.kind desc, u.season_number, u.episode_number, u.created_at, u.id`.execute(exec);
+  if (!rows.rows.length) return [];
+  const refs = await refCounts(
+    exec,
+    await contentUnitRefs(exec),
+    rows.rows.map((r) => r.id),
+  );
+  const groups = new Map<string, typeof rows.rows>();
+  for (const u of rows.rows) {
+    const k = [u.media_entity_id, u.kind, u.season_number, u.episode_number, u.content_type, u.segment].join("|");
+    groups.set(k, [...(groups.get(k) ?? []), u]);
+  }
+  const out: UnitMergeGroup[] = [];
+  for (const list of groups.values()) {
+    const units = list.map((u) => ({
+      id: u.id,
+      status: u.status,
+      source: u.source,
+      refs: refs.get(u.id) ?? 0,
+      createdAt: new Date(u.created_at),
+    }));
+    units.sort(
+      (a, b) =>
+        b.refs - a.refs ||
+        Number(b.status === "confirmed") - Number(a.status === "confirmed") ||
+        a.createdAt.getTime() - b.createdAt.getTime() ||
+        a.id.localeCompare(b.id),
+    );
+    const f = list[0]!;
+    out.push({
+      mediaEntityId: f.media_entity_id,
+      kind: f.kind,
+      seasonNumber: f.season_number,
+      episodeNumber: f.episode_number,
+      contentType: f.content_type,
+      segment: f.segment,
+      keep: units[0]!,
+      drop: units.slice(1),
+    });
+  }
+  return out;
+}
+
+/** 执行合并;必须在事务里调用(撞唯一约束时用保存点回退) */
+export async function applyUnitMerges(exec: Exec, groups: readonly UnitMergeGroup[]): Promise<UnitMergeResult> {
+  const res: UnitMergeResult = { groups: groups.length, merged: 0, repointed: 0, droppedRows: 0 };
+  if (!groups.length) return res;
+  const fks = await contentUnitRefs(exec);
+  // 脚本也可能对未迁移到 0053 的库执行:没有 source 列时不写
+  const hasSource = (
+    await sql<{ ok: boolean }>`
+      select exists (select 1 from pg_attribute where attrelid = 'content_unit'::regclass
+                      and attname = 'source' and not attisdropped) as ok`.execute(exec)
+  ).rows[0]?.ok;
+  const sourceSet = hasSource
+    ? sql`, source = case when k.source is null or (k.status <> 'confirmed' and d.status = 'confirmed')
+                         then coalesce(d.source, k.source) else k.source end`
+    : sql``;
+  for (const g of groups) {
+    const keep = g.keep.id;
+    for (const d of g.drop) {
+      // 保留单元的空列取重复单元的值(不覆盖已有值)
+      await sql`
+        update content_unit k set
+          title = coalesce(k.title, d.title),
+          air_date = coalesce(k.air_date, d.air_date),
+          episode_total = coalesce(k.episode_total, d.episode_total),
+          episode_total_source = case when k.episode_total is null then d.episode_total_source else k.episode_total_source end,
+          episode_total_at = case when k.episode_total is null then d.episode_total_at else k.episode_total_at end,
+          metadata = case when k.metadata = '{}'::jsonb then d.metadata else k.metadata end,
+          field_locks = (select array(select distinct x from unnest(k.field_locks || d.field_locks) x order by 1)),
+          numbering_locked = k.numbering_locked or d.numbering_locked,
+          status = case when k.status = 'confirmed' or d.status = 'confirmed' then 'confirmed' else k.status end
+          ${sourceSet}
+          from content_unit d
+         where k.id = ${keep}::uuid and d.id = ${d.id}::uuid`.execute(exec);
+      for (const f of fks) {
+        const t = sql.raw(quoteIdent(f.table));
+        const c = sql.raw(quoteIdent(f.column));
+        await sql`savepoint unit_merge`.execute(exec);
+        try {
+          const r = await sql`update ${t} set ${c} = ${keep}::uuid where ${c} = ${d.id}::uuid`.execute(exec);
+          await sql`release savepoint unit_merge`.execute(exec);
+          res.repointed += Number(r.numAffectedRows ?? 0);
+          continue;
+        } catch (e) {
+          await sql`rollback to savepoint unit_merge`.execute(exec);
+          if ((e as { code?: string }).code !== "23505") throw e;
+        }
+        // 整表改指向撞了唯一约束:逐行改,撞的那行删掉(保留的单元上已有同一行)
+        const rows = await sql<{
+          ctid: string;
+        }>`select ctid::text as ctid from ${t} where ${c} = ${d.id}::uuid`.execute(exec);
+        for (const row of rows.rows) {
+          await sql`savepoint unit_merge_row`.execute(exec);
+          try {
+            await sql`update ${t} set ${c} = ${keep}::uuid where ctid = ${row.ctid}::tid`.execute(exec);
+            await sql`release savepoint unit_merge_row`.execute(exec);
+            res.repointed++;
+          } catch (e) {
+            await sql`rollback to savepoint unit_merge_row`.execute(exec);
+            if ((e as { code?: string }).code !== "23505") throw e;
+            await sql`delete from ${t} where ctid = ${row.ctid}::tid`.execute(exec);
+            res.droppedRows++;
+          }
+        }
+      }
+      await sql`delete from content_unit where id = ${d.id}::uuid`.execute(exec);
+      res.merged++;
+    }
+  }
+  return res;
+}
+
+/** 规划 + 执行(refreshMetadata 的一步) */
+export async function mergeDuplicateUnits(exec: Exec, mediaEntityId: string): Promise<UnitMergeResult> {
+  return applyUnitMerges(exec, await planUnitMerges(exec, { mediaEntityId }));
+}
+
+/** 串行化同一(作品, 类型, 季, 集)的懒建:事务级咨询锁,防止并发识别建出重复单元 */
+export async function lockUnitKey(
+  exec: Exec,
+  mediaEntityId: string,
+  kind: string,
+  season: number | null,
+  episode: number | null,
+): Promise<void> {
+  await sql`select pg_advisory_xact_lock(hashtextextended(${`content_unit:${mediaEntityId}:${kind}:${season ?? ""}:${episode ?? ""}`}, 0))`.execute(
+    exec,
+  );
+}