// 身份数据清理(FIX-5):删除冲突别名、合并重复的临时内容单元、删除孤立的临时内容单元。默认只读(dry-run),加 --apply 才改库。
//
// 用法(DATABASE_URL 必须显式给出,脚本不带默认库):
//   DATABASE_URL=postgres://umt:***@<host>:5432/umt deno run -A scripts/identity-cleanup.ts                 # 看清单
//   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>                              # 只看 / 只处理一部作品
//
// 1. 冲突别名:只处理自动识别登记的站点别名(title_alias.source = 'site',作品级)。以下任一成立即删除:
//    - shared:同一归一化别名还挂在别的作品上,且它不是本作品自己的名字(主标题 / 原标题 / TMDB 译名与别名);
//    - script:作品有已接受的外部身份,且同一种文字(中文 / 拉丁)下已有确认的外部标题,这个别名对不上任何一个
//      (中文标题的拼音写法视为相容,保留);
//    - year:本作品下带这个名字的资源都解析出了年份,且都与作品年份相差超过 1 年。
// 2. 重复的临时内容单元:同一作品、同一类型、同季同集号有多条(未退役)且至少一条是临时(provisional)的,
//    保留被引用最多的那条(同数时 confirmed 优先,再取最早建的),其余的引用改指向保留的那条后删除;
//    改指向会撞唯一约束的行(同一字段来源 / 别名已在保留的单元上)直接删掉。
// 3. 孤立的临时内容单元(--keep-orphans 跳过):provisional、没有任何资源 / 资产 / 订阅 / 下载 / 整理等引用、
//    没有 TMDB 来源的字段(不是 TMDB 已知的季 / 集)、也没有子单元的,删除;字段来源 / 覆盖 / 单元别名随之删除。
// --apply 在一个事务里执行,并写一条审计事件(action = identity.cleanup)。

import { normalizeTitle, parseRelease } from "@umt/domain";
import pg from "pg";

