From 1535d8664eeccba56a1ee05228f414736e8e7c24 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 6 Oct 2026 19:37:34 +0800 Subject: [PATCH] =?UTF-8?q?feat(metadata):=20=E4=BD=9C=E5=93=81=E8=BF=9B?= =?UTF-8?q?=E5=85=A5=E5=85=B3=E6=B3=A8=E6=97=B6=E9=A2=84=E7=83=AD=E5=85=83?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=BC=93=E5=AD=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新任务 prewarmMetadata(网络道):拉基础详情、富详情(含标志)、各季详情、选中的剧集组, 海报 / 背景 / 标志原图进图片缓存;请求键与刷新元数据 / 刮削相同,之后直接命中且 pinned - enqueueWatchedRefresh(建订阅、用户搜索 / 手动识别、提交下载)每次额外入队预热,幂等键按作品 + 日期 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- .../service/metadata-prewarm.ts | 122 +++++++++++++++++ .../sources-identity/service/watched.ts | 32 ++++- .../src/modules/sources-identity/tasks.ts | 10 ++ .../sources-identity/metadata-prewarm.test.ts | 127 ++++++++++++++++++ .../test/sources-identity/mteam-tasks.test.ts | 3 +- apps/web/app/components/tasks/labels.ts | 1 + docs/runbook/dev.md | 1 + 7 files changed, 291 insertions(+), 5 deletions(-) create mode 100644 apps/server/src/modules/sources-identity/service/metadata-prewarm.ts create mode 100644 apps/server/test/sources-identity/metadata-prewarm.test.ts diff --git a/apps/server/src/modules/sources-identity/service/metadata-prewarm.ts b/apps/server/src/modules/sources-identity/service/metadata-prewarm.ts new file mode 100644 index 00000000..9d861916 --- /dev/null +++ b/apps/server/src/modules/sources-identity/service/metadata-prewarm.ts @@ -0,0 +1,122 @@ +import { join } from "node:path"; +import { fetchEpisodeGroup, fetchRichDetail, fetchRichSeason, TmdbError } from "@umt/adapters"; +import { dataPaths } from "../../../config.ts"; +import { externalFetch } from "../../../egress.ts"; +import { resetTrackedTmdbRefs } from "../../../shared-cache.ts"; +import { ImageCache } from "../../cache/images.ts"; +import type { SourcesRuntime } from "../runtime.ts"; +import { entityEpisodeGroup } from "./episode-structure.ts"; +import { metadataIncludeSpecials, metadataLanguages, metadataTextOptions } from "./metadata-rich.ts"; + +// 关注作品预热(统一元数据缓存,docs/runbook/cache.md):作品进入关注时(建订阅、用户搜索 / 手动识别、开始下载) +// 把之后刷新 / 刮削会用到的 TMDB 条目先拉进缓存:详情(含标志)、各季详情、选中的剧集组,海报 / 背景 / 标志图片进图片缓存。 +// 请求键与刷新元数据 / 刮削完全相同(同一套 fetchRichDetail / fetchRichSeason / fetchEpisodeGroup,同样的语言链), +// 所以后面直接命中;关注中作品的条目写入时 pinned(条数上限淘汰不清掉),命中时续期。 + +/** 一部作品最多预热多少季(长寿剧) */ +const MAX_SEASONS = 60; + +export interface PrewarmResult { + status: "warmed" | "skipped"; + reason?: string; + seasons: number; + episodeGroup: boolean; + images: number; + errors: string[]; +} + +/** 图片预热用的 fetch:测试注入优先;测试进程没注入就不出外网 */ +function imageFetch(rt: SourcesRuntime): typeof globalThis.fetch | null { + if (rt.overrides.fetch) return rt.overrides.fetch; + return Deno.env.get("VITEST") ? null : externalFetch; +} + +export async function prewarmMetadata(rt: SourcesRuntime, mediaEntityId: string): Promise { + const { ctx } = rt; + const out: PrewarmResult = { status: "skipped", seasons: 0, episodeGroup: false, images: 0, errors: [] }; + const ident = await ctx.db + .selectFrom("external_identity as x") + .innerJoin("media_entity as e", "e.id", "x.media_entity_id") + .select(["x.entity_type", "x.external_id", "e.status"]) + .where("x.media_entity_id", "=", mediaEntityId) + .where("x.provider", "=", "tmdb") + .where("x.status", "=", "accepted") + .executeTakeFirst(); + if (!ident || ident.status === "merged") return { ...out, reason: ident ? "merged" : "no_tmdb_identity" }; + if (ident.entity_type !== "movie" && ident.entity_type !== "tv") return { ...out, reason: "unsupported_type" }; + const tmdb = await rt.tmdb(); + if (!tmdb) return { ...out, reason: "tmdb_not_configured" }; + // 刚进入关注:关注集合按进程缓存 5 分钟,清掉让这次写入就标 pinned + resetTrackedTmdbRefs(ctx); + const type = ident.entity_type; + const id = ident.external_id; + const langs = await metadataLanguages(ctx); + const textOpts = await metadataTextOptions(ctx); + const note = (what: string, e: unknown) => out.errors.push(`${what}: ${(e as Error).message}`); + + // 刷新元数据的基础详情与富详情(含标志) + await tmdb.details(type, id).catch((e) => note("details", e)); + // 注入的简化客户端(测试)没有 getJson:只预热基础详情 + if (typeof tmdb.getJson !== "function") return { ...out, status: "warmed" }; + let rich: Awaited> | null = null; + try { + rich = await fetchRichDetail(tmdb, type, id, langs, textOpts); + } catch (e) { + // 详情都拿不到(404 / 鉴权 / 网络):网络类交给任务重试,其余记下结束 + if (e instanceof TmdbError && ["network", "timeout", "rate"].includes(e.kind)) throw e; + note("detail", e); + return { ...out, reason: "detail_unavailable" }; + } + out.status = "warmed"; + + if (type === "tv") { + const specials = await metadataIncludeSpecials(ctx); + const seasons = rich.seasons + .map((s) => s.seasonNumber) + .filter((n) => specials || n > 0) + .slice(0, MAX_SEASONS); + for (const n of seasons) { + try { + await fetchRichSeason(tmdb, id, n, langs, { + now: rt.now(), + ...textOpts, + originalLanguage: rich.originalLanguage, + }); + out.seasons++; + } catch (e) { + note(`season ${n}`, e); + } + } + const group = await entityEpisodeGroup(ctx.db, mediaEntityId); + if (group) { + try { + await fetchEpisodeGroup(tmdb, group.id, langs[0]); + out.episodeGroup = true; + } catch (e) { + note("episode group", e); + } + } + } + + // 图片:刮削取原图(original),这里预先拉进图片缓存目录 + const f = imageFetch(rt); + if (f) { + const images = new ImageCache({ + dir: join(dataPaths(ctx.config).dataDir, "cache", "images"), + fetch: f, + timeoutMs: 30_000, + }); + for (const p of [rich.posterPath, rich.backdropPath, rich.logoPath]) { + if (!p) continue; + try { + await images.bytes("original", p); + out.images++; + } catch (e) { + note(`image ${p}`, e); + } + } + } + if (out.errors.length) + ctx.logger.debug({ entityId: mediaEntityId, errors: out.errors }, "关注作品预热有部分条目没拉到"); + return out; +} diff --git a/apps/server/src/modules/sources-identity/service/watched.ts b/apps/server/src/modules/sources-identity/service/watched.ts index b3f0f7e5..b39579f8 100644 --- a/apps/server/src/modules/sources-identity/service/watched.ts +++ b/apps/server/src/modules/sources-identity/service/watched.ts @@ -62,9 +62,12 @@ export async function isEntityWatched(exec: Exec, entityId: string): Promise { const e = await ctx.db @@ -72,7 +75,9 @@ export async function enqueueWatchedRefresh(ctx: ServerContext, entityId: string .select(["id", "last_metadata_refresh_at", "status"]) .where("id", "=", entityId) .executeTakeFirst(); - if (!e || e.last_metadata_refresh_at || e.status === "merged") return false; + if (!e || e.status === "merged") return false; + await enqueuePrewarm(ctx, entityId, reason); + if (e.last_metadata_refresh_at) return false; try { await ctx.engine.enqueue( "refreshMetadata", @@ -88,3 +93,22 @@ export async function enqueueWatchedRefresh(ctx: ServerContext, entityId: string return false; } } + +/** 关注作品预热入队(幂等键按作品 + 日期);失败只记日志 */ +export async function enqueuePrewarm(ctx: ServerContext, entityId: string, reason: string): Promise { + const day = new Date().toISOString().slice(0, 10); + try { + const r = await ctx.engine.enqueue( + "prewarmMetadata", + { mediaEntityId: entityId }, + { idempotencyKey: `prewarmMetadata:${entityId}:${day}`, traceId: entityId }, + ); + return !r.deduplicated; + } catch (err) { + ctx.logger.debug( + { entityId, reason, err: { message: err instanceof Error ? err.message : String(err) } }, + "关注作品预热入队失败", + ); + return false; + } +} diff --git a/apps/server/src/modules/sources-identity/tasks.ts b/apps/server/src/modules/sources-identity/tasks.ts index f6287bc6..3633db46 100644 --- a/apps/server/src/modules/sources-identity/tasks.ts +++ b/apps/server/src/modules/sources-identity/tasks.ts @@ -6,6 +6,7 @@ import { refreshSiteAccount } from "./service/account.ts"; import { AUTO_ENRICH_TASK, runAutoEnrich } from "./service/enrich.ts"; import { ingestFeed } from "./service/feeds.ts"; import { refreshMetadata } from "./service/metadata.ts"; +import { prewarmMetadata } from "./service/metadata-prewarm.ts"; import { resolveRelease } from "./service/resolve.ts"; import { snapshotSite } from "./service/snapshot.ts"; import { liveSearch } from "./service/workbench.ts"; @@ -77,6 +78,15 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu handler: (t) => t.step.run("refresh", () => refreshMetadata(rt(), t.input.mediaEntityId, { full: !!t.input.full })), }), + // 关注作品预热(统一元数据缓存):详情 / 各季 / 剧集组 / 图片先进缓存;网络道,走 TMDB 客户端限流 + defineTask({ + name: "prewarmMetadata", + version: 1, + lane: "network", + inputSchema: z.object({ mediaEntityId: id }), + retryPolicy: { maximumAttempts: 3, initialInterval: "5s", backoffCoefficient: 2, maximumInterval: "60s" }, + handler: (t) => t.step.run("prewarm", () => prewarmMetadata(rt(), t.input.mediaEntityId)), + }), defineTask({ name: "searchSubscription", version: 1, diff --git a/apps/server/test/sources-identity/metadata-prewarm.test.ts b/apps/server/test/sources-identity/metadata-prewarm.test.ts new file mode 100644 index 00000000..c378e4a8 --- /dev/null +++ b/apps/server/test/sources-identity/metadata-prewarm.test.ts @@ -0,0 +1,127 @@ +import { existsSync, mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createTmdbClient } from "@umt/adapters"; +import { sql } from "@umt/db"; +import { createMockTmdb, type MockTmdb, SPLIT_COUR_FIXTURE, TMDB_FIXTURES } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { tmdbSharedCache } from "../../src/modules/sources-identity/runtime.ts"; +import { prewarmMetadata } from "../../src/modules/sources-identity/service/metadata-prewarm.ts"; +import { enqueueWatchedRefresh } from "../../src/modules/sources-identity/service/watched.ts"; +import { tmdbCachePolicy } from "../../src/shared-cache.ts"; +import { createHarness, type Harness, pgAvailable } from "./harness.ts"; + +// 关注作品预热:进入关注时入队 prewarmMetadata(幂等键按作品 + 日期);执行后详情(含标志)、各季、选中的剧集组 +// 在 TMDB 请求缓存里且 pinned,海报 / 背景 / 标志进图片缓存;再跑一次全部命中,不发请求。 + +const skip = !(await pgAvailable()); +const PNG = new Uint8Array([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 1]); +const JPEG = new Uint8Array([0xff, 0xd8, 0xff, 0xe0, 1]); + +describe.skipIf(skip)("关注作品预热", () => { + let h: Harness; + let tmdb: MockTmdb; + let entityId: string; + const images: string[] = []; + const dataDir = mkdtempSync(join(tmpdir(), "umt-prewarm-")); + + beforeAll(async () => { + tmdb = await createMockTmdb({ + fixtures: [ + ...TMDB_FIXTURES, + { + ...SPLIT_COUR_FIXTURE, + rich: { ...SPLIT_COUR_FIXTURE.rich, backdrop_path: "/split-bg.jpg" }, + logos: [ + { file_path: "/split-en.png", iso_639_1: "en" }, + { file_path: "/split-ja.png", iso_639_1: "ja" }, + ], + }, + ], + }); + let policy: Awaited> | null = null; + h = await createHarness("prewarm", { + // 和服务端一样走统一元数据缓存(PG 请求缓存) + tmdb: () => + createTmdbClient({ + apiKey: "test-tmdb-key", + baseUrl: `${tmdb.url}/3`, + retries: 0, + sharedCache: tmdbSharedCache(h.ctx, policy!), + }), + // 图片上游(image.tmdb.org) + fetch: async (input) => { + const url = String(input); + images.push(url); + return new Response(url.endsWith(".png") ? PNG : JPEG); + }, + }); + policy = await tmdbCachePolicy(h.ctx); + (h.ctx.config as { DATA_DIR?: string }).DATA_DIR = dataDir; + entityId = ( + await sql<{ id: string }>`insert into media_entity (kind, title, episode_group_id, episode_group_name) + values ('series', '分割番', 'grp-production', '制作季') returning id`.execute(h.db) + ).rows[0]!.id; + await sql`insert into external_identity (media_entity_id, provider, entity_type, external_id, status) + values (${entityId}, 'tmdb', 'tv', ${String(SPLIT_COUR_FIXTURE.id)}, 'accepted')`.execute(h.db); + }); + afterAll(async () => { + await h?.close(); + await tmdb?.close(); + rmSync(dataDir, { recursive: true, force: true }); + }); + + it("进入关注(建订阅)时入队预热,幂等键按作品 + 日期;同一天再进入不重复排", async () => { + await sql`insert into subscription (media_entity_id, media_kind, status) values (${entityId}, 'series', 'active')`.execute( + h.db, + ); + await enqueueWatchedRefresh(h.ctx, entityId, "subscription"); + await enqueueWatchedRefresh(h.ctx, entityId, "download"); + const jobs = h.engine.enqueued.filter((e) => e.task === "prewarmMetadata"); + expect(jobs).toHaveLength(1); + expect(jobs[0]).toMatchObject({ + input: { mediaEntityId: entityId }, + key: `prewarmMetadata:${entityId}:${new Date().toISOString().slice(0, 10)}`, + }); + }); + + it("执行后:详情 / 各季 / 剧集组 / 标志在缓存里且 pinned;海报、背景、标志进图片缓存", async () => { + const r = await prewarmMetadata(h.rt, entityId); + expect(r).toMatchObject({ status: "warmed", episodeGroup: true, images: 3, errors: [] }); + expect(r.seasons).toBe(SPLIT_COUR_FIXTURE.seasons!.length); + const rows = ( + await sql<{ key: string; pinned: boolean }>`select key, pinned from kv_cache where region = 'tmdb.request' + order by key`.execute(h.db) + ).rows; + const has = (re: RegExp) => rows.filter((x) => re.test(x.key)); + const id = SPLIT_COUR_FIXTURE.id; + expect(has(new RegExp(`^/tv/${id}\\?.*images`)).length).toBeGreaterThan(0); + for (const s of SPLIT_COUR_FIXTURE.seasons!) + expect(has(new RegExp(`^/tv/${id}/season/${s.season_number}\\?`)).length).toBeGreaterThan(0); + expect(has(/^\/tv\/episode_group\/grp-production\?/)).toHaveLength(1); + // 原语言是日语:另取了日语标志 + expect(has(new RegExp(`^/tv/${id}/images\\?include_image_language=ja`))).toHaveLength(1); + expect(rows.every((x) => x.pinned)).toBe(true); + for (const f of ["split.jpg", "split-bg.jpg", "split-ja.png"]) + expect(existsSync(join(dataDir, "cache/images/t/p/original", f))).toBe(true); + expect(images.every((u) => u.startsWith("https://image.tmdb.org/t/p/original/"))).toBe(true); + }); + + it("再预热一次全部命中:不发 TMDB 请求、不再下载图片", async () => { + const calls = tmdb.calls.length; + const fetched = images.length; + const r = await prewarmMetadata(h.rt, entityId); + expect(r.status).toBe("warmed"); + expect(tmdb.calls.length).toBe(calls); + expect(images.length).toBe(fetched); + }); + + it("没有 TMDB 身份的作品跳过", async () => { + const other = ( + await sql<{ id: string }>`insert into media_entity (kind, title) values ('movie', '本地') returning id`.execute( + h.db, + ) + ).rows[0]!.id; + expect(await prewarmMetadata(h.rt, other)).toMatchObject({ status: "skipped", reason: "no_tmdb_identity" }); + }); +}); diff --git a/apps/server/test/sources-identity/mteam-tasks.test.ts b/apps/server/test/sources-identity/mteam-tasks.test.ts index 4d68d56e..ecb84487 100644 --- a/apps/server/test/sources-identity/mteam-tasks.test.ts +++ b/apps/server/test/sources-identity/mteam-tasks.test.ts @@ -69,12 +69,13 @@ describe.skipIf(!up)("M-Team 快照与任务注册", () => { api.setFailMode(null); }); - it("模块注册 7 个任务与 3 条 outbox 路由;任务处理器经 step.run 调服务", async () => { + it("模块注册 8 个任务与 3 条 outbox 路由;任务处理器经 step.run 调服务", async () => { const reg = new TaskRegistry(); h.module.registerTasks(reg, h.ctx); expect(reg.names()).toEqual([ "enrichPoolEntry", "ingestFeed", + "prewarmMetadata", "refreshMetadata", "refreshSiteAccount", "resolveObservation", diff --git a/apps/web/app/components/tasks/labels.ts b/apps/web/app/components/tasks/labels.ts index 825f2fa1..5d9c1ae1 100644 --- a/apps/web/app/components/tasks/labels.ts +++ b/apps/web/app/components/tasks/labels.ts @@ -39,6 +39,7 @@ export const WORKFLOW_LABEL: Record = { verifyAndImport: "核验整理", scrapeMedia: "刮削", refreshMetadata: "刷新元数据", + prewarmMetadata: "预热元数据缓存", reconcileLibrary: "文件核对", cleanupDownloadCache: "清理缓存", resolveDecision: "处理待处理事项", diff --git a/docs/runbook/dev.md b/docs/runbook/dev.md index 21b45ad1..955a6c5f 100644 --- a/docs/runbook/dev.md +++ b/docs/runbook/dev.md @@ -99,6 +99,7 @@ ci 逐行核对这张表与任务注册表(`apps/server/test/worker-lanes.test.t | `resolveObservation` | network | 标题识别 + TMDB | | `enrichPoolEntry` | network | 未识别资源的元数据补全 | | `refreshMetadata` | network | TMDB 元数据刷新 | +| `prewarmMetadata` | network | 关注作品预热:详情 / 各季 / 剧集组 / 海报背景标志进缓存(docs/runbook/cache.md) | | `searchSubscription` | network | 订阅搜索 | | `snapshotSite` | network | 站点快照 | | `refreshSiteAccount` | network | 站点登录 / 用户数据刷新 |