UHD_Media_Tool/apps/server/test/sources-identity/data-hygiene.test.ts

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<string, string> = {};

  const unit = async (
    kind: "season" | "episode",
    season: number,
    episode: number | null,
    extra: Record<string, unknown> = {},
  ) =>
    (
      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 期");
    // 结构内对不上的集保持临时;结构外的季(TMDB 只有 0–2 季)连同其临时集退役
    expect(byLabel.get("S1E13")).toMatchObject({ status: "provisional" });
    for (const l of ["S3", "S3E1"]) expect(byLabel.get(l)).toMatchObject({ status: "retired" });
    // 季 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<void>((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("季范围资源(把全 24 集写成 S01-S24)只建第一季、挂已有的季;刷新时退役空的临时季", async () => {
    const seasonsNow = async () =>
      (
        await h.db
          .selectFrom("content_unit")
          .select(["id", "season_number", "status"])
          .where("media_entity_id", "=", id)
          .where("kind", "=", "season")
          .where("status", "<>", "retired")
          .orderBy("season_number")
          .execute()
      ).map((s) => s.season_number);
    const before = await seasonsNow();
    const parsed = { season: 1, seasonEnd: 24 } as ParsedRelease;
    const r = await h.db.transaction().execute((trx) => ensureUnits(trx, id, "series", parsed));
    expect(await seasonsNow()).toEqual(before);
    expect(r.unitIds.length).toBe(before.filter((n) => n !== null && n >= 1).length);

    // 结构外的季:新资源不再建(S05 单季也不建);已有的 S7(空)、S8(被资源挂着)刷新时退役,资源解开;
    // S9 下的集有文件覆盖 → 整季留着要人看
    const s5 = await h.db
      .transaction()
      .execute((trx) => ensureUnits(trx, id, "series", { season: 5 } as ParsedRelease));
    expect(s5).toEqual({ unitIds: [], primary: null });
    const s7 = await unit("season", 7, null);
    const s8 = await unit("season", 8, null);
    const s9 = await unit("season", 9, null);
    const s9e1 = await unit("episode", 9, 1, { parent_unit_id: s9 });
    await cover(await asset(), s9e1);
    await release("s8-pack", s8);
    await refreshMetadata(h.rt, id);
    const status = async (uid: string) =>
      (await h.db.selectFrom("content_unit").select("status").where("id", "=", uid).executeTakeFirstOrThrow()).status;
    expect(await status(s7)).toBe("retired");
    expect(await status(s8)).toBe("retired");
    expect(await status(s9)).toBe("provisional");
    const pack = await h.db
      .selectFrom("source_release")
      .select(["content_unit_id", "media_entity_id"])
      .where("site_release_id", "=", "s8-pack")
      .executeTakeFirstOrThrow();
    expect(pack).toEqual({ content_unit_id: null, media_entity_id: id });
  });

  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);
  });
});