const argv = [...Deno.args];
const APPLY = argv.includes("--apply");
const KEEP_ORPHANS = argv.includes("--keep-orphans");
const entityArg = argv.indexOf("--entity");
const ONLY_ENTITY = entityArg >= 0 ? (argv[entityArg + 1] ?? null) : null;
const url = Deno.env.get("DATABASE_URL");
if (!url) {
  console.error("需要 DATABASE_URL(脚本不带默认库)");
  Deno.exit(2);
}
const db = new pg.Client({ connectionString: url });
await db.connect();
const q = async <T>(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`);

const isHan = (s: string) => /\p{Script=Han}/u.test(s);

// ---------------------------------------------------------------- 拼音判断(中文标题的拉丁写法视为相容)
const SYLLABLE =
  /^(?:(?:zh|ch|sh|[bpmfdtnlgkhjqxrzcsyw])?(?:iang|iong|uang|ueng|ang|eng|ing|ong|iao|ian|uai|uan|van|ai|ei|ui|ao|ou|iu|ie|ve|ue|er|an|en|in|un|vn|ia|ua|uo|a|o|e|i|u|v))$/;
/** 全部由拼音音节组成(允许音节连写,如 Dongbei) */
function looksLikePinyin(text: string): boolean {
  const words = text
    .toLowerCase()
    .replace(/[^a-z\s'-]/g, " ")
    .split(/[\s'-]+/)
    .filter(Boolean);
  if (!words.length || /[^a-z\s'-]/i.test(text.replace(/[\s\d.:,!?·]/g, ""))) return false;
  const seg = (w: string): boolean => {
    if (!w) return true;
    for (let k = Math.min(6, w.length); k >= 1; k--) if (SYLLABLE.test(w.slice(0, k)) && seg(w.slice(k))) return true;
    return false;
  };
  return words.every(seg);
}

// ---------------------------------------------------------------- 1. 冲突别名
interface AliasRow {
  id: string;
  media_entity_id: string;
  alias: string;
  normalized: string;
  source: string;
}
interface EntityRow {
  id: string;
  title: string;
  original_title: string | null;
  year: number | null;
  confirmed: boolean;
}
const entityFilter = ONLY_ENTITY ? "and a.media_entity_id = $1" : "";
const entityVals = ONLY_ENTITY ? [ONLY_ENTITY] : [];
const siteAliases = await q<AliasRow>(
  `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(
  (
    await q<EntityRow>(
      `select e.id, e.title, e.original_title, e.year,
              exists(select 1 from external_identity x where x.media_entity_id = e.id and x.status = 'accepted') as confirmed
         from media_entity e where e.id = any($1::uuid[])`,
      [entityIds],
    )
  ).map((e) => [e.id, e]),
);
const tmdbAliases = await q<{ media_entity_id: string; alias: string }>(
  `select media_entity_id, alias from title_alias where source = 'tmdb' and content_unit_id is null
     and media_entity_id = any($1::uuid[])`,
  [entityIds],
);
/** 作品自己的名字(归一化),按文字分组 */
const ownNames = new Map<string, { han: Set<string>; latin: Set<string> }>();
for (const e of entities.values()) {
  const own = { han: new Set<string>(), latin: new Set<string>() };
  for (const t of [
    e.title,
    e.original_title,
    ...tmdbAliases.filter((a) => a.media_entity_id === e.id).map((a) => a.alias),
  ])
    if (t) {
      const n = normalizeTitle(t);
      if (n) own[isHan(t) ? "han" : "latin"].add(n);
    }
  ownNames.set(e.id, own);
}
const shared = new Map<string, Set<string>>();
for (const r of await q<{ normalized: string; media_entity_id: string }>(
  `select distinct a.normalized, a.media_entity_id from title_alias a join media_entity e on e.id = a.media_entity_id
    where a.content_unit_id is null and e.status <> 'merged' and a.normalized = any($1::text[])`,
  [[...new Set(siteAliases.map((a) => a.normalized))]],
)) {
  const s = shared.get(r.normalized) ?? new Set<string>();
  s.add(r.media_entity_id);
  shared.set(r.normalized, s);
}
// 同名作品的主标题也算"别的作品的名字"
for (const r of await q<{ id: string; title: string; original_title: string | null }>(
  `select id, title, original_title from media_entity where status <> 'merged'`,
)) {
  for (const t of [r.title, r.original_title]) {
    if (!t) continue;
    const n = normalizeTitle(t);
    const s = shared.get(n);
    if (s) s.add(r.id);
  }
}
const releasesByEntity = new Map<string, { title: string; subtitle: string | null }[]>();
for (const r of await q<{ media_entity_id: string; title: string; subtitle: string | null }>(
  `select media_entity_id, title, subtitle from source_release where media_entity_id = any($1::uuid[])`,
  [entityIds],
)) {
  const l = releasesByEntity.get(r.media_entity_id) ?? [];
  l.push(r);
  releasesByEntity.set(r.media_entity_id, l);
}

type AliasReason = "shared" | "script" | "year";
const aliasDeletes: (AliasRow & { reason: AliasReason; others?: string[] })[] = [];
for (const a of siteAliases) {
  const e = entities.get(a.media_entity_id);
  if (!e) continue;
  const own = ownNames.get(e.id)!;
  const n = normalizeTitle(a.alias) || a.normalized;
  const script = isHan(a.alias) ? "han" : "latin";
  if (own.han.has(n) || own.latin.has(n)) continue;
  const others = [...(shared.get(a.normalized) ?? [])].filter((id) => id !== e.id);
  if (others.length) {
    aliasDeletes.push({ ...a, reason: "shared", others });
    continue;
  }
  if (e.confirmed && own[script].size > 0 && !(script === "latin" && looksLikePinyin(a.alias))) {
    aliasDeletes.push({ ...a, reason: "script" });
    continue;
  }
  if (e.year !== null) {
    const years = (releasesByEntity.get(e.id) ?? [])
      .map((r) => parseRelease(r.title, r.subtitle ? { subtitle: r.subtitle } : {}))
      .filter((p) => [p.titleCn, p.titleEn, p.title].some((t) => t && normalizeTitle(t) === n))
      .map((p) => p.year ?? null);
    if (years.length && years.every((y) => y !== null && Math.abs(y - e.year!) > 1))
      aliasDeletes.push({ ...a, reason: "year" });
  }
}
const REASON_TEXT: Record<AliasReason, string> = {
  shared: "同名别名挂在别的作品上",
  script: "与已确认的外部标题对不上",
  year: "带这个名字的资源年份都与作品不符",
};
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} 个`);
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}`);
  for (const a of aliasDeletes.filter((x) => x.media_entity_id === eid)) {
    const also = a.others?.length
      ? `:${(await Promise.all(a.others.slice(0, 3).map(entityTitle))).map((t) => `《${t}》`).join("")}`
      : "";
    console.log(`  - 删除「${a.alias}」(${REASON_TEXT[a.reason]}${also})`);
  }
}

// ---------------------------------------------------------------- 引用 content_unit 的表
interface FkRef {
  table: string;
  column: string;
}
const fks = await q<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`,
);
/** 只是单元自身的元数据(随单元删除),不算"被使用" */
const META_TABLES = new Set(["metadata_field", "field_override", "title_alias"]);
const usageRefs = fks.filter((f) => !META_TABLES.has(f.table.replace(/^.*\./, "")) && f.table !== "content_unit");
const quoteIdent = (s: string) =>
  s
    .split(".")
    .map((p) => `"${p.replace(/"/g, '""')}"`)
    .join(".");
