diff --git a/apps/server/src/modules/sources-identity/index.ts b/apps/server/src/modules/sources-identity/index.ts index 457b32f1..8fc18caf 100644 --- a/apps/server/src/modules/sources-identity/index.ts +++ b/apps/server/src/modules/sources-identity/index.ts @@ -9,6 +9,7 @@ import { workbenchRoutes } from "./routes/workbench.ts"; import { type RuntimeOverrides, SourcesRuntime } from "./runtime.ts"; import { readOnlyReconcile, tickFeeds, tickReviews, tickSnapshots } from "./scheduler.ts"; import { tickSiteHealth, tickSiteUserData } from "./service/account.ts"; +import { BACKFILL_LANGUAGES_JOB, backfillLanguages } from "./service/backfill-languages.ts"; import { registerSourcesTasks, sourcesOutboxRoutes } from "./tasks.ts"; /** @@ -89,6 +90,23 @@ export function createSourcesIdentityModule(overrides: RuntimeOverrides = {}): S }, run: async () => ({ summary: `${await tickSiteHealth(rt)} 个站点` }), }); + // FIX-5 存量元数据回刷:部署后首次启动即运行,跑完自动停用;可在计划页手动再执行 + registerScheduledJob(ctx, { + key: BACKFILL_LANGUAGES_JOB, + defaults: { + name: "元数据语言回刷", + description: + "对已确认 TMDB 身份的作品按语言回退链完整刷新一次(标题、简介、每集信息),TMDB 每秒最多 2 次请求;跑完自动停用,可手动再执行", + group: "maintain", + schedule: { type: "interval", everyMs: 24 * 3_600_000 }, + firstRun: "now", + }, + run: async ({ runId, jobId }) => { + if (await readOnlyReconcile(rt)) return { status: "skipped", summary: "只读对账模式" }; + const r = await backfillLanguages(rt, { runId, jobId, disableWhenDone: true }); + return { status: r.status, summary: r.summary }; + }, + }); }, }; } diff --git a/apps/server/src/modules/sources-identity/runtime.ts b/apps/server/src/modules/sources-identity/runtime.ts index 917afda7..7c096ad7 100644 --- a/apps/server/src/modules/sources-identity/runtime.ts +++ b/apps/server/src/modules/sources-identity/runtime.ts @@ -79,6 +79,38 @@ export class SourcesRuntime { if (this.tmdbClient === undefined) this.tmdbClient = this.overrides.tmdb(); return this.tmdbClient; } + const { sig, cfg } = await this.tmdbConfig(); + if (this.tmdbClient !== undefined && this.tmdbSig === sig) return this.tmdbClient; + this.tmdbSig = sig; + this.tmdbClient = cfg ? createTmdbClient(cfg) : null; + return this.tmdbClient; + } + + /** + * 批量任务用的独立 TMDB 客户端(FIX-5 元数据回刷):串行、网络请求间隔 ≥ minIntervalMs(缓存命中不占额度), + * 同样走 PG 请求缓存;requests() 返回实际发出的网络请求数。测试注入的客户端原样返回(requests 为 null)。 + */ + async tmdbForBatch(opts: { + minIntervalMs: number; + }): Promise<{ client: TmdbClient; requests: () => number | null } | null> { + if (this.overrides.tmdb) { + const client = await this.tmdb(); + return client ? { client, requests: () => null } : null; + } + const { cfg } = await this.tmdbConfig(); + if (!cfg) return null; + let n = 0; + const base = cfg.fetch ?? globalThis.fetch; + const counted: typeof fetch = (input, init) => { + n++; + return base(input, init); + }; + const client = createTmdbClient({ ...cfg, fetch: counted, maxConcurrent: 1, minIntervalMs: opts.minIntervalMs }); + return { client, requests: () => n }; + } + + /** TMDB 客户端配置(未配置凭证时 cfg 为 null)与签名(设置变化时重建客户端) */ + private async tmdbConfig(): Promise<{ sig: string; cfg: Parameters[0] | null }> { const meta = await this.ctx.settings.get("metadata").catch(() => null); const policy = await tmdbCachePolicy(this.ctx); const env = (k: string) => Deno.env.get(k); @@ -90,7 +122,6 @@ export class SourcesRuntime { policy, ref ? null : [!!env("TMDB_API_KEY"), !!env("TMDB_ACCESS_TOKEN")], ]); - if (this.tmdbClient !== undefined && this.tmdbSig === sig) return this.tmdbClient; let apiKey: string | undefined; let accessToken: string | undefined; if (ref) { @@ -106,10 +137,9 @@ export class SourcesRuntime { apiKey = env("TMDB_API_KEY"); accessToken = env("TMDB_ACCESS_TOKEN"); } - this.tmdbSig = sig; - this.tmdbClient = + const cfg = apiKey || accessToken - ? createTmdbClient({ + ? { ...(apiKey ? { apiKey } : {}), ...(accessToken ? { accessToken } : {}), ...(baseUrl ? { baseUrl } : {}), @@ -121,9 +151,9 @@ export class SourcesRuntime { emptyTtlMs: policy.emptyTtl * 1000, }, sharedCache: tmdbSharedCache(this.ctx, policy), - }) + } : null; - return this.tmdbClient; + return { sig, cfg }; } /** 站点默认(settings.sites),读库并刷新缓存;读失败用上次值 */ diff --git a/apps/server/src/modules/sources-identity/service/backfill-languages.ts b/apps/server/src/modules/sources-identity/service/backfill-languages.ts new file mode 100644 index 00000000..444e77cc --- /dev/null +++ b/apps/server/src/modules/sources-identity/service/backfill-languages.ts @@ -0,0 +1,90 @@ +import { sql } from "@umt/db"; +import type { SourcesRuntime } from "../runtime.ts"; +import { refreshMetadata } from "./metadata.ts"; + +// FIX-5 存量元数据回刷(维护任务 metadata.backfill_languages):MD-1 语言回退链上线前建档的作品,标题 / 简介 / 每集信息 +// 仍是旧语言。对所有已确认 TMDB 身份的作品按当前语言链跑一次 full 刷新(同 POST /media-entities/{id}/refresh?full=1)。 +// - 限速:独立 TMDB 客户端串行、网络请求间隔 ≥ 500 ms(每秒最多 2 次);请求走 PG 请求缓存(对照 MP @cached), +// 命中不发请求、不占额度,中断后重跑已取过的作品几乎不再请求 TMDB。 +// - 进度写本次运行记录的摘要(scheduled_job_run.summary),结束时写最终统计。 +// - 跑完(不论个别失败)把任务停用(customized,之后同步定义不会再打开);手动「立即执行」可再跑一次。 + +export const BACKFILL_LANGUAGES_JOB = "metadata.backfill_languages"; +/** TMDB 网络请求最小间隔:每秒最多 2 次 */ +export const BACKFILL_MIN_INTERVAL_MS = 500; +const PROGRESS_EVERY_MS = 5_000; + +export interface BackfillResult { + status: "ok" | "skipped"; + total: number; + refreshed: number; + skipped: number; + failed: number; + requests: number | null; + errors: { mediaEntityId: string; title: string; error: string }[]; + summary: string; +} + +export async function backfillLanguages( + rt: SourcesRuntime, + opts: { runId?: string; jobId?: string; minIntervalMs?: number; disableWhenDone?: boolean } = {}, +): Promise { + const { ctx } = rt; + const empty = { total: 0, refreshed: 0, skipped: 0, failed: 0, requests: null, errors: [] }; + const batch = await rt.tmdbForBatch({ minIntervalMs: opts.minIntervalMs ?? BACKFILL_MIN_INTERVAL_MS }); + if (!batch) return { status: "skipped", ...empty, summary: "TMDB 未配置,未执行(配置后可手动执行)" }; + const rows = await sql<{ id: string; title: string }>` + select e.id, e.title from media_entity e + where e.status in ('active', 'provisional') + and exists (select 1 from external_identity x where x.media_entity_id = e.id and x.provider = 'tmdb' + and x.status = 'accepted' and x.entity_type in ('movie', 'tv')) + order by e.created_at, e.id`.execute(ctx.db); + const total = rows.rows.length; + const r: BackfillResult = { status: "ok", ...empty, total, summary: "" }; + let lastProgress = Date.now(); + const writeProgress = async (done: number) => { + if (!opts.runId) return; + await ctx.db + .updateTable("scheduled_job_run") + .set({ summary: `进度 ${done}/${total}:刷新 ${r.refreshed},跳过 ${r.skipped},失败 ${r.failed}` }) + .where("id", "=", opts.runId) + .execute() + .catch(() => undefined); + }; + for (const [i, e] of rows.rows.entries()) { + try { + const res = await refreshMetadata(rt, e.id, { full: true, tmdb: batch.client }); + if (res.status === "refreshed") r.refreshed++; + else if (res.status === "skipped") r.skipped++; + else { + r.failed++; + r.errors.push({ mediaEntityId: e.id, title: e.title, error: res.reason ?? "failed" }); + } + } catch (err) { + r.failed++; + r.errors.push({ mediaEntityId: e.id, title: e.title, error: (err as Error).message.slice(0, 300) }); + ctx.logger.warn({ entityId: e.id, err: { message: (err as Error).message } }, "元数据回刷失败"); + } + if (Date.now() - lastProgress >= PROGRESS_EVERY_MS) { + lastProgress = Date.now(); + await writeProgress(i + 1); + } + } + r.requests = batch.requests(); + const failedList = r.errors + .slice(0, 5) + .map((x) => `「${x.title}」${x.error}`) + .join(";"); + r.summary = + `${total} 部作品:刷新 ${r.refreshed},跳过 ${r.skipped},失败 ${r.failed}` + + (r.requests !== null ? `;TMDB 请求 ${r.requests} 次` : "") + + (failedList ? `。失败:${failedList}${r.errors.length > 5 ? " 等" : ""}` : ""); + if (opts.disableWhenDone && opts.jobId) { + await ctx.db + .updateTable("scheduled_job") + .set({ enabled: false, customized: true }) + .where("id", "=", opts.jobId) + .execute(); + } + return r; +} diff --git a/apps/server/src/modules/sources-identity/service/metadata.ts b/apps/server/src/modules/sources-identity/service/metadata.ts index f6192976..d9650715 100644 --- a/apps/server/src/modules/sources-identity/service/metadata.ts +++ b/apps/server/src/modules/sources-identity/service/metadata.ts @@ -2,6 +2,7 @@ import { fetchRichDetail, fetchRichSeason, type MetadataCandidate, + type TmdbClient, TmdbError, type TmdbRichDetail, type TmdbRichSeasonDetail, @@ -129,6 +130,8 @@ export interface RefreshResult { export interface RefreshOptions { /** true = 拉全部正片季的每集信息(POST /media-entities/{id}/refresh?full=1);默认只拉追踪中 / 库里已有 / 最新的季 */ full?: boolean; + /** 指定 TMDB 客户端(批量回刷用限速客户端);缺省 rt.tmdb() */ + tmdb?: TmdbClient; } export async function refreshMetadata( @@ -146,7 +149,7 @@ export async function refreshMetadata( .where("provider", "=", "tmdb") .where("status", "=", "accepted") .executeTakeFirst(); - const tmdb = await rt.tmdb(); + const tmdb = opts.tmdb ?? (await rt.tmdb()); if (!ident || !tmdb || (ident.entity_type !== "movie" && ident.entity_type !== "tv")) { return { status: "skipped", diff --git a/apps/server/test/sources-identity/backfill-languages.test.ts b/apps/server/test/sources-identity/backfill-languages.test.ts new file mode 100644 index 00000000..764190dd --- /dev/null +++ b/apps/server/test/sources-identity/backfill-languages.test.ts @@ -0,0 +1,125 @@ +import { createMockTmdb, type MockTmdb } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { registerScheduledJob, scheduleRegistry } from "../../src/modules/schedules/registry.ts"; +import { syncDefinitions } from "../../src/modules/schedules/store.ts"; +import { SourcesRuntime } from "../../src/modules/sources-identity/runtime.ts"; +import { + BACKFILL_LANGUAGES_JOB, + backfillLanguages, +} from "../../src/modules/sources-identity/service/backfill-languages.ts"; +import { createEntityFromTmdbCandidate } from "../../src/modules/sources-identity/service/entity-from-candidate.ts"; +import { createHarness, type Harness, pgAvailable } from "./harness.ts"; + +// FIX-5 存量元数据回刷(metadata.backfill_languages) + +const up = await pgAvailable(); + +describe.skipIf(!up)("元数据语言回刷", () => { + let h: Harness; + let tmdb: MockTmdb; + const ids: string[] = []; + const actor = { type: "user" as const, id: "tester" }; + + beforeAll(async () => { + tmdb = await createMockTmdb(); + h = await createHarness("backfill", { tmdbUrl: tmdb.url }); + for (const [externalId, kind] of [ + ["95396", "tv"], + ["872585", "movie"], + ] as const) { + const r = await createEntityFromTmdbCandidate(h.db, () => h.rt.tmdb(), { externalId, kind }, actor); + ids.push(r.mediaEntityId); + } + // 没有 TMDB 身份的作品不参与 + await h.db + .insertInto("media_entity") + .values({ kind: "series", title: "本地短剧", status: "provisional" } as never) + .execute(); + }); + afterAll(async () => { + await h?.close(); + await tmdb?.close(); + }); + + it("登记为维护组任务,首次登记即到期;跑完按新语言链刷新并自动停用", async () => { + registerScheduledJob(h.ctx, { + key: BACKFILL_LANGUAGES_JOB, + defaults: { + name: "元数据语言回刷", + group: "maintain", + schedule: { type: "interval", everyMs: 86_400_000 }, + firstRun: "now", + }, + run: async () => undefined, + }); + const now = new Date(); + await syncDefinitions(h.db, scheduleRegistry(h.ctx).list(), now); + const before = await h.db + .selectFrom("scheduled_job") + .selectAll() + .where("id", "=", BACKFILL_LANGUAGES_JOB) + .executeTakeFirstOrThrow(); + expect(before).toMatchObject({ enabled: true, board_group: "maintain" }); + expect(before.next_run_at!.getTime()).toBeLessThanOrEqual(now.getTime()); + + const r = await backfillLanguages(h.rt, { jobId: BACKFILL_LANGUAGES_JOB, disableWhenDone: true, minIntervalMs: 0 }); + expect(r).toMatchObject({ status: "ok", total: 2, refreshed: 2, failed: 0 }); + expect(r.summary).toContain("2 部作品:刷新 2"); + const tv = await h.db.selectFrom("media_entity").selectAll().where("id", "=", ids[0]!).executeTakeFirstOrThrow(); + expect(tv.title).toBe("人生切割术"); + const eps = await h.db + .selectFrom("content_unit") + .select(["episode_number", "title"]) + .where("media_entity_id", "=", ids[0]!) + .where("kind", "=", "episode") + .where("season_number", "=", 1) + .orderBy("episode_number") + .execute(); + expect(eps.length).toBeGreaterThan(0); + + const after = await h.db + .selectFrom("scheduled_job") + .selectAll() + .where("id", "=", BACKFILL_LANGUAGES_JOB) + .executeTakeFirstOrThrow(); + expect(after).toMatchObject({ enabled: false, customized: true }); + // 重启后同步定义不会重新打开 + await syncDefinitions(h.db, scheduleRegistry(h.ctx).list(), new Date()); + const synced = await h.db + .selectFrom("scheduled_job") + .select(["enabled"]) + .where("id", "=", BACKFILL_LANGUAGES_JOB) + .executeTakeFirstOrThrow(); + expect(synced.enabled).toBe(false); + }); + + it("按设置建的客户端:网络请求限速,走 PG 请求缓存,第二次不再请求 TMDB", async () => { + await h.db + .insertInto("settings") + .values({ key: "metadata", value: JSON.stringify({ tmdbBaseUrl: `${tmdb.url}/3` }) } as never) + .onConflict((oc) => oc.column("key").doUpdateSet({ value: JSON.stringify({ tmdbBaseUrl: `${tmdb.url}/3` }) })) + .execute(); + const prev = Deno.env.get("TMDB_API_KEY"); + Deno.env.set("TMDB_API_KEY", "test-tmdb-key"); + try { + const rt = new SourcesRuntime(h.ctx, {}); + const t0 = performance.now(); + const first = await backfillLanguages(rt, { minIntervalMs: 100 }); + const elapsed = performance.now() - t0; + expect(first).toMatchObject({ status: "ok", total: 2, refreshed: 2 }); + expect(first.requests).toBeGreaterThan(2); + expect(elapsed).toBeGreaterThanOrEqual((first.requests! - 1) * 90); + const second = await backfillLanguages(rt, { minIntervalMs: 100 }); + expect(second).toMatchObject({ refreshed: 2, requests: 0 }); + } finally { + if (prev === undefined) Deno.env.delete("TMDB_API_KEY"); + else Deno.env.set("TMDB_API_KEY", prev); + } + }); + + it("TMDB 未配置:跳过,不停用", async () => { + const rt = new SourcesRuntime(h.ctx, { tmdb: () => null }); + const r = await backfillLanguages(rt, { jobId: "x", disableWhenDone: true }); + expect(r.status).toBe("skipped"); + }); +}); diff --git a/docs/runbook/metadata.md b/docs/runbook/metadata.md index 810675a5..b21f83ec 100644 --- a/docs/runbook/metadata.md +++ b/docs/runbook/metadata.md @@ -21,7 +21,13 @@ - 新建档时标题 / 简介当场按语言链取(富详情失败沿用基础详情),每集信息由随后入队的 full 刷新补齐,不用手动刷新。自动识别(RSS)新建的作品不入队 full,仍按追踪刷新。 - 刷新是后台任务 `refreshMetadata`(输入 `{mediaEntityId, full?}`),结果里 `titles` 是本次取到的标题 / 语言 / 中文名,`episodes[]` 是各季写入统计(`created` 新建单元数、`changedUnits` 有变化的单元数)。 - 增量:来源值和语言都没变的字段不写;只投影有变化的单元。不改编号、状态、路径绑定和资产覆盖。 -- 现有作品补齐中文名 / 每集信息:对作品跑一次 `full=1`。批量: +- 存量回刷(FIX-5):维护任务 `metadata.backfill_languages`「元数据语言回刷」在部署后首次启动即运行,对所有已确认 TMDB 身份 + (active / provisional)的作品逐个跑 `full` 刷新。用独立的 TMDB 客户端串行请求,网络请求间隔 ≥ 500 ms(每秒最多 2 次), + 请求照常走 `tmdb.request` PG 缓存(对照 MP `@cached`),中断后重跑时已取过的不再请求。进度每 5 秒写进本次运行记录的摘要, + 结束写「N 部作品:刷新 / 跳过 / 失败;TMDB 请求 X 次」和前 5 条失败原因。跑完(有个别失败也算)任务自动停用; + 计划页「立即执行」可再跑一次。TMDB 未配置时记为跳过、保持启用,第二天再试。只读对账模式下跳过。 + 代码:`apps/server/src/modules/sources-identity/service/backfill-languages.ts`。 +- 单部作品补齐中文名 / 每集信息:对作品跑一次 `full=1`。手动批量: ```sh # 所有有 TMDB 身份的作品各跑一次(TMDB 请求走缓存;数量多时分批) diff --git a/docs/runbook/schedules.md b/docs/runbook/schedules.md index 9931696c..285351f5 100644 --- a/docs/runbook/schedules.md +++ b/docs/runbook/schedules.md @@ -36,6 +36,7 @@ | `cache.images_lru` | 图片缓存容量控制 | 维护 | 30 分钟 | 超过 `storage.budgets.imagesGb` 按最久未访问删到 90% | | `storage.retention.*` | 保留清理(观察 / SSE / 任务运行 / Agent 事件 / 审计 / 能力统计 / 待识别池归档) | 维护 | 1–24 小时 | 分批删除超过保留期的流水数据,见 [storage.md](storage.md) | | `storage.analyze` | 统计信息更新 | 维护 | 7 天 | 全库 `ANALYZE` | +| `metadata.backfill_languages` | 元数据语言回刷 | 维护 | 24 小时,跑完自动停用 | 部署后首次启动即运行:对已确认 TMDB 身份的作品按语言回退链 full 刷新一次(TMDB 每秒最多 2 次,走请求缓存),进度写运行记录;跑完把自己停用,可手动再执行。见 [metadata.md](metadata.md) | | `maintain.backup` | 备份 | 维护 | 每天 03:30(占位) | 未配置。备份目前由宿主机上的 `scripts/backup.sh` 执行 | | `check.pending` | 待处理巡检 | Agent | 6 小时,默认关闭 | 只读。有未关闭事项时才会运行 | | `check.downloads` | 下载诊断 | Agent | 2 小时,默认关闭 | 默认只读。打开 `allowRecover` 后,在授权范围内重试或重新校验 |