From 044e718ec6c66629de00ca6a5d546d65faf85742 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 29 Sep 2026 20:49:56 +0800 Subject: [PATCH] =?UTF-8?q?feat(identity):=20content=20unit=20data=20hygie?= =?UTF-8?q?ne=20=E2=80=94=20specials,=20provisional=20promotion,=20duplica?= =?UTF-8?q?te=20merge=20(DH-1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- .../acquisition-organize/services/organize.ts | 1 + .../sources-identity/routes/identity.ts | 6 +- .../sources-identity/service/entities.ts | 4 + .../sources-identity/service/metadata-rich.ts | 36 ++- .../service/metadata-totals.ts | 37 ++- .../sources-identity/service/metadata.ts | 19 +- .../sources-identity/service/unit-merge.ts | 238 +++++++++++++++ apps/server/test/e2e/e7-tv-series.test.ts | 3 +- .../sources-identity/data-hygiene-dh1.test.ts | 273 ++++++++++++++++++ .../sources-identity/metadata-md1.test.ts | 10 +- .../subscriptions-c1a-sources.test.ts | 3 +- apps/web/app/components/settings/metadata.tsx | 9 + docs/plan/00-first-version-plan.md | 1 + docs/runbook/identity.md | 21 +- docs/runbook/metadata.md | 17 +- packages/contracts/src/content.ts | 5 + packages/contracts/src/settings.ts | 13 +- packages/contracts/test/db-sync.test.ts | 2 + .../db/migrations/0053_content_unit_source.ts | 24 ++ packages/db/src/types.ts | 2 + scripts/identity-cleanup.ts | 121 +++----- 21 files changed, 734 insertions(+), 111 deletions(-) create mode 100644 apps/server/src/modules/sources-identity/service/unit-merge.ts create mode 100644 apps/server/test/sources-identity/data-hygiene-dh1.test.ts create mode 100644 packages/db/migrations/0053_content_unit_source.ts diff --git a/apps/server/src/modules/acquisition-organize/services/organize.ts b/apps/server/src/modules/acquisition-organize/services/organize.ts index 0f8483df..9221222a 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize.ts +++ b/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") diff --git a/apps/server/src/modules/sources-identity/routes/identity.ts b/apps/server/src/modules/sources-identity/routes/identity.ts index 7fca1f93..6d6c4292 100644 --- a/apps/server/src/modules/sources-identity/routes/identity.ts +++ b/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)[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) diff --git a/apps/server/src/modules/sources-identity/service/entities.ts b/apps/server/src/modules/sources-identity/service/entities.ts index c6126e6e..eab8b86a 100644 --- a/apps/server/src/modules/sources-identity/service/entities.ts +++ b/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 { + // 并发识别同一集时串行化,避免建出重复单元(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"]) diff --git a/apps/server/src/modules/sources-identity/service/metadata-rich.ts b/apps/server/src/modules/sources-identity/service/metadata-rich.ts index c7dc4286..a5b402ff 100644 --- a/apps/server/src/modules/sources-identity/service/metadata-rich.ts +++ b/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 { return langs?.length ? langs : [...DEFAULT_METADATA_LANGUAGES]; } +/** 设置 metadata.includeSpecials:full 刷新 / 回刷是否包含季 0(特别篇);缺省 true */ +export async function metadataIncludeSpecials(ctx: ServerContext): Promise { + const m = await ctx.settings.get("metadata").catch(() => null); + return (m as { includeSpecials?: boolean } | null)?.includeSpecials !== false; +} + /** 富详情 → 作品字段(值 + 语言);海报 / 背景图没有时不写(保留旧值) */ export function richEntityFields(r: TmdbRichDetail): Partial> { 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 { 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(); + const unitOf = new Map(); 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; 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 { 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 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([...opts.tracked, ...opts.inLibrary].filter((n) => all.includes(n))); return [...want].sort((a, b) => a - b).slice(-8); diff --git a/apps/server/src/modules/sources-identity/service/metadata-totals.ts b/apps/server/src/modules/sources-identity/service/metadata-totals.ts index f76f93f5..8e72f55b 100644 --- a/apps/server/src/modules/sources-identity/service/metadata-totals.ts +++ b/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 { 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 { + 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`to_char(air_date, 'YYYY-MM-DD')`.as("air"), "title"]) + .select(["id", "episode_number", "status", sql`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, diff --git a/apps/server/src/modules/sources-identity/service/metadata.ts b/apps/server/src/modules/sources-identity/service/metadata.ts index d9650715..5a5835fe 100644 --- a/apps/server/src/modules/sources-identity/service/metadata.ts +++ b/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 diff --git a/apps/server/src/modules/sources-identity/service/unit-merge.ts b/apps/server/src/modules/sources-identity/service/unit-merge.ts new file mode 100644 index 00000000..bba6a36a --- /dev/null +++ b/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 | Transaction; + +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 { + const r = await sql` + 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> { + const out = new Map(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 { + 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(); + 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 { + 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 { + return applyUnitMerges(exec, await planUnitMerges(exec, { mediaEntityId })); +} + +/** 串行化同一(作品, 类型, 季, 集)的懒建:事务级咨询锁,防止并发识别建出重复单元 */ +export async function lockUnitKey( + exec: Exec, + mediaEntityId: string, + kind: string, + season: number | null, + episode: number | null, +): Promise { + await sql`select pg_advisory_xact_lock(hashtextextended(${`content_unit:${mediaEntityId}:${kind}:${season ?? ""}:${episode ?? ""}`}, 0))`.execute( + exec, + ); +} diff --git a/apps/server/test/e2e/e7-tv-series.test.ts b/apps/server/test/e2e/e7-tv-series.test.ts index dd6e6581..f1f97b20 100644 --- a/apps/server/test/e2e/e7-tv-series.test.ts +++ b/apps/server/test/e2e/e7-tv-series.test.ts @@ -296,8 +296,9 @@ describe.skipIf(!hasPg)("E7 常规剧集纵切", () => { .where("season_number", "=", 1) .orderBy("episode_number") .execute(); + // DH-1:元数据刷新拉了 S1 季详情,对上 TMDB 的 E01 转正式;E02 不在 mock TMDB 的 S1 里,保持临时 expect(eps).toEqual([ - { episode_number: 1, status: "provisional" }, + { episode_number: 1, status: "confirmed" }, { episode_number: 2, status: "provisional" }, ]); // 第三集没出现、总集数未知:Slot 不满足,订阅继续追踪 diff --git a/apps/server/test/sources-identity/data-hygiene-dh1.test.ts b/apps/server/test/sources-identity/data-hygiene-dh1.test.ts new file mode 100644 index 00000000..8a833f4b --- /dev/null +++ b/apps/server/test/sources-identity/data-hygiene-dh1.test.ts @@ -0,0 +1,273 @@ +import { execFileSync } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import type { ParsedRelease } from "@umt/domain"; +import { createMockTmdb, type MockTmdb } from "@umt/testkit"; +import { sql } from "kysely"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { backfillLanguages } from "../../src/modules/sources-identity/service/backfill-languages.ts"; +import { ensureUnits } from "../../src/modules/sources-identity/service/entities.ts"; +import { refreshMetadata } from "../../src/modules/sources-identity/service/metadata.ts"; +import { seasonsToRefresh } from "../../src/modules/sources-identity/service/metadata-rich.ts"; +import { createHarness, type Harness, PG_URL, pgAvailable } from "./harness.ts"; + +// DH-1 内容单元数据治理:季 0(特别篇)、临时转正式、重复单元合并(refreshMetadata 一步 + identity-cleanup --merge-duplicates)。 + +const up = await pgAvailable(); +const ROOT = fileURLToPath(new URL("../../../../", import.meta.url)); +const SCRIPT = fileURLToPath(new URL("../../../../scripts/identity-cleanup.ts", import.meta.url)); + +describe("seasonsToRefresh 季 0", () => { + it("specials 开且 TMDB 有季 0:full 含季 0;库里有季 0 集单元时默认刷新也含;关时不含", () => { + expect(seasonsToRefresh([0, 1, 2], { full: true, tracked: [], inLibrary: [], specials: true })).toEqual([0, 1, 2]); + expect(seasonsToRefresh([0, 1, 2], { full: true, tracked: [], inLibrary: [], specials: false })).toEqual([1, 2]); + expect(seasonsToRefresh([1, 2], { full: true, tracked: [], inLibrary: [], specials: true })).toEqual([1, 2]); + expect(seasonsToRefresh([0, 1, 2], { full: false, tracked: [2], inLibrary: [0], specials: true })).toEqual([0, 2]); + expect(seasonsToRefresh([0, 1, 2], { full: false, tracked: [2], inLibrary: [0], specials: false })).toEqual([2]); + }); +}); + +describe.skipIf(!up)("DH-1 内容单元数据治理", () => { + let h: Harness; + let tmdb: MockTmdb; + let id: string; + let siteId: string; + const ids: Record = {}; + + const unit = async ( + kind: "season" | "episode", + season: number, + episode: number | null, + extra: Record = {}, + ) => + ( + await h.db + .insertInto("content_unit") + .values({ + media_entity_id: id, + kind, + season_number: season, + episode_number: episode, + status: "provisional", + source: "release", + ...extra, + } as never) + .returning("id") + .executeTakeFirstOrThrow() + ).id; + const release = async (key: string, unitId: string) => + h.db + .insertInto("source_release") + .values({ site_id: siteId, site_release_id: key, title: key, media_entity_id: id, content_unit_id: unitId }) + .execute(); + const asset = async () => + ( + await h.db + .insertInto("media_asset") + .values({ media_entity_id: id, kind: "video" }) + .returning("id") + .executeTakeFirstOrThrow() + ).id; + const cover = (assetId: string, unitId: string) => + h.db.insertInto("asset_unit_coverage").values({ media_asset_id: assetId, content_unit_id: unitId }).execute(); + const units = () => + h.db + .selectFrom("content_unit") + .select(["id", "kind", "season_number", "episode_number", "status", "source", "title"]) + .where("media_entity_id", "=", id) + .orderBy("season_number") + .orderBy("kind", "desc") + .orderBy("episode_number") + .execute(); + const label = (u: { kind: string; season_number: number | null; episode_number: number | null }) => + u.kind === "season" ? `S${u.season_number}` : `S${u.season_number}E${u.episode_number}`; + + beforeAll(async () => { + tmdb = await createMockTmdb(); + h = await createHarness("dh1", { tmdbUrl: tmdb.url, now: () => new Date("2026-01-01T00:00:00Z") }); + siteId = ( + await h.db + .insertInto("site") + .values({ key: "dh1", name: "DH1", adapter: "nexusphp", base_url: "https://dh1.example/" } as never) + .returning("id") + .executeTakeFirstOrThrow() + ).id; + // 《向往的生活》(mock TMDB 900001:季 0 两集特别篇、第 1 季 12 集、第 2 季 3 集) + id = ( + await h.db + .insertInto("media_entity") + .values({ kind: "variety", title: "向往的生活", year: 2017, status: "active" }) + .returning("id") + .executeTakeFirstOrThrow() + ).id; + await h.db + .insertInto("external_identity") + .values({ media_entity_id: id, provider: "tmdb", entity_type: "tv", external_id: "900001", status: "accepted" }) + .execute(); + // 资源解析懒建的临时单元:S01 / S01E01 在 TMDB 结构内;S01E13、S03、S03E01 对不上 + ids.s1 = await unit("season", 1, null); + ids.e1 = await unit("episode", 1, 1, { parent_unit_id: ids.s1 }); + ids.e13 = await unit("episode", 1, 13, { parent_unit_id: ids.s1 }); + ids.s3 = await unit("season", 3, null); + ids.s3e1 = await unit("episode", 3, 1, { parent_unit_id: ids.s3 }); + // S01E02 并发懒建的三条重复:e2a 最早、无引用、带一条字段来源;e2b 引用 2;e2c 引用 4(保留) + ids.e2a = await unit("episode", 1, 2, { parent_unit_id: ids.s1 }); + ids.e2b = await unit("episode", 1, 2, { parent_unit_id: ids.s1 }); + ids.e2c = await unit("episode", 1, 2, { parent_unit_id: ids.s1 }); + await release("r1", ids.e2b); + await release("r2", ids.e2c); + await release("r3", ids.e2c); + const a = await asset(); + const b = await asset(); + await cover(a, ids.e2b); + await cover(a, ids.e2c); + await cover(b, ids.e2c); + for (const u of [ids.e2a, ids.e2c]) + await h.db + .insertInto("metadata_field") + .values({ content_unit_id: u, field: "runtime", value: JSON.stringify(1), source: "user" } as never) + .execute(); + }); + afterAll(async () => { + await h?.close(); + await tmdb?.close(); + }); + + it("full 刷新:合并重复单元,对上 TMDB 的临时单元转正式,季 0 特别篇建单元;对不上的保持临时", async () => { + const r = await refreshMetadata(h.rt, id, { full: true }); + expect(r.status).toBe("refreshed"); + expect(r.mergedUnits).toBe(2); + expect(r.episodes?.map((x) => x.season)).toEqual([0, 1, 2]); + + const list = await units(); + const byLabel = new Map(list.map((u) => [label(u), u])); + // 重复只剩一条,保留被引用最多的 e2c(不是最早的 e2a) + expect(list.filter((u) => label(u) === "S1E2")).toHaveLength(1); + expect(byLabel.get("S1E2")!.id).toBe(ids.e2c); + for (const k of ["e2a", "e2b"]) + expect(await h.db.selectFrom("content_unit").select("id").where("id", "=", ids[k]!).execute()).toEqual([]); + // 引用全部改指向保留的单元;同一资产的覆盖撞唯一约束的那行删掉 + const rels = await h.db + .selectFrom("source_release") + .select(["site_release_id", "content_unit_id"]) + .where("site_id", "=", siteId) + .orderBy("site_release_id") + .execute(); + expect(rels.map((x) => x.content_unit_id)).toEqual([ids.e2c, ids.e2c, ids.e2c]); + const cov = await h.db + .selectFrom("asset_unit_coverage") + .select("content_unit_id") + .where("content_unit_id", "in", [ids.e2a!, ids.e2b!, ids.e2c!]) + .execute(); + expect(cov.map((x) => x.content_unit_id)).toEqual([ids.e2c, ids.e2c]); + const rt = await h.db + .selectFrom("metadata_field") + .select("content_unit_id") + .where("field", "=", "runtime") + .where("source", "=", "user") + .execute(); + expect(rt).toEqual([{ content_unit_id: ids.e2c }]); + + // 临时转正式:保留原 id,来源 tmdb + for (const [k, l] of [ + ["s1", "S1"], + ["e1", "S1E1"], + ["e2c", "S1E2"], + ] as const) + expect(byLabel.get(l)).toMatchObject({ id: ids[k], status: "confirmed", source: "tmdb" }); + expect(byLabel.get("S1E1")!.title).toBe("第 1 期"); + // 对不上的保持临时 + for (const l of ["S1E13", "S3", "S3E1"]) expect(byLabel.get(l)).toMatchObject({ status: "provisional" }); + // 季 0:特别篇季与两集按 TMDB 建(confirmed、来源 tmdb) + expect(byLabel.get("S0")).toMatchObject({ status: "confirmed", source: "tmdb" }); + expect(list.filter((u) => u.season_number === 0 && u.kind === "episode").map((u) => [u.title, u.status])).toEqual([ + ["先导片 1", "confirmed"], + ["先导片 2", "confirmed"], + ]); + // 季 0 不进正片季总集数账本 + const s0 = await h.db + .selectFrom("content_unit") + .select(["episode_total"]) + .where("media_entity_id", "=", id) + .where("kind", "=", "season") + .where("season_number", "=", 0) + .executeTakeFirstOrThrow(); + expect(s0.episode_total).toBeNull(); + }); + + it("设置 metadata.includeSpecials = false:full 刷新不拉季 0;回刷任务按设置覆盖季 0", async () => { + const dropS0 = () => + sql`delete from content_unit where media_entity_id = ${id} and season_number = 0 and kind = 'episode'` + .execute(h.db) + .then(() => sql`delete from content_unit where media_entity_id = ${id} and season_number = 0`.execute(h.db)); + await dropS0(); + await h.db + .insertInto("settings") + .values({ + key: "metadata", + value: JSON.stringify({ languages: ["zh-CN", "en-US"], includeSpecials: false }), + } as never) + .execute(); + const r = await refreshMetadata(h.rt, id, { full: true }); + expect(r.episodes?.map((x) => x.season)).toEqual([1, 2]); + expect((await units()).filter((u) => u.season_number === 0)).toEqual([]); + + await h.db.deleteFrom("settings").where("key", "=", "metadata").execute(); + const b = await backfillLanguages(h.rt, { minIntervalMs: 0, disableWhenDone: false }); + expect(b.refreshed).toBe(1); + expect((await units()).filter((u) => u.season_number === 0).map(label)).toEqual(["S0", "S0E1", "S0E2"]); + }); + + it("并发懒建同一集只建一条(咨询锁)", async () => { + const parsed = { season: 1, episodes: { start: 20, end: 20 } } as unknown as ParsedRelease; + // 事务 1 建了单元但还没提交时,事务 2 懒建同一集:没有锁时看不到未提交的行,会再建一条 + let created!: () => void; + const inserted = new Promise((r) => { + created = r; + }); + const t1 = h.db.transaction().execute(async (trx) => { + await ensureUnits(trx, id, "series", parsed); + created(); + await new Promise((r) => setTimeout(r, 300)); + }); + await inserted; + const t2 = h.db.transaction().execute((trx) => ensureUnits(trx, id, "series", parsed)); + await Promise.all([t1, t2]); + expect((await units()).filter((u) => label(u) === "S1E20")).toHaveLength(1); + }); + + it("identity-cleanup --merge-duplicates:默认 dry-run 列清单不改库,--apply 只合并重复、不删孤立单元", async () => { + const schema = (await sql<{ s: string }>`select current_schema() as s`.execute(h.db)).rows[0]!.s; + const u = new URL(PG_URL); + u.searchParams.set("options", `-c search_path=${schema},public`); + const run = (...args: string[]) => + execFileSync(Deno.execPath(), ["run", "-A", SCRIPT, "--merge-duplicates", ...args], { + cwd: ROOT, + env: { ...process.env, DATABASE_URL: u.toString(), NO_COLOR: "1" }, + encoding: "utf8", + }); + // 再造重复:S01E03 已有 TMDB 建的正式单元,再懒建两条临时的,第二条有资源引用 + const d1 = await unit("episode", 1, 3); + const d2 = await unit("episode", 1, 3); + await release("r4", d2); + const before = await units(); + + const out = run(); + expect(out).toContain("DRY-RUN"); + expect(out).toContain("## 重复的内容单元:1 组,删除 2 条"); + expect(out).toContain(`《向往的生活》S01E03:保留 ${d2}(provisional,引用 1)`); + expect(out).not.toContain("## 冲突别名"); + expect(out).not.toContain("## 孤立的临时内容单元"); + expect(await units()).toEqual(before); + + const applied = run("--apply"); + expect(applied).toContain("合并 2 个重复单元"); + const after = await units(); + expect(after.map((x) => x.id)).not.toContain(d1); + // 组内有正式单元:保留的那条转正式,标题取被合并单元的值 + expect(after.filter((x) => label(x) === "S1E3")).toEqual([ + expect.objectContaining({ id: d2, status: "confirmed", source: "tmdb", title: "第 3 期" }), + ]); + // 孤立的临时单元(S1E13 / S3)不动 + expect(after.filter((x) => ["S1E13", "S3", "S3E1"].includes(label(x)))).toHaveLength(3); + }); +}); diff --git a/apps/server/test/sources-identity/metadata-md1.test.ts b/apps/server/test/sources-identity/metadata-md1.test.ts index 92536ea0..220bf705 100644 --- a/apps/server/test/sources-identity/metadata-md1.test.ts +++ b/apps/server/test/sources-identity/metadata-md1.test.ts @@ -119,12 +119,12 @@ describe.skipIf(!up)("MD-1 语言回退链 + 每集元数据", () => { expect(m(e3!).overview?.value).toBeNull(); // 未播出:占位名,语言未知 expect(m(s2e1!).name).toMatchObject({ value: "第 1 集", language: null }); - // 编号 / 状态不被元数据刷新改动(provisional 仍是 provisional) + // 编号不被元数据刷新改动;与 TMDB 季集结构对上的临时单元转正式(DH-1,id 不变) expect(units.map((u) => [u.season_number, u.episode_number, u.status, u.numbering_locked])).toEqual([ - [1, 1, "provisional", false], - [1, 2, "provisional", false], - [1, 3, "provisional", false], - [2, 1, "provisional", false], + [1, 1, "confirmed", false], + [1, 2, "confirmed", false], + [1, 3, "confirmed", false], + [2, 1, "confirmed", false], ]); const s1 = await h.db .selectFrom("content_unit") diff --git a/apps/server/test/subscriptions/subscriptions-c1a-sources.test.ts b/apps/server/test/subscriptions/subscriptions-c1a-sources.test.ts index 97e447e6..e53f5d73 100644 --- a/apps/server/test/subscriptions/subscriptions-c1a-sources.test.ts +++ b/apps/server/test/subscriptions/subscriptions-c1a-sources.test.ts @@ -136,7 +136,8 @@ describe.skipIf(!up)("订阅搜索关键词与集播出日期", () => { expect(eps.filter((x) => x.season_number === 3)).toHaveLength(10); expect(eps.filter((x) => x.season_number === 1)).toHaveLength(0); const first = eps.find((x) => x.season_number === 2 && x.episode_number === 1)!; - expect(first).toMatchObject({ id: e1, status: "provisional", air: "2023-06-22" }); + // 已有的临时集单元:id 不变,补日期,对上 TMDB 季集结构后转正式(DH-1) + expect(first).toMatchObject({ id: e1, status: "confirmed", air: "2023-06-22" }); expect(eps.find((x) => x.season_number === 2 && x.episode_number === 2)).toMatchObject({ status: "confirmed", air: "2023-06-29", diff --git a/apps/web/app/components/settings/metadata.tsx b/apps/web/app/components/settings/metadata.tsx index f253d33d..dc0aedb5 100644 --- a/apps/web/app/components/settings/metadata.tsx +++ b/apps/web/app/components/settings/metadata.tsx @@ -54,16 +54,25 @@ const TMDB_CARD: CardSpec = { hint: "标题、简介和每集信息按顺序取第一个有值的语言；中文名取链中第一个中文值。逗号分隔，如 zh-CN, zh-SG", wide: true, }, + { + name: "metadata.includeSpecials", + label: "包含特别篇（季 0）", + kind: "switch", + hint: "全量刷新和元数据回刷时，TMDB 有季 0 就一并拉取每集信息。特别篇不计入正片季的总集数和订阅完成判定", + wide: true, + }, ], fromSettings: (s) => ({ "metadata.tmdbApiKey": "", "metadata.tmdbBaseUrl": s.metadata.tmdbBaseUrl ?? "", "metadata.languages": (s.metadata.languages ?? DEFAULT_LANGUAGES).join(", "), + "metadata.includeSpecials": s.metadata.includeSpecials !== false, }), toPatch: (v) => ({ metadata: { tmdbBaseUrl: str(v["metadata.tmdbBaseUrl"]) || null, languages: parseLanguages(str(v["metadata.languages"])), + includeSpecials: v["metadata.includeSpecials"] === true, ...(str(v["metadata.tmdbApiKey"]) ? { tmdbApiKey: str(v["metadata.tmdbApiKey"]) } : {}), }, }), diff --git a/docs/plan/00-first-version-plan.md b/docs/plan/00-first-version-plan.md index 61f4a163..14099d7b 100644 --- a/docs/plan/00-first-version-plan.md +++ b/docs/plan/00-first-version-plan.md @@ -173,3 +173,4 @@ NFO / 图片投影、身份巡检与挂载根因聚合、HTTP 失败自动切浏 - **浏览器会话重启自愈(BR-3,用户 2026-09-29 反馈"标签页不存在或已关闭")**:camofox 重启后旧 tab 404 → 服务端自动重建上下文、注入 Cookie、回到原 URL 并重试;`POST /browser/sessions/{id}/reconnect`;前端自动重连一次并提示;容器实例变化时全部会话标 needs_reconnect。 - **VER-3 真机核对(2026-09-29)**:通过——登录态注入、搜索两阶段、待处理 6、Agent 抽屉、快照字段/时区/手机折叠;问题——严重:RSS 标题「[电视剧]」前缀被解析为中文名导致 15 条误映射到《东北有个周东北》(ID-2 修:分类前缀、别名守卫、快照更新重解析、错映射撤销脚本);中:RSS 条目未规范化(分类/大小/标题)、详情页只抓到 MediaInfo/豆瓣/标签(HH-2 修);低:请求计数不持久、「12 Memories」误判 E12、会话页乱码字符、原站视图默认显示「Agent」。 - **数据治理后续(FIX-6 之后)**:全量回刷覆盖季 0(特别篇);TMDB 已确认的季集单元由临时转正式(沙丘:预言 仍为临时);同作品同季同集重复临时单元合并(征途 E09/E10 各 3 条);清理脚本修正后对用户库 dry-run 为 0 删除,未执行 apply。 + - DH-1 已做(分支 dh1):`metadata.includeSpecials`(缺省开)full 刷新 / 回刷含季 0;刷新时对上 TMDB 季集结构的临时单元转正式(`content_unit.source`,迁移 0053);重复单元合并(refresh 一步 + `identity-cleanup.ts --merge-duplicates`),懒建加咨询锁防竞态。用户库只读 dry-run:征途 S01E09/E10 各 4 条 → 各保留 1 条,共删 6 条,未 apply。见 runbook metadata.md / identity.md。 diff --git a/docs/runbook/identity.md b/docs/runbook/identity.md index 235769db..d935c74b 100644 --- a/docs/runbook/identity.md +++ b/docs/runbook/identity.md @@ -105,19 +105,28 @@ DATABASE_URL=postgres://umt:<密码>@$PGIP:5432/umt deno run -A scripts/identity ## 冲突别名与临时内容单元清理(FIX-5,`scripts/identity-cleanup.ts`) -守卫上线前登记的错误别名、重复 / 孤立的临时内容单元用这个脚本清理。缺省只读,`--apply` 在一个事务里执行并写审计 -`identity.cleanup`;`--dry-run` 显式只读;`--keep-orphans` 不删孤立单元;`--entity ` 只处理一部作品。连接方式同上。 +守卫上线前登记的错误别名、重复内容单元、孤立的临时内容单元用这个脚本清理。缺省只读,`--apply` 在一个事务里执行并写审计 +`identity.cleanup`;`--dry-run` 显式只读;`--keep-orphans` 不删孤立单元;`--entity ` 只处理一部作品; +`--merge-duplicates` 只做重复单元合并(不动别名和孤立单元,DH-1)。连接方式同上。 ```sh DATABASE_URL=postgres://umt:<密码>@$PGIP:5432/umt deno run -A scripts/identity-cleanup.ts # 看清单 DATABASE_URL=postgres://umt:<密码>@$PGIP:5432/umt deno run -A scripts/identity-cleanup.ts --apply # 确认后执行 +# 只合并重复单元;只读演练可在连接串加 ?options=-c%20default_transaction_read_only%3Don 强制只读会话 +DATABASE_URL=postgres://umt:<密码>@$PGIP:5432/umt deno run -A scripts/identity-cleanup.ts --merge-duplicates ``` - 冲突别名:只动 `source = site` 的作品级别名,且不是作品自己的名字(主标题 / 原标题 / TMDB 别名)。同一归一化别名还挂在别的作品上 (或是别的作品的标题)→ 删;作品有已接受外部身份、同一种文字下已有确认外部标题而它对不上 → 删(中文标题的拼音写法保留); 本作品下带这个名字的资源都解析出年份且都与作品年份差超过 1 年 → 删。 -- 重复的临时内容单元:同作品同类型同季同集号(未退役)有多条且有临时的,保留被引用最多的(同数时 confirmed 优先、再取最早的), - 其余的引用改指向保留的那条(撞唯一约束的元数据行直接删),子单元改挂,然后删除。 +- 重复的内容单元(DH-1 起与 `refreshMetadata` 共用 `apps/server/src/modules/sources-identity/service/unit-merge.ts`): + 同作品、同类型、同季同集号、同内容类型与分段的多条未退役单元(不论临时 / 正式),保留被引用最多的(资源、资产覆盖、下载 / + 版本槽、订阅覆盖等,不含单元自身的字段来源;同数时 confirmed 优先、再取最早的),其余单元的全部引用按外键自动发现改指向保留的那条 + (撞唯一约束的行——保留的单元上已有同一行——直接删),子单元改挂,然后删除。保留的单元:组内有正式的即转正式, + 标题 / 播出日期 / 总集数 / metadata 为空时取被合并单元的值。成因是并发识别同一集时懒建竞态,`ensureUnits` 已加咨询锁。 +- 2026-09-29 用户库只读 dry-run(`--merge-duplicates`,未 apply):2 组删除 6 条——《征途》S01E09、S01E10 各 4 条临时单元 + (同一时刻建出,都无引用),各保留最早且带 TMDB 字段来源的那条(`8d5e09fb…`、`1132f75b…`),其余 3 条合并。之后对该作品跑 + 元数据刷新也会自动合并。 - 孤立的临时内容单元(FIX-6 收紧):只删**资源解析建出、外部元数据不认识**的临时单元。同时满足才删:`provisional`;没有资源 / 资产 / 订阅 / 下载 / 整理等任何引用;没有任何元数据痕迹(字段来源 / 字段覆盖 / 标题 / 播出日期 / 总集数来源 / metadata 非空); 作品**没有**已接受的外部身份,或这个季 / 集**超出**作品的外部季集结构(季号 > TMDB 季数,或集号 > 该季已知总集数; @@ -129,8 +138,8 @@ DATABASE_URL=postgres://umt:<密码>@$PGIP:5432/umt deno run -A scripts/identity - 事故(FIX-6,2026-09-29):FIX-5 版本只看"无引用 + 没有 TMDB 字段来源",把旧版本建的、还没回写字段来源的 TMDB 季集账本单元 (《人生切割术》S00–S03 与 20 集、《沙丘：预言》S00 与 17 集等)当孤立删除。恢复:对受影响作品跑 full 元数据刷新—— 计划页执行「元数据语言回刷」(`POST /schedules/metadata.backfill_languages/run-now`,停用状态也可手动执行)或逐个 - `POST /media-entities/{id}/refresh?full=1`,正片季的季 / 集单元会按 TMDB 重建为 confirmed。**季 0(特别篇)不在 full 刷新范围内**, - 删掉的特别篇单元只能等资源 / 文件再次命中时懒建。 + `POST /media-entities/{id}/refresh?full=1`,季 / 集单元会按 TMDB 重建为 confirmed。季 0(特别篇)自 DH-1 起也在 full 刷新 + 范围内(`metadata.includeSpecials` 缺省开),回刷即可重建。 ## 排查 diff --git a/docs/runbook/metadata.md b/docs/runbook/metadata.md index b21f83ec..c2aa6bec 100644 --- a/docs/runbook/metadata.md +++ b/docs/runbook/metadata.md @@ -14,13 +14,22 @@ | 触发 | 拉哪些季的每集信息 | 新建集单元 | |---|---|---| -| `POST /media-entities/{id}/refresh`(定时追踪刷新同) | 追踪中订阅的季 ∪ 库里已有集单元的季(最近 8 季) | 只在追踪中的季 | -| `POST /media-entities/{id}/refresh?full=1`(详情页「刷新元数据」「更新每集信息」) | 全部正片季(最多 60) | 是 | +| `POST /media-entities/{id}/refresh`(定时追踪刷新同) | 追踪中订阅的季 ∪ 库里已有集单元的季(含季 0,最近 8 季) | 只在追踪中的季 | +| `POST /media-entities/{id}/refresh?full=1`(详情页「刷新元数据」「更新每集信息」) | 全部季(最多 60);TMDB 有季 0 且 `metadata.includeSpecials` 开时含季 0 | 是 | | 新建档:按 TMDB 候选建作品(订阅建档、搜索 / TMDB 预览页建档、整理向导)、手动指定未登记的 TMDB 身份 | 同 `full=1`(幂等键 `refreshMetadata:created:{id}`) | 是 | - 新建档时标题 / 简介当场按语言链取(富详情失败沿用基础详情),每集信息由随后入队的 full 刷新补齐,不用手动刷新。自动识别(RSS)新建的作品不入队 full,仍按追踪刷新。 - 刷新是后台任务 `refreshMetadata`(输入 `{mediaEntityId, full?}`),结果里 `titles` 是本次取到的标题 / 语言 / 中文名,`episodes[]` 是各季写入统计(`created` 新建单元数、`changedUnits` 有变化的单元数)。 -- 增量:来源值和语言都没变的字段不写;只投影有变化的单元。不改编号、状态、路径绑定和资产覆盖。 +- 增量:来源值和语言都没变的字段不写;只投影有变化的单元。不改编号、路径绑定和资产覆盖。 +- 季 0(特别篇,DH-1):设置 → 元数据 → TMDB「包含特别篇(季 0)」(`metadata.includeSpecials`,缺省开)。开时 full 刷新和存量回刷 + 在 TMDB 有季 0 时一并拉取每集信息并建季 0 的季 / 集单元(对照 MP);季 0 不进正片季总集数账本,不影响订阅完成判定。关掉后 + 已建的季 0 单元保留,只是不再主动拉取。 +- 临时转正式(DH-1):刷新时与 TMDB 季集结构对得上的临时(provisional)季 / 集单元转正式(`status = confirmed`、`source = tmdb`), + id 与所有引用不变;对不上的(季号不在 TMDB、集号超出该季集数)保持临时。只看本次拉了季详情的季(full 刷新即全部季)。 + `content_unit.source`(迁移 0053):`release` 资源解析懒建、`file` 整理 / 文件懒建、`tmdb` 按 TMDB 建或转正、`user` 人工;旧行为空。 +- 重复单元合并(DH-1):每次刷新先把同作品、同类型、同季同集号、同内容类型与分段的多条未退役单元合并为一条(规则同 + `scripts/identity-cleanup.ts --merge-duplicates`,见 `identity.md`),结果里 `mergedUnits` / `promotedUnits` 是合并 / 转正数。 + 懒建单元(`ensureUnits`)按(作品, 类型, 季, 集)取事务级咨询锁,并发识别不再建出重复。 - 存量回刷(FIX-5):维护任务 `metadata.backfill_languages`「元数据语言回刷」在部署后首次启动即运行,对所有已确认 TMDB 身份 (active / provisional)的作品逐个跑 `full` 刷新。用独立的 TMDB 客户端串行请求,网络请求间隔 ≥ 500 ms(每秒最多 2 次), 请求照常走 `tmdb.request` PG 缓存(对照 MP `@cached`),中断后重跑时已取过的不再请求。进度每 5 秒写进本次运行记录的摘要, @@ -58,3 +67,5 @@ curl -s -H "Authorization: Bearer $KEY" "$BASE/api/v1/media-entities?limit=200" | 标题仍是英文原名 | 作品详情「字段来源与锁」里 `标题` 的来源与语言;是否被锁;设置里的语言链;对作品跑 `full=1` | | 某集没有中文标题 | 该集在 TMDB 各语言都没有译名时保留占位名 / 英文名,属正常;可在每集行手动改并锁定 | | 季里没有集 | 默认刷新不为未追踪的季建集单元,点「更新每集信息」(full) | +| 没有特别篇 | TMDB 该剧有没有季 0;设置 `metadata.includeSpecials` 是否关了;对作品跑 `full=1` | +| 集还是「临时」 | 该季是否拉过季详情(跑 `full=1`);集号是否超出 TMDB 该季集数(对不上的保持临时,属正常) | diff --git a/packages/contracts/src/content.ts b/packages/contracts/src/content.ts index ff43ac31..8b6663c3 100644 --- a/packages/contracts/src/content.ts +++ b/packages/contracts/src/content.ts @@ -60,6 +60,9 @@ export const Segment = z.enum(["full", "upper", "middle", "lower", "unknown"]); export type Segment = z.infer; export const ContentUnitStatus = z.enum(["provisional", "confirmed", "retired"]); export type ContentUnitStatus = z.infer; +/** 内容单元来源(DH-1):资源解析懒建 / 整理·文件懒建 / 按 TMDB 季集结构建或对上后转正式 / 人工;旧行为 null */ +export const ContentUnitSource = z.enum(["release", "file", "tmdb", "user"]); +export type ContentUnitSource = z.infer; /** 季总集数来源:tmdb 元数据 / 插件注入(只升)/ 人工锁定 / 回退被已覆盖最高集兜住 / 沿用 */ export const EpisodeTotalSource = z.enum(["tmdb", "injected", "manual", "covered_floor", "previous"]); @@ -89,6 +92,7 @@ export const ContentUnit = EntityBase.extend({ /** 人工锁定总集数:元数据刷新与注入不再改写 */ episodeTotalLocked: z.boolean(), episodeTotalAt: Timestamp.nullable(), + source: ContentUnitSource.nullable(), }); export type ContentUnit = z.infer; export const ContentUnitCreate = ContentUnit.omit(SERVER_MANAGED).partial({ @@ -109,6 +113,7 @@ export const ContentUnitCreate = ContentUnit.omit(SERVER_MANAGED).partial({ episodeTotalSource: true, episodeTotalLocked: true, episodeTotalAt: true, + source: true, }); export type ContentUnitCreate = z.infer; export const ContentUnitUpdate = ContentUnitCreate.omit({ mediaEntityId: true }).partial(); diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index 173e034f..93f452cb 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -226,6 +226,11 @@ export const MetadataSettings = z.object({ .min(1) .max(8) .default(["zh-CN", "zh-SG", "zh-TW", "zh-HK", "en-US"]), + /** + * DH-1 季 0(特别篇):true = full 刷新与元数据回刷在 TMDB 有季 0 时一并拉取每集信息并建集单元(对照 MP); + * 季 0 不计入正片季总集数与订阅完成判定 + */ + includeSpecials: z.boolean().default(true), }); export type MetadataSettings = z.infer; @@ -611,7 +616,12 @@ export const DEFAULT_SETTINGS: GlobalSettings = { sessionDir: null, webSearch: DEFAULT_WEB_SEARCH, }, - metadata: { tmdbApiKeySecretRef: null, tmdbBaseUrl: null, languages: ["zh-CN", "zh-SG", "zh-TW", "zh-HK", "en-US"] }, + metadata: { + tmdbApiKeySecretRef: null, + tmdbBaseUrl: null, + languages: ["zh-CN", "zh-SG", "zh-TW", "zh-HK", "en-US"], + includeSpecials: true, + }, sites: { defaultRateLimit: { requestsPerMinute: 20, concurrency: 2 }, defaultSnapshotIntervalSec: 3600, @@ -906,6 +916,7 @@ export const SETTINGS_SECTIONS: readonly SettingsSectionDef[] = [ "metadata.tmdbApiKeySecretRef", "metadata.tmdbBaseUrl", "metadata.languages", + "metadata.includeSpecials", "cache.tmdbMaxSize", "identity.eagerTmdbLookup", "identity.autoEnrich", diff --git a/packages/contracts/test/db-sync.test.ts b/packages/contracts/test/db-sync.test.ts index be10f090..197081db 100644 --- a/packages/contracts/test/db-sync.test.ts +++ b/packages/contracts/test/db-sync.test.ts @@ -283,6 +283,8 @@ const ENUMS: Record = { BrowserControlLeaseStatus: C.BrowserLeaseStatus, // W5-C(0025) ContentUnitEpisodeTotalSource: C.EpisodeTotalSource, + // DH-1(0053) + ContentUnitSource: C.ContentUnitSource, }; describe("contracts ⇔ db 表结构同步", () => { diff --git a/packages/db/migrations/0053_content_unit_source.ts b/packages/db/migrations/0053_content_unit_source.ts new file mode 100644 index 00000000..b4d55cb6 --- /dev/null +++ b/packages/db/migrations/0053_content_unit_source.ts @@ -0,0 +1,24 @@ +import type { Kysely } from "kysely"; +import { execDdl, oneOf } from "../src/migration-helpers.ts"; + +// DH-1 内容单元来源:release(资源解析懒建)、file(整理 / 文件扫描懒建)、tmdb(TMDB 季集结构建或对上后转正式)、user(人工)。 +// 旧行:已确认且带 TMDB 字段来源 / TMDB 总集数的单元回填 tmdb;其余留空(来源未知)。 + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl( + db, + ` +alter table content_unit add column source text ${oneOf("source", ["release", "file", "tmdb", "user"])}; +update content_unit u set source = 'tmdb' + where u.status = 'confirmed' + and (u.episode_total_source = 'tmdb' + or exists (select 1 from metadata_field f where f.content_unit_id = u.id and f.source = 'tmdb')); +`, + ); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl(db, `alter table content_unit drop column if exists source;`); +} diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index e244423f..69d345d5 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -100,6 +100,7 @@ export type ContentUnitContentType = export type ContentUnitSegment = "full" | "upper" | "middle" | "lower" | "unknown"; export type ContentUnitStatus = "provisional" | "confirmed" | "retired"; export type ContentUnitEpisodeTotalSource = "tmdb" | "injected" | "manual" | "covered_floor" | "previous"; +export type ContentUnitSource = "release" | "file" | "tmdb" | "user"; export type DecisionKind = | "identity_ambiguous" | "download_stalled" @@ -642,6 +643,7 @@ export interface ContentUnitTable { episode_total_source: ContentUnitEpisodeTotalSource | null; episode_total_locked: Generated; episode_total_at: Timestamptz | null; + source: ContentUnitSource | null; } export interface DecisionTable { diff --git a/scripts/identity-cleanup.ts b/scripts/identity-cleanup.ts index 4d3d8800..65599dad 100644 --- a/scripts/identity-cleanup.ts +++ b/scripts/identity-cleanup.ts @@ -1,27 +1,36 @@ -// 身份数据清理(FIX-5):删除冲突别名、合并重复的临时内容单元、删除孤立的临时内容单元。默认只读(dry-run),加 --apply 才改库。 +// 身份数据清理(FIX-5):删除冲突别名、合并重复的内容单元、删除孤立的临时内容单元。默认只读(dry-run),加 --apply 才改库。 // // 用法(DATABASE_URL 必须显式给出,脚本不带默认库): // DATABASE_URL=postgres://umt:***@:5432/umt deno run -A scripts/identity-cleanup.ts [--dry-run] # 看清单(按类别列删除 / 保留) // DATABASE_URL=… deno run -A scripts/identity-cleanup.ts --apply # 执行(单个事务) // DATABASE_URL=… deno run -A scripts/identity-cleanup.ts --keep-orphans # 不删孤立单元 // DATABASE_URL=… deno run -A scripts/identity-cleanup.ts --entity <作品 id> # 只看 / 只处理一部作品 +// DATABASE_URL=… deno run -A scripts/identity-cleanup.ts --merge-duplicates [--apply] # 只做第 2 步:合并重复单元(DH-1) // // 1. 冲突别名:只处理自动识别登记的站点别名(title_alias.source = 'site',作品级)。以下任一成立即删除: // - shared:同一归一化别名还挂在别的作品上,且它不是本作品自己的名字(主标题 / 原标题 / TMDB 译名与别名); // - script:作品有已接受的外部身份,且同一种文字(中文 / 拉丁)下已有确认的外部标题,这个别名对不上任何一个 // (中文标题的拼音写法视为相容,保留); // - year:本作品下带这个名字的资源都解析出了年份,且都与作品年份相差超过 1 年。 -// 2. 重复的临时内容单元:同一作品、同一类型、同季同集号有多条(未退役)且至少一条是临时(provisional)的, -// 保留被引用最多的那条(同数时 confirmed 优先,再取最早建的),其余的引用改指向保留的那条后删除; -// 改指向会撞唯一约束的行(同一字段来源 / 别名已在保留的单元上)直接删掉。 +// 2. 重复的内容单元(DH-1,与 refreshMetadata 共用 apps/server/.../unit-merge.ts):同一作品、同一类型、同季同集号、 +// 同内容类型与分段有多条未退役单元时,保留被引用最多的那条(同数时 confirmed 优先,再取最早建的),其余的全部引用 +// (资源覆盖、资产、下载计划、订阅覆盖、字段来源…按外键自动发现)改指向保留的那条后删除; +// 改指向会撞唯一约束的行(同一行已在保留的单元上)直接删掉。 // 3. 孤立的临时内容单元(--keep-orphans 跳过):只删资源解析建出的临时单元——provisional、没有任何资源 / 资产 / 订阅 / // 下载 / 整理等引用、没有任何元数据痕迹(字段来源 / 覆盖 / 标题 / 播出日期 / 总集数来源),且作品没有外部映射或 // 这个季 / 集超出作品的外部季集结构;没有保留的子单元。TMDB / 元数据来源的单元永不删除(FIX-6)。 // 字段来源 / 覆盖 / 单元别名随之删除。 // --apply 在一个事务里执行,并写一条审计事件(action = identity.cleanup)。 +import type { DB } from "@umt/db"; import { normalizeTitle, parseRelease } from "@umt/domain"; +import { Kysely, PostgresDialect } from "kysely"; import pg from "pg"; +import { + applyUnitMerges, + planUnitMerges, + type UnitMergeGroup, +} from "../apps/server/src/modules/sources-identity/service/unit-merge.ts"; const argv = [...Deno.args]; const APPLY = argv.includes("--apply"); @@ -29,7 +38,8 @@ if (APPLY && argv.includes("--dry-run")) { console.error("--apply 与 --dry-run 不能同时给出"); Deno.exit(2); } -const KEEP_ORPHANS = argv.includes("--keep-orphans"); +const MERGE_ONLY = argv.includes("--merge-duplicates"); +const KEEP_ORPHANS = argv.includes("--keep-orphans") || MERGE_ONLY; const entityArg = argv.indexOf("--entity"); const ONLY_ENTITY = entityArg >= 0 ? (argv[entityArg + 1] ?? null) : null; const url = Deno.env.get("DATABASE_URL"); @@ -39,8 +49,16 @@ if (!url) { } const db = new pg.Client({ connectionString: url }); await db.connect(); +// 合并模块用 Kysely:与 q() 共用同一个连接(同一事务) +const kdb = new Kysely({ + dialect: new PostgresDialect({ + pool: { connect: async () => Object.assign(db, { release() {} }), end: async () => {} } as never, + }), +}); const q = async (text: string, values: unknown[] = []) => (await db.query(text, values)).rows as T[]; -console.log(`# identity-cleanup (${APPLY ? "APPLY" : "DRY-RUN"})${KEEP_ORPHANS ? " --keep-orphans" : ""}\n`); +console.log( + `# identity-cleanup (${APPLY ? "APPLY" : "DRY-RUN"})${MERGE_ONLY ? " --merge-duplicates" : KEEP_ORPHANS ? " --keep-orphans" : ""}\n`, +); const isHan = (s: string) => /\p{Script=Han}/u.test(s); @@ -80,11 +98,13 @@ interface EntityRow { } const entityFilter = ONLY_ENTITY ? "and a.media_entity_id = $1" : ""; const entityVals = ONLY_ENTITY ? [ONLY_ENTITY] : []; -const siteAliases = await q( - `select a.id, a.media_entity_id, a.alias, a.normalized, a.source from title_alias a - where a.source = 'site' and a.content_unit_id is null ${entityFilter}`, - entityVals, -); +const siteAliases = MERGE_ONLY + ? [] + : await q( + `select a.id, a.media_entity_id, a.alias, a.normalized, a.source from title_alias a + where a.source = 'site' and a.content_unit_id is null ${entityFilter}`, + entityVals, + ); const entityIds = [...new Set(siteAliases.map((a) => a.media_entity_id))]; const entities = new Map( ( @@ -183,7 +203,7 @@ const entityTitle = async (id: string) => entities.get(id)?.title ?? (await q<{ title: string }>(`select title from media_entity where id = $1`, [id]))[0]?.title ?? id; -console.log(`## 冲突别名:${aliasDeletes.length} 个`); +if (!MERGE_ONLY) console.log(`## 冲突别名:${aliasDeletes.length} 个`); for (const eid of [...new Set(aliasDeletes.map((a) => a.media_entity_id))]) { const e = entities.get(eid)!; console.log(`\n### ${e.title}(${e.year ?? "-"}) ${eid}`); @@ -228,7 +248,7 @@ async function refCounts(ids: string[]): Promise> { return out; } -// ---------------------------------------------------------------- 2. 重复的临时内容单元 +// ---------------------------------------------------------------- 2. 重复的内容单元 interface UnitRow { id: string; media_entity_id: string; @@ -239,43 +259,19 @@ interface UnitRow { created_at: Date; } const unitFilter = ONLY_ENTITY ? "and u.media_entity_id = $1" : ""; -const dupRows = await q( - `select u.id, u.media_entity_id, u.kind, u.season_number, u.episode_number, u.status, u.created_at - from content_unit u - join (select media_entity_id, kind, season_number, episode_number from content_unit - where status <> 'retired' group by 1, 2, 3, 4 - having count(*) > 1 and bool_or(status = 'provisional')) 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 - where u.status <> 'retired' ${unitFilter} - order by u.media_entity_id, u.season_number, u.episode_number, u.created_at`, - ONLY_ENTITY ? [ONLY_ENTITY] : [], -); -const dupRefs = await refCounts(dupRows.map((u) => u.id)); -const groups = new Map(); -for (const u of dupRows) { - const k = `${u.media_entity_id}|${u.kind}|${u.season_number}|${u.episode_number}`; - groups.set(k, [...(groups.get(k) ?? []), u]); -} -const merges: { keep: UnitRow; drop: UnitRow[] }[] = []; -for (const list of groups.values()) { - const sorted = [...list].sort( - (a, b) => - (dupRefs.get(b.id) ?? 0) - (dupRefs.get(a.id) ?? 0) || - Number(b.status === "confirmed") - Number(a.status === "confirmed") || - new Date(a.created_at).getTime() - new Date(b.created_at).getTime(), - ); - merges.push({ keep: sorted[0]!, drop: sorted.slice(1) }); -} +const merges: UnitMergeGroup[] = await planUnitMerges(kdb, ONLY_ENTITY ? { mediaEntityId: ONLY_ENTITY } : {}); const unitLabel = (u: Pick) => u.kind === "season" ? `S${String(u.season_number ?? 0).padStart(2, "0")}` : `S${String(u.season_number ?? 0).padStart(2, "0")}E${String(u.episode_number ?? 0).padStart(2, "0")}`; -console.log(`\n## 重复的临时内容单元:${merges.length} 组,删除 ${merges.reduce((s, m) => s + m.drop.length, 0)} 条`); +const groupLabel = (g: UnitMergeGroup) => + unitLabel({ kind: g.kind, season_number: g.seasonNumber, episode_number: g.episodeNumber }) + + (g.contentType !== "main" || g.segment !== "full" ? `(${g.contentType}/${g.segment})` : ""); +console.log(`\n## 重复的内容单元:${merges.length} 组,删除 ${merges.reduce((s, m) => s + m.drop.length, 0)} 条`); for (const m of merges) { console.log( - ` - 《${await entityTitle(m.keep.media_entity_id)}》${unitLabel(m.keep)}:保留 ${m.keep.id}(${m.keep.status},引用 ${dupRefs.get(m.keep.id) ?? 0}),` + - `合并 ${m.drop.map((d) => `${d.id}(${d.status},引用 ${dupRefs.get(d.id) ?? 0})`).join("、")}`, + ` - 《${await entityTitle(m.mediaEntityId)}》${groupLabel(m)}:保留 ${m.keep.id}(${m.keep.status},引用 ${m.keep.refs}),` + + `合并 ${m.drop.map((d) => `${d.id}(${d.status},引用 ${d.refs})`).join("、")}`, ); } @@ -395,7 +391,7 @@ if (!KEEP_ORPHANS) { console.log(` - 保留·${KEEP_TEXT[r]}:${l.length} 条`); if (r !== "referenced") await printGroup(l); } -} else console.log("\n## 孤立的临时内容单元:跳过(--keep-orphans)"); +} else if (!MERGE_ONLY) console.log("\n## 孤立的临时内容单元:跳过(--keep-orphans)"); // ---------------------------------------------------------------- 执行 if (APPLY) { @@ -403,37 +399,7 @@ if (APPLY) { try { if (aliasDeletes.length) await q(`delete from title_alias where id = any($1::uuid[])`, [aliasDeletes.map((a) => a.id)]); - let repointed = 0; - let droppedRows = 0; - const allRefTables = fks.filter((f) => f.table !== "content_unit"); - for (const m of merges) { - for (const d of m.drop) { - for (const f of allRefTables) { - const rows = await q<{ ctid: string }>( - `select ctid::text from ${quoteIdent(f.table)} where ${quoteIdent(f.column)} = $1`, - [d.id], - ); - for (const r of rows) { - await db.query("savepoint sp"); - try { - await q(`update ${quoteIdent(f.table)} set ${quoteIdent(f.column)} = $1 where ctid = $2::tid`, [ - m.keep.id, - r.ctid, - ]); - await db.query("release savepoint sp"); - repointed++; - } catch (e) { - await db.query("rollback to savepoint sp"); - if ((e as { code?: string }).code !== "23505") throw e; - await q(`delete from ${quoteIdent(f.table)} where ctid = $1::tid`, [r.ctid]); - droppedRows++; - } - } - } - await q(`update content_unit set parent_unit_id = $1 where parent_unit_id = $2`, [m.keep.id, d.id]); - await q(`delete from content_unit where id = $1`, [d.id]); - } - } + const merged = await applyUnitMerges(kdb, merges); // 孤立单元:先集后季 const orphanIds = orphans .sort((a, b) => Number(a.kind === "season") - Number(b.kind === "season")) @@ -453,6 +419,7 @@ if (APPLY) { JSON.stringify({ aliases: aliasDeletes.map((a) => ({ entityId: a.media_entity_id, alias: a.alias, reason: a.reason })), mergedUnits: merges.map((m) => ({ keep: m.keep.id, drop: m.drop.map((d) => d.id) })), + mode: MERGE_ONLY ? "merge-duplicates" : "full", orphanUnits: orphans.map((u) => ({ id: u.id, entityId: u.media_entity_id, @@ -464,7 +431,7 @@ if (APPLY) { ); await db.query("commit"); console.log( - `\n已执行:删除 ${aliasDeletes.length} 个别名;合并 ${dropping.size} 个重复单元(改指向 ${repointed} 行,删除冲突行 ${droppedRows} 行);` + + `\n已执行:删除 ${aliasDeletes.length} 个别名;合并 ${merged.merged} 个重复单元(改指向 ${merged.repointed} 行,删除冲突行 ${merged.droppedRows} 行);` + `删除 ${orphans.length} 个孤立单元`, ); } catch (e) {