async function refCounts(ids: 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 usageRefs) {
    const rows = await q<{ id: string; n: number }>(
      `select ${quoteIdent(f.column)} as id, count(*)::int as n from ${quoteIdent(f.table)}
        where ${quoteIdent(f.column)} = any($1::uuid[]) group by 1`,
      [ids],
    );
    for (const r of rows) out.set(r.id, (out.get(r.id) ?? 0) + r.n);
  }
  return out;
}

// ---------------------------------------------------------------- 2. 重复的临时内容单元
interface UnitRow {
  id: string;
  media_entity_id: string;
  kind: string;
  season_number: number | null;
  episode_number: number | null;
  status: string;
  created_at: Date;
}
const unitFilter = ONLY_ENTITY ? "and u.media_entity_id = $1" : "";
const dupRows = await q<UnitRow>(
  `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<string, UnitRow[]>();
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 unitLabel = (u: Pick<UnitRow, "kind" | "season_number" | "episode_number">) =>
  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)} 条`);
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("、")}`,
  );
}

// ---------------------------------------------------------------- 3. 孤立的临时内容单元
const dropping = new Set(merges.flatMap((m) => m.drop.map((d) => d.id)));
let orphans: UnitRow[] = [];
if (!KEEP_ORPHANS) {
  const cand = await q<UnitRow>(
    `select u.id, u.media_entity_id, u.kind, u.season_number, u.episode_number, u.status, u.created_at
       from content_unit u
      where u.status = 'provisional' ${unitFilter}
        and not exists (select 1 from metadata_field f where f.content_unit_id = u.id and f.source = 'tmdb')
      order by u.media_entity_id, u.kind desc, u.season_number, u.episode_number`,
    ONLY_ENTITY ? [ONLY_ENTITY] : [],
  );
  const refs = await refCounts(cand.map((u) => u.id));
  const unused = cand.filter((u) => (refs.get(u.id) ?? 0) === 0 && !dropping.has(u.id));
  const unusedIds = new Set(unused.map((u) => u.id));
  // 季单元:子单元全都要删(或没有子单元)才删
  const children = await q<{ parent_unit_id: string; id: string }>(
    `select parent_unit_id, id from content_unit where parent_unit_id = any($1::uuid[])`,
    [unused.filter((u) => u.kind === "season").map((u) => u.id)],
  );
  orphans = unused.filter(
    (u) =>
      u.kind !== "season" ||
      children.filter((c) => c.parent_unit_id === u.id).every((c) => unusedIds.has(c.id) || dropping.has(c.id)),
  );
  console.log(`\n## 孤立的临时内容单元:${orphans.length} 条`);
  const byEntity = new Map<string, UnitRow[]>();
  for (const u of orphans) byEntity.set(u.media_entity_id, [...(byEntity.get(u.media_entity_id) ?? []), u]);
  for (const [eid, list] of byEntity) console.log(`  - 《${await entityTitle(eid)}》:${list.map(unitLabel).join(" ")}`);
} else console.log("\n## 孤立的临时内容单元:跳过(--keep-orphans)");

// ---------------------------------------------------------------- 执行
if (APPLY) {
  await db.query("begin");
  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 orphanIds = orphans
      .sort((a, b) => Number(a.kind === "season") - Number(b.kind === "season"))
      .map((u) => u.id);
    for (const t of META_TABLES)
      if (fks.some((f) => f.table.replace(/^.*\./, "") === t))
        await q(`delete from ${quoteIdent(t)} where content_unit_id = any($1::uuid[])`, [orphanIds]);
    for (const u of orphans.filter((x) => x.kind !== "season"))
      await q(`delete from content_unit where id = $1`, [u.id]);
    for (const u of orphans.filter((x) => x.kind === "season"))
      await q(`delete from content_unit where id = $1`, [u.id]);
    await q(
      `insert into audit_event (actor_type, actor_id, action, target_type, target_id, payload)
       values ('system', 'identity-cleanup', 'identity.cleanup', 'media_entity', $1, $2::jsonb)`,
      [
        ONLY_ENTITY,
        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) })),
          orphanUnits: orphans.length,
        }),
      ],
    );
    await db.query("commit");
    console.log(
      `\n已执行:删除 ${aliasDeletes.length} 个别名;合并 ${dropping.size} 个重复单元(改指向 ${repointed} 行,删除冲突行 ${droppedRows} 行);` +
        `删除 ${orphans.length} 个孤立单元`,
    );
  } catch (e) {
    await db.query("rollback");
    throw e;
  }
} else console.log("\n(dry-run:未改库;确认后加 --apply 执行)");
await db.end();