merge: fix/e-scrape

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
This commit is contained in:
rootandClaude Opus 5.5 committed 2026-10-03 21:17:16 +08:00
commit b3555c04b5
24 files changed
+1039 -79

No files matched your search

@@ -1,5 +1,5 @@
import type { TmdbRichDetail, TmdbRichSeasonDetail } from "@umt/adapters";
import type { NfoEpisode, NfoIds, NfoPerson, NfoSeason, NfoWork } from "@umt/domain";
import { type NfoEpisode, type NfoIds, type NfoPerson, type NfoSeason, type NfoWork, scrapeKindOf } from "@umt/domain";
import { richEntityFields } from "../../sources-identity/service/metadata-rich.ts";
// SCR-1:本地作品 / 季 / 集的最终字段(media_entity.metadata、content_unit.metadata,字段级 {value, …})→ NFO 输入。
@@ -71,7 +71,7 @@ export function nfoWork(e: EntityForNfo, ids: IdentityMap): NfoWork {
douban: ids.douban ?? null,
};
const crew = credits(val(m, "crew"));
const isTv = e.kind !== "movie";
const isTv = scrapeKindOf(e.kind) === "tv";
return {
kind: isTv ? "tvshow" : "movie",
title: str(val(m, "title")) ?? e.title,
@@ -136,6 +136,22 @@ export function nfoEpisode(season: number, episode: number, u: UnitForNfo | unde
};
}
/** 单元被元数据来源填过(名称 / 简介 / 日期至少有一个字段,值为空也算:TMDB 就是没有,刷新也补不出来) */
const filled = (m: unknown) => !!m && typeof m === "object" && ["name", "overview", "airDate"].some((k) => k in m);
/**
* 季单元缺元数据(没有单元,或从没被元数据来源填过、也没有名称):不写「季 N」占位 NFO。
* 「缺」指还没取到(整理懒建、改身份后清空),不是 TMDB 本身没写简介。
*/
export function seasonMetadataMissing(u: UnitForNfo | undefined): boolean {
return !u || (!filled(u.metadata) && !str(u.title));
}
/** 集单元缺元数据(没有单元,或从没被填过、也没有标题和播出日期):不写「第 N 集」占位 NFO */
export function episodeMetadataMissing(u: UnitForNfo | undefined): boolean {
return !u || (!filled(u.metadata) && !str(u.title) && !u.air_date);
}
/** 图片路径:作品海报 / 背景图,季海报,集剧照 */
export function imagePathsOf(entityMeta: unknown) {
return {
@@ -1,5 +1,5 @@
import { randomUUID } from "node:crypto";
import { lstat, mkdir, readdir, realpath, rename, rm, writeFile } from "node:fs/promises";
import { createHash, randomUUID } from "node:crypto";
import { lstat, mkdir, readdir, readFile, realpath, rename, rm, writeFile } from "node:fs/promises";
import { dirname, isAbsolute, join, relative, resolve } from "node:path";
import { checkRoot, DEFAULT_ROOT_SENTINEL } from "@umt/adapters";
import {
@@ -19,6 +19,7 @@ import {
planScrapeTargets,
type ScrapeTarget,
type ScrapeVideo,
scrapeKindOf,
seasonNfo,
seasonOfDirName,
workNfo,
@@ -31,11 +32,13 @@ import { errMessage, jsonb } from "../util.ts";
import type { AoDeps } from "./common.ts";
import {
type EntityForNfo,
episodeMetadataMissing,
type IdentityMap,
imagePathsOf,
nfoEpisode,
nfoSeason,
nfoWork,
seasonMetadataMissing,
tmdbEntity,
tmdbSeasonUnits,
type UnitForNfo,
@@ -51,7 +54,10 @@ import {
// 剧集目录 → tvshow.nfo + 海报 / 背景图 + 季目录(Season 1 / S01 / Specials)的 season.nfo 与季海报 + 每集 NFO 与剧照;
// 季目录 → season.nfo + 季海报(季目录 poster.jpg 与上级 seasonNN-poster.jpg)+ 每集;单集文件 → 该集 NFO 与剧照。
// - 字段:有本地作品且有元数据时用最终字段(字段锁、人工覆盖优先);否则按 TMDB id 直接取 TMDB(与刷新元数据同一映射)。
// - 覆盖:missing 只补缺失 / overwrite 全部覆盖;写入走同目录临时文件 + rename;目标必须在登记目录内。
// - 覆盖:update(缺省)只覆盖本工具上次写入且没被改过的文件(逐文件 sha256 记在 scrape_record.files,
// 改过的记 scrape_user_modified)/ missing 只补缺失 / overwrite 全部覆盖;写入走同目录临时文件 + rename;
// 目标必须在登记目录内。
// - 季 / 集缺元数据时先按 TMDB id 现取季详情补齐,仍缺就跳过(scrape_metadata_missing,记录 partial),不写占位 NFO。
// - 每次刮削一条 scrape_record(目录、识别结果、逐文件结果、原因);审计 organize.scrape;SSE files.changed。
export const SCRAPE_TASK = "scrapeMedia";
@@ -59,7 +65,10 @@ const IMAGE_CONCURRENCY = 4;
const DEDUPE_MS = 30 * 60_000;
const TEMP_RE = /(\.uhd-partial|\.!qB|\.part)$/i;
export async function scrapeSettings(d: AoDeps): Promise<OrganizeSettings> {
/** 排队一侧只用到 ctx(元数据刷新后自动重刮从 sources-identity 调用) */
type QueueDeps = Pick<AoDeps, "ctx">;
export async function scrapeSettings(d: QueueDeps): Promise<OrganizeSettings> {
const s = await d.ctx.settings.get("organize").catch(() => DEFAULT_SETTINGS.organize);
return { ...DEFAULT_SETTINGS.organize, ...s };
}
@@ -83,6 +92,8 @@ export interface QueueScrapeInput extends ScrapeLocation {
idempotencyKey?: string | undefined;
/** 只刮这些视频(库根相对路径;整理后刮削用,作品 / 季目录元数据照常初始化);缺省 = 目录下全部 */
videos?: readonly string[] | undefined;
/** 任务最早开始时间(元数据刷新后的自动重刮延后执行,期间再来的刷新合进同一条排队记录) */
availableAt?: Date | undefined;
}
type Row = {
@@ -158,7 +169,7 @@ export async function latestEntityScrape(d: AoDeps, entityId: string): Promise<S
* 建一条刮削记录并排任务 scrapeMedia。同一目录已有排队中(还没开始,30 分钟内)的同方式、同作品刮削时直接复用。
*/
export async function queueScrape(
d: AoDeps,
d: QueueDeps,
input: QueueScrapeInput,
): Promise<{ recordId: string; taskId: string | null; deduplicated: boolean }> {
const db = d.ctx.db;
@@ -212,7 +223,11 @@ export async function queueScrape(
queued = await d.ctx.engine.enqueue(
SCRAPE_TASK,
{ recordId: rec.id },
{ idempotencyKey: input.idempotencyKey ?? `${SCRAPE_TASK}:${rec.id}`, traceId: randomUUID() },
{
idempotencyKey: input.idempotencyKey ?? `${SCRAPE_TASK}:${rec.id}`,
traceId: randomUUID(),
...(input.availableAt ? { availableAt: input.availableAt } : {}),
},
);
} catch (err) {
// 只有新记录的入队失败才标失败;复用的记录可能已有接受但未关联的 run。
@@ -259,11 +274,9 @@ export async function queueScrape(
// ── 作品 → 目录(作品详情「重新刮削」、入库后自动刮削) ──
const isMovieKind = (k: string) => k === "movie" || k === "collection" || k === "other";
/** 本地作品 → 识别结果(带已接受的 TMDB id) */
/** 本地作品 → 识别结果(带已接受的、与作品类型一致的 TMDB id) */
export async function entityScrapeMatch(
d: AoDeps,
d: QueueDeps,
entityId: string,
source: ScrapeMatch["source"],
): Promise<ScrapeMatch | null> {
@@ -273,12 +286,15 @@ export async function entityScrapeMatch(
.where("id", "=", entityId)
.executeTakeFirst();
if (!e || e.status === "merged") return null;
const kind = scrapeKindOf(e.kind);
// 同一作品可同时有 movie 与 tv 两个 accepted 映射:只取与作品类型一致的那个
const tmdb = await d.ctx.db
.selectFrom("external_identity")
.select(["external_id"])
.where("media_entity_id", "=", entityId)
.where("content_unit_id", "is", null)
.where("provider", "=", "tmdb")
.where("entity_type", "=", kind)
.where("status", "=", "accepted")
.orderBy("updated_at", "desc")
.executeTakeFirst();
@@ -286,7 +302,7 @@ export async function entityScrapeMatch(
mediaEntityId: e.id,
provider: tmdb ? "tmdb" : null,
externalId: tmdb?.external_id ?? null,
kind: isMovieKind(e.kind) ? "movie" : "tv",
kind,
title: e.title,
year: e.year,
source,
@@ -298,7 +314,7 @@ export async function entityScrapeMatch(
* 推不出(视频直接在库根)时对视频文件本身刮削。只取启用的库根里 active 视频的位置。
*/
export async function entityScrapeLocations(
d: AoDeps,
d: QueueDeps,
entityId: string,
kind: "movie" | "tv",
): Promise<ScrapeLocation[]> {
@@ -348,13 +364,14 @@ export async function entityScrapeLocations(
/** 对作品的目录刮削(每个目录一条记录);没有已入库文件时返回空 */
export async function scrapeEntity(
d: AoDeps,
d: QueueDeps,
entityId: string,
opts: {
mode?: ScrapeMode | undefined;
trigger: "auto" | "manual";
organizationRecordId?: string | undefined;
actor?: { type: string; id: string | null } | undefined;
availableAt?: Date | undefined;
},
): Promise<{ recordId: string; taskId: string | null; deduplicated: boolean }[]> {
const match = await entityScrapeMatch(d, entityId, "library");
@@ -370,6 +387,7 @@ export async function scrapeEntity(
trigger: opts.trigger,
organizationRecordId: opts.organizationRecordId,
actor: opts.actor,
availableAt: opts.availableAt,
idempotencyKey: opts.organizationRecordId
? `${SCRAPE_TASK}:record:${opts.organizationRecordId}:${i}`
: undefined,
@@ -378,6 +396,31 @@ export async function scrapeEntity(
return out;
}
/** 元数据刷新后自动重刮的延迟:这段时间内再来的刷新合进同一条排队记录(防抖) */
export const RESCRAPE_DELAY_MS = 2 * 60_000;
/**
* 元数据刷新(或改身份后的刷新)改变了作品的 NFO 字段:对刮削过的作品(有成功 / 部分成功的刮削记录)排一次
* 「更新」模式重刮,延后 RESCRAPE_DELAY_MS 执行;还没开始的同目录更新刮削直接复用(queueScrape 合批)。
* 返回排上(含复用)的记录数;没刮削过的作品不排。
*/
export async function scheduleEntityRescrape(d: QueueDeps, entityId: string): Promise<number> {
const scraped = await d.ctx.db
.selectFrom("scrape_record")
.select("id")
.where("media_entity_id", "=", entityId)
.where("status", "in", ["succeeded", "partial"])
.limit(1)
.executeTakeFirst();
if (!scraped) return 0;
const out = await scrapeEntity(d, entityId, {
mode: "update",
trigger: "auto",
availableAt: new Date(Date.now() + RESCRAPE_DELAY_MS),
});
return out.length;
}
/** 整理入库成功后(设置打开或本次要求刮削):照 MP 对作品目录刮削,集只刮本次整理的文件。 */
export async function scrapeAfterImport(
d: AoDeps,
@@ -552,6 +595,50 @@ async function coverageOf(
return out;
}
const sha256 = (c: Uint8Array | string) => createHash("sha256").update(c).digest("hex");
/** 旧版本写的文件没有 hash:修改时间不晚于那次刮削结束(加这点余量)就当没被改过 */
const LEGACY_MTIME_SLACK_MS = 2_000;
const missingMeta = (t: ScrapeTarget): ScrapeFileResult => ({
path: t.path,
kind: t.kind,
role: t.role,
action: "skipped",
reason: "scrape_metadata_missing",
});
interface PrevWrite {
/** 写入时的 sha256;旧版本记录没有 */
hash: string | null;
/** 那次刮削结束时间(毫秒) */
at: number;
}
/**
* 同一登记目录里以前的刮削记录 → 本工具最近一次写入(或核对未变)的文件:路径 → hash。
* 从新到旧取每个路径第一条带 hash 的结果;旧版本只有 written、没有 hash 的按结束时间兜底。
*/
async function previousWrites(d: AoDeps, rootId: string, exceptId: string): Promise<Map<string, PrevWrite>> {
const rows = await d.ctx.db
.selectFrom("scrape_record")
.select(["files", "finished_at"])
.where("root_id", "=", rootId)
.where("id", "<>", exceptId)
.where("finished_at", "is not", null)
.orderBy("finished_at", "desc")
.limit(500)
.execute();
const out = new Map<string, PrevWrite>();
for (const r of rows) {
const at = new Date(r.finished_at!).getTime();
for (const f of (r.files ?? []) as ScrapeFileResult[]) {
if (out.has(f.path)) continue;
if (f.hash) out.set(f.path, { hash: f.hash, at });
else if (f.action === "written") out.set(f.path, { hash: null, at });
}
}
return out;
}
interface Fields {
entity: EntityForNfo;
ids: IdentityMap;
@@ -579,6 +666,8 @@ export async function runScrapeRecord(
const mode = rec.mode as ScrapeMode;
let match = (rec.match ?? null) as ScrapeMatch | null;
const dirs = new Set<string>();
/** 逐文件结果;兜底异常时也保留已经写了的 */
const files: ScrapeFileResult[] = [];
const finish = async (r: {
status: ScrapeRecord["status"];
@@ -681,7 +770,7 @@ export async function runScrapeRecord(
const ids: IdentityMap = {};
const idRows = await db
.selectFrom("external_identity")
.select(["provider", "external_id"])
.select(["provider", "external_id", "entity_type"])
.where("media_entity_id", "=", e.id)
.where("content_unit_id", "is", null)
.where("status", "=", "accepted")
@@ -689,6 +778,8 @@ export async function runScrapeRecord(
.execute();
for (const r of idRows) {
const p = r.provider.toLowerCase() as keyof IdentityMap;
// TMDB 映射只认与刮削类型一致的(电影 / 剧集各一个时不串)
if (p === "tmdb" && r.entity_type !== kind) continue;
if (["tmdb", "imdb", "tvdb", "douban"].includes(p) && !ids[p]) ids[p] = r.external_id;
}
if (!match.externalId && ids.tmdb) match = { ...match, provider: "tmdb", externalId: ids.tmdb };
@@ -757,6 +848,7 @@ export async function runScrapeRecord(
}
// ── 没有本地字段:按 TMDB 条目直接取 ──
let fromTmdb = false;
if (!fields) {
if (!match.externalId) return await fail("scrape_no_metadata");
let detail: Awaited<ReturnType<typeof d.ports.scrapeTmdbDetail>>;
@@ -770,22 +862,38 @@ export async function runScrapeRecord(
const t = tmdbEntity(detail);
fields = { entity: t.entity, ids: t.ids, seasonUnit: new Map(), episodeUnit: new Map() };
match = { ...match, provider: "tmdb", title: detail.title, year: detail.year };
if (kind === "tv") {
const seasons = [...new Set(videos.map((v) => v.season).filter((s): s is number => typeof s === "number"))];
for (const n of seasons) {
const sd = await d.ports.scrapeTmdbSeason(match.externalId!, n).catch((err) => {
d.ctx.logger.warn({ recordId, season: n, err: { message: errMessage(err) } }, "刮削取 TMDB 季详情失败");
return null;
});
if (!sd) continue;
const u = tmdbSeasonUnits(sd);
fields.seasonUnit.set(n, u.season);
for (const [ep, eu] of u.episodes) fields.episodeUnit.set(`${n}:${ep}`, eu);
fromTmdb = true;
}
// ── 季 / 集元数据:TMDB 分支取每个涉及的季;本地分支只在季 / 集单元缺元数据时按 TMDB id 现取补齐 ──
const { seasonUnit, episodeUnit } = fields;
if (kind === "tv") {
const tmdbId = match.externalId ?? fields.ids.tmdb ?? null;
const seasons = [...new Set(videos.map((v) => v.season).filter((s): s is number => typeof s === "number"))];
for (const n of seasons.sort((a, b) => a - b)) {
const lacking =
(lay.parts.season && seasonMetadataMissing(seasonUnit.get(n))) ||
videos.some(
(v) =>
v.season === n &&
typeof v.episode === "number" &&
episodeMetadataMissing(episodeUnit.get(`${n}:${v.episode}`)),
);
if (!tmdbId || (!fromTmdb && !lacking)) continue;
const sd = await d.ports.scrapeTmdbSeason(tmdbId, n).catch((err) => {
d.ctx.logger.warn({ recordId, season: n, err: { message: errMessage(err) } }, "刮削取 TMDB 季详情失败");
return null;
});
if (!sd) continue;
const u = tmdbSeasonUnits(sd);
// 本地已有的字段(含人工覆盖)不动,只补缺元数据的单元
if (fromTmdb || seasonMetadataMissing(seasonUnit.get(n))) seasonUnit.set(n, u.season);
for (const [ep, eu] of u.episodes) {
const k = `${n}:${ep}`;
if (fromTmdb || episodeMetadataMissing(episodeUnit.get(k))) episodeUnit.set(k, eu);
}
}
}
const { seasonUnit, episodeUnit } = fields;
const work = nfoWork(fields.entity, fields.ids);
const seasonPosters = new Map<number, string | null>();
for (const [n, u] of seasonUnit) seasonPosters.set(n, unitPosterPath(u.metadata));
@@ -800,7 +908,9 @@ export async function runScrapeRecord(
options: { images: settings.scrapeImages, episodeThumbs: settings.scrapeEpisodeThumbs },
parts: lay.parts,
});
const files: ScrapeFileResult[] = [...noEpisode];
files.push(...noEpisode);
// 「更新」模式:本工具以前写过的文件(路径 → 写入时的 hash / 时间)
const ours = mode === "update" ? await previousWrites(d, rec.root_id, recordId) : new Map<string, PrevWrite>();
const imageCache = new Map<string, Promise<Uint8Array>>();
const fetchImage = (p: string) => {
let hit = imageCache.get(p);
@@ -810,21 +920,37 @@ export async function runScrapeRecord(
}
return hit;
};
const decide = async (t: ScrapeTarget): Promise<{ abs: string } | ScrapeFileResult> => {
type Decision = { abs: string; prevHash: string | null } | ScrapeFileResult;
const decide = async (t: ScrapeTarget): Promise<Decision> => {
const target = resolve(rootReal, t.path);
const base = { path: t.path, kind: t.kind, role: t.role };
if (!inside(rootReal, target)) return { ...base, action: "failed", reason: "path_escape" };
const ex = await lstat(target).catch(() => null);
if (ex && (!ex.isFile() || ex.isSymbolicLink())) return { ...base, action: "skipped", reason: "scrape_exists" };
if (ex && mode === "missing") return { ...base, action: "skipped", reason: "scrape_exists" };
return { abs: target };
if (!ex) return { abs: target, prevHash: null };
if (!ex.isFile() || ex.isSymbolicLink()) return { ...base, action: "skipped", reason: "scrape_exists" };
if (mode === "missing") return { ...base, action: "skipped", reason: "scrape_exists" };
if (mode === "overwrite") return { abs: target, prevHash: null };
// update:只覆盖本工具写过且之后没被改过的
const prev = ours.get(t.path);
if (!prev) return { ...base, action: "skipped", reason: "scrape_exists" };
const cur = sha256(await readFile(target));
const untouched = prev.hash ? prev.hash === cur : ex.mtimeMs <= prev.at + LEGACY_MTIME_SLACK_MS;
if (!untouched) return { ...base, action: "skipped", reason: "scrape_user_modified" };
return { abs: target, prevHash: cur };
};
const write = async (t: ScrapeTarget, target: string, content: Uint8Array | string): Promise<ScrapeFileResult> => {
const write = async (
t: ScrapeTarget,
dec: { abs: string; prevHash: string | null },
content: Uint8Array | string,
): Promise<ScrapeFileResult> => {
const base = { path: t.path, kind: t.kind, role: t.role };
const hash = sha256(content);
// 更新模式下内容没变就不重写(不打扰媒体服务器)
if (dec.prevHash === hash) return { ...base, action: "skipped", reason: "scrape_unchanged", hash };
try {
await atomicWrite(target, content);
await atomicWrite(dec.abs, content);
dirs.add(dirOf(t.path));
return { ...base, action: "written", reason: null };
return { ...base, action: "written", reason: null, hash };
} catch (err) {
d.ctx.logger.warn({ recordId, path: t.path, err: { message: errMessage(err) } }, "刮削文件写入失败");
return { ...base, action: "failed", reason: "scrape_write_failed" };
@@ -841,9 +967,22 @@ export async function runScrapeRecord(
}
let xml: string;
if (t.role === "movie_nfo" || t.role === "tvshow_nfo") xml = workNfo(work);
else if (t.role === "season_nfo") xml = seasonNfo(nfoSeason(t.season!, seasonUnit.get(t.season!)));
else xml = episodeNfo(nfoEpisode(t.season!, t.episode!, episodeUnit.get(`${t.season}:${t.episode}`)));
files.push(await write(t, dec.abs, xml));
else if (t.role === "season_nfo") {
const u = seasonUnit.get(t.season!);
if (seasonMetadataMissing(u)) {
files.push(missingMeta(t));
continue;
}
xml = seasonNfo(nfoSeason(t.season!, u));
} else {
const u = episodeUnit.get(`${t.season}:${t.episode}`);
if (episodeMetadataMissing(u)) {
files.push(missingMeta(t));
continue;
}
xml = episodeNfo(nfoEpisode(t.season!, t.episode!, u));
}
files.push(await write(t, dec, xml));
}
// 图片:限并发,失败不影响 NFO
const imgs = targets.filter((x) => x.kind === "image" && x.imagePath);
@@ -859,23 +998,37 @@ export async function runScrapeRecord(
d.ctx.logger.warn({ recordId, image: t.imagePath, err: { message: errMessage(err) } }, "刮削图片下载失败");
return { path: t.path, kind: t.kind, role: t.role, action: "failed", reason: "scrape_image_failed" };
}
return write(t, dec.abs, bytes);
return write(t, dec, bytes);
})),
);
const written = files.filter((f) => f.action === "written").length;
const failed = files.filter((f) => f.action === "failed").length;
// 缺元数据跳过的季 / 集不算成功(要刷新元数据后再刮)
const lacking = files.some((f) => f.reason === "scrape_metadata_missing");
const status: ScrapeRecord["status"] =
failed === 0 ? "succeeded" : written > 0 || files.some((f) => f.action === "skipped") ? "partial" : "failed";
failed === 0
? lacking
? "partial"
: "succeeded"
: written > 0 || files.some((f) => f.action === "skipped")
? "partial"
: "failed";
return await finish({
status,
reason: failed ? (files.find((f) => f.action === "failed")?.reason ?? null) : null,
reason: failed
? (files.find((f) => f.action === "failed")?.reason ?? null)
: lacking
? "scrape_metadata_missing"
: null,
files,
});
} catch (err) {
if (signal?.aborted) throw err;
d.ctx.logger.warn({ recordId, err: { message: errMessage(err) } }, "刮削失败");
return fail("scrape_write_failed");
// 不是写文件的错(如查库失败):记内部错误,已写的文件结果照留
const anyWritten = files.some((f) => f.action === "written");
return finish({ status: anyWritten ? "partial" : "failed", reason: "scrape_internal_error", files });
}
}
@@ -29,6 +29,7 @@ import {
settleEpisodeDecision,
} from "./episode-resolution.ts";
import { detachFromIdentityDecisions } from "./identity-decisions.ts";
import { clearTmdbMetadata } from "./metadata.ts";
import { metadataLanguages, metadataTextOptions } from "./metadata-rich.ts";
import { recordAttempt } from "./resolve.ts";
import { backfillRecognizeCache } from "./tmdb-recognize-cache.ts";
@@ -271,10 +272,12 @@ async function retireIdentity(exec: Exec, row: ExternalIdentityRow, actor: Actor
/**
* 改作品的外部映射:同 provider+类型的旧 accepted 映射退役(留修订并锁定,防止自动匹配改回),
* 新映射 accepted + 人工锁。目标身份已被其它作品接受时 409(应走合并)。
* TMDB 映射变了:同一事务里清掉该作品(及季 / 集单元)tmdb 来源的元数据(人工覆盖 / 锁定保留),提交后排一次
* full 刷新;刷新写回新身份的字段后,刮削过的作品自动排「更新」模式重刮(refreshMetadata → scheduleEntityRescrape)。
*/
export async function setEntityIdentity(rt: SourcesRuntime, entityId: string, input: EntityIdentitySet, actor: Actor) {
const { ctx } = rt;
return ctx.db.transaction().execute(async (trx) => {
const r = await ctx.db.transaction().execute(async (trx) => {
const id = await resolveEntityId(trx, entityId);
if (!id) throw new ApiException("not_found", "作品不存在");
const other = await entityByExternal(trx, input);
@@ -297,6 +300,21 @@ export async function setEntityIdentity(rt: SourcesRuntime, entityId: string, in
reason: input.reason ?? null,
});
if (!newId) throw new ApiException("conflict", "外部身份唯一约束冲突");
if (input.provider === "tmdb") {
// 旧 TMDB 带出来的 IMDb 映射(没人工锁的)一并退役:NFO 里 IMDb 是 default uniqueid,留着会被认成旧作品;
// 刷新后 NFO 取新详情里的 IMDb
const derived = await trx
.selectFrom("external_identity")
.selectAll()
.where("media_entity_id", "=", id)
.where("content_unit_id", "is", null)
.where("provider", "=", "imdb")
.where("status", "=", "accepted")
.where("locked", "=", false)
.execute();
for (const o of derived) await retireIdentity(trx, o, actor, "tmdb_mapping_changed");
await clearTmdbMetadata(trx, id, rt.now());
}
await audit(
trx,
actor,
@@ -310,6 +328,20 @@ export async function setEntityIdentity(rt: SourcesRuntime, entityId: string, in
);
return { mediaEntityId: id, changed: true, identityId: newId, retired: olds.map((o) => o.id) };
});
if (r.changed && input.provider === "tmdb" && r.identityId)
await ctx.engine
.enqueue(
"refreshMetadata",
{ mediaEntityId: r.mediaEntityId, full: true },
{ idempotencyKey: `refreshMetadata:identity:${r.identityId}`, traceId: r.mediaEntityId },
)
.catch((err: unknown) =>
ctx.logger.warn(
{ entityId: r.mediaEntityId, err: { message: (err as Error).message } },
"改映射后元数据刷新没排上",
),
);
return r;
}
/** 外部身份字段修订(乐观并发 expectedRevision);每次变更写 identity_revision */
@@ -315,8 +315,20 @@ export interface SeasonMetadataResult {
changedUnits: number;
/** DH-1:临时转正式的集单元数(季单元转正不计) */
promoted: number;
/** 写进 NFO 的字段(名称、简介、日期、时长、剧照、季海报)有变化的单元数;刷新后据此决定要不要自动重刮 */
nfoChangedUnits?: number;
}
/** 内容单元里写进 NFO 的字段(评分 / 票数变化不触发自动重刮) */
export const UNIT_NFO_FIELDS: ReadonlySet<string> = new Set([
"name",
"overview",
"airDate",
"runtime",
"stillPath",
"posterPath",
]);
/**
* 季详情 → 季单元与集单元的字段级元数据。集单元按(作品, 季, 集号)找最早建的未退役单元,没有时:
* create = true(full 刷新 / 追踪中的季)建 confirmed 单元;否则只更新已有单元(集单元仍由发布 / 文件懒建)。
@@ -331,7 +343,14 @@ export async function applySeasonMetadata(
opts: { create?: boolean } = {},
): Promise<SeasonMetadataResult> {
const create = opts.create ?? true;
const out: SeasonMetadataResult = { season: sd.seasonNumber, episodes: 0, created: 0, changedUnits: 0, promoted: 0 };
const out: SeasonMetadataResult = {
season: sd.seasonNumber,
episodes: 0,
created: 0,
changedUnits: 0,
promoted: 0,
nfoChangedUnits: 0,
};
if (sd.seasonNumber < 0) return out;
const seasonUnit = await ensureSeason(exec, entityId, sd.seasonNumber, sd.name);
const units = await exec
@@ -399,6 +418,7 @@ export async function applySeasonMetadata(
if (written.length || t.fresh) {
const r = await projectUnit(exec, t.unitId, now);
if (r.changedFields.length) out.changedUnits++;
if (r.changedFields.some((f) => UNIT_NFO_FIELDS.has(f))) out.nfoChangedUnits = (out.nfoChangedUnits ?? 0) + 1;
}
}
out.episodes = eps.length;
@@ -17,6 +17,7 @@ import type {
} from "@umt/contracts";
import { sql } from "@umt/db";
import { ApiException } from "../../../http/errors.ts";
import { scheduleEntityRescrape } from "../../acquisition-organize/services/scrape.ts";
import { jsonb } from "../rows.ts";
import type { SourcesRuntime } from "../runtime.ts";
import { type Actor, audit, type Exec } from "./common.ts";
@@ -118,6 +119,102 @@ export async function writeEnrichmentFields(
}
}
/** 作品里写进 NFO 的字段:刷新后这些字段有变化才自动重刮(评分、票数、播出状态等不算) */
const ENTITY_NFO_FIELDS: ReadonlySet<string> = new Set([
"title",
"originalTitle",
"year",
"overview",
"tagline",
"releaseDate",
"genres",
"originCountry",
"platform",
"runtime",
"cast",
"crew",
"externalIds",
"posterPath",
"backdropPath",
]);
/**
* 改作品的 TMDB 映射后:清掉该作品及其季 / 集单元的 tmdb 来源字段(metadata_field 与投影出的最终值),
* 人工覆盖与锁定字段保留;单元的 title / air_date 列随 name / airDate 一起清空。之后由 full 刷新按新身份重新填,
* 避免新身份的 NFO 混进旧作品的标题、简介、IMDb、海报(海报 / 背景图「TMDB 没有时保留旧值」不能保留旧作品的)。
*/
export async function clearTmdbMetadata(trx: Exec, entityId: string, now: Date): Promise<void> {
const keep = (meta: Record<string, FinalFieldValue>, locks: Set<string>, overridden: Set<string>) => {
const out: Record<string, FinalFieldValue> = {};
for (const [k, v] of Object.entries(meta))
if (v?.source !== "tmdb" || locks.has(k) || overridden.has(k) || v.locked) out[k] = v;
return out;
};
const e = await trx
.selectFrom("media_entity")
.select(["metadata", "field_locks"])
.where("id", "=", entityId)
.forUpdate()
.executeTakeFirstOrThrow();
const entityOverrides = await trx
.selectFrom("field_override")
.select("field")
.where("media_entity_id", "=", entityId)
.execute();
await trx
.deleteFrom("metadata_field")
.where("media_entity_id", "=", entityId)
.where("content_unit_id", "is", null)
.where("source", "=", "tmdb")
.execute();
await trx
.updateTable("media_entity")
.set({
metadata: jsonb(
keep(
(e.metadata ?? {}) as Record<string, FinalFieldValue>,
new Set(e.field_locks),
new Set(entityOverrides.map((o) => o.field)),
),
),
updated_at: now,
})
.where("id", "=", entityId)
.execute();
// 其它来源(web / agent 补全)按原规则重新投影
await project(trx, entityId, now);
const units = await trx
.selectFrom("content_unit")
.select(["id", "metadata", "field_locks"])
.where("media_entity_id", "=", entityId)
.execute();
if (!units.length) return;
const ids = units.map((u) => u.id);
const unitOverrides = await trx
.selectFrom("field_override")
.select(["content_unit_id", "field"])
.where("content_unit_id", "in", ids)
.execute();
await trx.deleteFrom("metadata_field").where("content_unit_id", "in", ids).where("source", "=", "tmdb").execute();
for (const u of units) {
const meta = (u.metadata ?? {}) as Record<string, FinalFieldValue>;
const overridden = new Set(unitOverrides.filter((o) => o.content_unit_id === u.id).map((o) => o.field));
const next = keep(meta, new Set(u.field_locks), overridden);
if (Object.keys(next).length === Object.keys(meta).length) continue;
await trx
.updateTable("content_unit")
.set({
metadata: jsonb(next),
...(next.name ? {} : { title: null }),
...(next.airDate ? {} : { air_date: null }),
updated_at: now,
})
.where("id", "=", u.id)
.execute();
}
}
export interface RefreshResult {
status: "refreshed" | "skipped" | "failed";
reason?: string;
@@ -240,7 +337,7 @@ export async function refreshMetadata(
}
const now = rt.now();
const rev = `tmdb:${ident.entity_type}/${ident.external_id}`;
return ctx.db.transaction().execute(async (trx) => {
const result = await ctx.db.transaction().execute(async (trx) => {
if (richDetail) await writeRichEntityFields(trx, id, richDetail, rev, now);
else await writeTmdbFields(trx, id, detail, rev, now);
await trx
@@ -282,6 +379,15 @@ export async function refreshMetadata(
: {}),
};
});
// 写进 NFO 的字段有变化:刮削过的作品延后排一次「更新」模式重刮(期间再刷新合进同一条);失败只记日志
const nfoChanged =
result.changedFields.some((f) => ENTITY_NFO_FIELDS.has(f)) ||
(result.episodes ?? []).some((x) => (x.nfoChangedUnits ?? 0) > 0);
if (nfoChanged)
await scheduleEntityRescrape({ ctx }, id).catch((err: unknown) =>
ctx.logger.warn({ entityId: id, err: { message: (err as Error).message } }, "元数据刷新后重刮没排上"),
);
return result;
}
/**