feat(metadata): 作品进入关注时预热元数据缓存
- 新任务 prewarmMetadata(网络道):拉基础详情、富详情(含标志)、各季详情、选中的剧集组, 海报 / 背景 / 标志原图进图片缓存;请求键与刷新元数据 / 刮削相同,之后直接命中且 pinned - enqueueWatchedRefresh(建订阅、用户搜索 / 手动识别、提交下载)每次额外入队预热,幂等键按作品 + 日期 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
1 parent
680419bb64
commit
1535d8664e
7 files changed
+291
-5
No files matched your search
@@ -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<PrewarmResult> {
|
||||
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<ReturnType<typeof fetchRichDetail>> | 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;
|
||||
}
|
||||
@@ -62,9 +62,12 @@ export async function isEntityWatched(exec: Exec, entityId: string): Promise<boo
|
||||
}
|
||||
|
||||
/**
|
||||
* 作品进入关注范围(建订阅、用户搜索、开始下载)时补排一次 full 元数据刷新:只对从没刷新过的作品排
|
||||
* (自动建档时因不在关注范围而没排的),刷新完成会重算待确认 / 滞后放行资源的集数(必要时开集数待确认)。
|
||||
* 入队失败只记日志;同一作品同一原因只排一次(幂等键)。
|
||||
* 作品进入关注范围(建订阅、用户搜索 / 手动识别、开始下载)时:
|
||||
* - 补排一次 full 元数据刷新:只对从没刷新过的作品排(自动建档时因不在关注范围而没排的),
|
||||
* 刷新完成会重算待确认 / 滞后放行资源的集数(必要时开集数待确认);同一作品只排一次(幂等键);
|
||||
* - 每次都排一次缓存预热(prewarmMetadata:详情、各季、剧集组、海报 / 背景 / 标志进缓存),幂等键按作品 + 日期,
|
||||
* 同一天重复进入只跑一次;条目已在缓存里时预热只是命中,不发请求。
|
||||
* 入队失败只记日志。返回是否排了刷新。
|
||||
*/
|
||||
export async function enqueueWatchedRefresh(ctx: ServerContext, entityId: string, reason: string): Promise<boolean> {
|
||||
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<boolean> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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<ReturnType<typeof tmdbCachePolicy>> | 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" });
|
||||
});
|
||||
});
|
||||
@@ -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",
|
||||
|
||||
@@ -39,6 +39,7 @@ export const WORKFLOW_LABEL: Record<string, string> = {
|
||||
verifyAndImport: "核验整理",
|
||||
scrapeMedia: "刮削",
|
||||
refreshMetadata: "刷新元数据",
|
||||
prewarmMetadata: "预热元数据缓存",
|
||||
reconcileLibrary: "文件核对",
|
||||
cleanupDownloadCache: "清理缓存",
|
||||
resolveDecision: "处理待处理事项",
|
||||
|
||||
@@ -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 | 站点登录 / 用户数据刷新 |
|
||||
|
||||
Reference in new issue
Block a user