feat(识别): 集数识别分两层,只对关注中的作品拉元数据、开集数待确认
- 新增 isEntityWatched(service/watched.ts):有进行中订阅、用户搜索 / 手动识别 / 手动下载 / 人工确认身份、 正在下载或整理、已在媒体库的作品算关注中 - 自动新建作品后的 full 刷新只在关注范围内排(用户搜索 / 手动识别 / 接管照常);集数待确认事项只给关注中的作品开 - 进入关注范围(建订阅、用户搜索命中已有作品、开始下载)时对从没刷新过的作品补排刷新,刷新重算后再开事项 - 确定性的集数校验 / 换算对所有已绑定作品照做 Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
This commit is contained in:
1 parent
731336ccb1
commit
7b7fadbfb6
7 files changed
+257
-8
No files matched your search
@@ -14,6 +14,7 @@ import type { SubscriptionRow } from "@umt/db";
|
||||
import type { Context } from "hono";
|
||||
import type { AuthEnv } from "../../../http/auth.ts";
|
||||
import { ApiException } from "../../../http/errors.ts";
|
||||
import { enqueueWatchedRefresh } from "../../sources-identity/service/watched.ts";
|
||||
import type { ModuleApp } from "../../types.ts";
|
||||
import { ensureSubscriptionSlot } from "../services/acquisition.ts";
|
||||
import type { AoDeps, Exec } from "../services/common.ts";
|
||||
@@ -378,6 +379,8 @@ export function registerSubscriptionRoutes(app: ModuleApp, d: AoDeps) {
|
||||
return r;
|
||||
});
|
||||
if (row.media_entity_id) {
|
||||
// 建订阅 = 作品进入关注范围:从没刷新过元数据的补排 full 刷新(集数两层,见 sources-identity/service/watched.ts)
|
||||
await enqueueWatchedRefresh(d.ctx, row.media_entity_id, "subscription");
|
||||
await kickEvaluate(d, { subscriptionId: row.id });
|
||||
// 新订阅:立即主动搜索一次(之后按周期);剧集先刷新元数据拿季总集数
|
||||
await enqueueSubscriptionSearch(d, row.id, "created").catch((err) =>
|
||||
|
||||
@@ -35,6 +35,7 @@ import {
|
||||
import { mergeObservedDownloadState, toDomainDownloadStatus } from "@umt/mapping";
|
||||
import { withAttemptToken } from "@umt/tasks";
|
||||
import { createEntityEpisodeMapper, publishedAtOf } from "../../sources-identity/service/episode-resolution.ts";
|
||||
import { enqueueWatchedRefresh } from "../../sources-identity/service/watched.ts";
|
||||
import { errMessage, jsonb } from "../util.ts";
|
||||
import { parsedOf } from "./acquisition.ts";
|
||||
import {
|
||||
@@ -266,6 +267,9 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
|
||||
.selectAll()
|
||||
.where("id", "=", plan.source_release_id)
|
||||
.executeTakeFirstOrThrow();
|
||||
// 开始下载 = 作品进入关注范围:从没刷新过元数据的(自动建档时不在关注范围)补排一次 full 刷新(集数两层,见 watched.ts)
|
||||
const watchedEntity = ev.mediaEntityId ?? rel.media_entity_id;
|
||||
if (watchedEntity) await enqueueWatchedRefresh(d.ctx, watchedEntity, "download");
|
||||
const site = await db.selectFrom("site").selectAll().where("id", "=", rel.site_id).executeTakeFirst();
|
||||
const client = await d.ports.downloader(topo.inst);
|
||||
const tag = topo.inst.managed_tag;
|
||||
|
||||
@@ -25,6 +25,7 @@ import { jsonb } from "../rows.ts";
|
||||
import { closeDecisions, type Exec, openDecision } from "./common.ts";
|
||||
import { ensureUnits } from "./entities.ts";
|
||||
import { seasonOrigins } from "./episode-structure.ts";
|
||||
import { isEntityWatched } from "./watched.ts";
|
||||
|
||||
// 集数解析接入(docs/plan/episode-resolution.md 批次 2):识别绑定作品后、建集单元前,按作品季表 + 映射规则把
|
||||
// 「发布者声称的季集」换成「媒体库应当使用的季集」。判定本身在 domain resolveEpisodes,这里只装配数据与落库:
|
||||
@@ -274,6 +275,9 @@ export async function settleEpisodeDecision(
|
||||
: null;
|
||||
await detachFromEpisodeDecisions(ctx, exec, sourceReleaseId, target);
|
||||
if (!target) return;
|
||||
// 两层(2026-10-06):集数待确认是内部事项(供智能体调查,用户在订阅详情 / 作品页集数映射里看原因),
|
||||
// 只给关注中的作品开;不在关注范围的作品资源照样待确认(不建集、订阅跳过),进入关注范围补刷新时重算再开
|
||||
if (!(await isEntityWatched(exec, entityId))) return;
|
||||
|
||||
const existing = await exec
|
||||
.selectFrom("decision")
|
||||
|
||||
@@ -55,6 +55,7 @@ import {
|
||||
import { loadSubject, searchBangumi } from "./meta-subjects.ts";
|
||||
import { tmdbRecallInput, tmdbToCandidate } from "./tmdb-candidate.ts";
|
||||
import { loadRecognizeCache, type RecognizeMeta, recognizeMetaOf, saveRecognizeCache } from "./tmdb-recognize-cache.ts";
|
||||
import { enqueueWatchedRefresh, isEntityWatched } from "./watched.ts";
|
||||
|
||||
// resolveObservation:parseRelease → 本地召回(别名 pg_trgm + 已有外部身份 + 人工锁)→ [按需] TMDB 候选 → domain.resolveIdentity。
|
||||
// auto 才建/关联正式作品;ambiguous / no_match / 查询失败进待识别池(候选只存 JSON),有界退避复查。
|
||||
@@ -900,7 +901,14 @@ async function bindAuto(
|
||||
reason: resolution.reason,
|
||||
};
|
||||
});
|
||||
if (created) await enqueueNewEntityRefresh(ctx, created);
|
||||
// 两层(2026-10-06):自动新建的作品只在关注范围内才排刷新(用户搜索 / 手动识别 / 接管下载,或已被订阅等);
|
||||
// RSS / 列表扫进来、没人关心的作品不拉 TMDB 季表,等它进入关注范围(建订阅、搜索、开始下载)时再补排
|
||||
const userAsked = lookup === "search" || lookup === "manual" || lookup === "takeover";
|
||||
if (created) {
|
||||
if (userAsked || (await isEntityWatched(ctx.db, created))) await enqueueNewEntityRefresh(ctx, created);
|
||||
} else if (userAsked && result.mediaEntityId) {
|
||||
await enqueueWatchedRefresh(ctx, result.mediaEntityId, `lookup:${lookup}`);
|
||||
}
|
||||
return result;
|
||||
} catch (e) {
|
||||
if (e instanceof IdentityConflictError && attempt < 2) continue;
|
||||
|
||||
@@ -0,0 +1,70 @@
|
||||
import { sql } from "@umt/db";
|
||||
import type { ServerContext } from "../../types.ts";
|
||||
import type { Exec } from "./common.ts";
|
||||
|
||||
// 关注中的作品(2026-10-06,集数识别分两层):
|
||||
// 确定性的集数校验 / 换算对所有已绑定作品的资源照做(无外部请求、无打扰);
|
||||
// 有成本或会打扰的后续动作——拉 TMDB、自动新建作品后的元数据刷新、开集数待确认、派智能体——只对关注中的作品做。
|
||||
//
|
||||
// 关注中 = 满足任一:
|
||||
// 1. 有进行中的订阅(subscription.status in active / pending);
|
||||
// 2. 用户明确表达过兴趣:手动下载过(acquisition_plan.evaluation.manual = true)、用户搜索 / 手动识别出的资源
|
||||
// (source_release.resolution_evidence.lookup in search / manual)、人工 / 智能体确认的外部身份(evidence_level = manual);
|
||||
// 3. 正在下载或整理:未移除且没入库完的下载任务(直接、经资源或经下载计划关联),或排队 / 执行中的整理记录;
|
||||
// 4. 已在媒体库:有在用(active)的已入库文件。
|
||||
// RSS / 站点列表扫进来、自动建档却没人关心的作品不算。
|
||||
|
||||
const OPEN_SUBSCRIPTION = ["active", "pending"];
|
||||
const DOWNLOAD_DONE = ["imported", "removed"];
|
||||
|
||||
export async function isEntityWatched(exec: Exec, entityId: string): Promise<boolean> {
|
||||
const r = await sql<{ watched: boolean }>`
|
||||
select (
|
||||
exists (select 1 from subscription s where s.media_entity_id = ${entityId} and s.status = any(${OPEN_SUBSCRIPTION}))
|
||||
or exists (select 1 from media_asset a where a.media_entity_id = ${entityId} and a.status = 'active')
|
||||
or exists (select 1 from external_identity x
|
||||
where x.media_entity_id = ${entityId} and x.status = 'accepted' and x.evidence_level = 'manual')
|
||||
or exists (select 1 from source_release r
|
||||
where r.media_entity_id = ${entityId} and r.resolution_evidence->>'lookup' in ('search', 'manual'))
|
||||
or exists (select 1 from acquisition_plan p left join acquisition_slot sl on sl.id = p.slot_id
|
||||
where (sl.media_entity_id = ${entityId} or p.evaluation->>'mediaEntityId' = ${entityId}::text)
|
||||
and (p.evaluation->>'manual')::boolean is true)
|
||||
or exists (select 1 from download_task t
|
||||
left join source_release r on r.id = t.source_release_id
|
||||
left join acquisition_plan p on p.id = t.acquisition_plan_id
|
||||
left join acquisition_slot sl on sl.id = p.slot_id
|
||||
where t.removed_at is null and t.status <> all(${DOWNLOAD_DONE})
|
||||
and ${entityId} in (t.media_entity_id, r.media_entity_id, sl.media_entity_id))
|
||||
or exists (select 1 from organization_record o
|
||||
where o.media_entity_id = ${entityId} and o.status in ('queued', 'running'))
|
||||
) as watched`.execute(exec);
|
||||
return r.rows[0]?.watched === true;
|
||||
}
|
||||
|
||||
/**
|
||||
* 作品进入关注范围(建订阅、用户搜索、开始下载)时补排一次 full 元数据刷新:只对从没刷新过的作品排
|
||||
* (自动建档时因不在关注范围而没排的),刷新完成会重算待确认 / 滞后放行资源的集数(必要时开集数待确认)。
|
||||
* 入队失败只记日志;同一作品同一原因只排一次(幂等键)。
|
||||
*/
|
||||
export async function enqueueWatchedRefresh(ctx: ServerContext, entityId: string, reason: string): Promise<boolean> {
|
||||
const e = await ctx.db
|
||||
.selectFrom("media_entity")
|
||||
.select(["id", "last_metadata_refresh_at", "status"])
|
||||
.where("id", "=", entityId)
|
||||
.executeTakeFirst();
|
||||
if (!e || e.last_metadata_refresh_at || e.status === "merged") return false;
|
||||
try {
|
||||
await ctx.engine.enqueue(
|
||||
"refreshMetadata",
|
||||
{ mediaEntityId: entityId, full: true },
|
||||
{ idempotencyKey: `refreshMetadata:watched:${entityId}`, traceId: entityId },
|
||||
);
|
||||
return true;
|
||||
} catch (err) {
|
||||
ctx.logger.debug(
|
||||
{ entityId, reason, err: { message: err instanceof Error ? err.message : String(err) } },
|
||||
"进入关注范围的元数据刷新入队失败",
|
||||
);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,150 @@
|
||||
import { createMockTmdb, type MockTmdb, TMDB_FIXTURES, type TmdbFixture } from "@umt/testkit";
|
||||
import { sql } from "kysely";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { refreshMetadata } from "../../src/modules/sources-identity/service/metadata.ts";
|
||||
import { resolveRelease } from "../../src/modules/sources-identity/service/resolve.ts";
|
||||
import { enqueueWatchedRefresh, isEntityWatched } from "../../src/modules/sources-identity/service/watched.ts";
|
||||
import { createHarness, type Harness, pgAvailable, seedSite } from "./harness.ts";
|
||||
|
||||
// 集数识别分两层(2026-10-06):
|
||||
// 确定性校验 / 换算对所有已绑定作品的资源照做;拉 TMDB(自动建档后的元数据刷新)、开集数待确认只对关注中的作品做。
|
||||
// RSS 扫进来自动建档、没人关心的作品:不排刷新、不开事项;进入关注范围(建订阅等)补排刷新,刷新重算后才开内部事项。
|
||||
|
||||
const pg = await pgAvailable();
|
||||
|
||||
const season = (n: number, eps: number, start: string) => ({
|
||||
season_number: n,
|
||||
air_date: start,
|
||||
names: { "zh-CN": `第 ${n} 季`, "en-US": `Season ${n}` },
|
||||
episodes: Array.from({ length: eps }, (_, k) => ({
|
||||
episode_number: k + 1,
|
||||
air_date: new Date(Date.parse(start) + k * 7 * 86_400_000).toISOString().slice(0, 10),
|
||||
names: { "zh-CN": `第 ${k + 1} 集`, "en-US": `Episode ${k + 1}` },
|
||||
})),
|
||||
});
|
||||
|
||||
// 已完结:S1 = 12、S2 = 13;绝对集号 40 超出全部季,刷新后换算不了 → 集数待确认
|
||||
const SHOW: TmdbFixture = {
|
||||
type: "tv",
|
||||
id: 990301,
|
||||
original_title: "Yoru no Kurage wa Oyogenai",
|
||||
original_language: "ja",
|
||||
origin_country: ["JP"],
|
||||
date: "2024-04-06",
|
||||
titles: { "zh-CN": "夜晚的水母不会游泳", "en-US": "Yoru no Kurage wa Oyogenai" },
|
||||
popularity: 30,
|
||||
alternative_titles: [],
|
||||
external_ids: {},
|
||||
rich: { status: "Ended" },
|
||||
seasons: [season(1, 12, "2024-04-06"), season(2, 13, "2025-04-05")],
|
||||
};
|
||||
|
||||
describe.skipIf(!pg)("集数识别两层:只对关注中的作品拉元数据、开集数待确认", () => {
|
||||
let h: Harness;
|
||||
let tmdb: MockTmdb;
|
||||
let siteId: string;
|
||||
let n = 0;
|
||||
|
||||
beforeAll(async () => {
|
||||
tmdb = await createMockTmdb({ fixtures: [...TMDB_FIXTURES, SHOW] });
|
||||
h = await createHarness("episode_watch_scope", { tmdbUrl: tmdb.url });
|
||||
siteId = (await seedSite(h, { baseUrl: "https://acgrip.example/" })).site.id;
|
||||
// 不靠订阅也查 TMDB(旧行为):RSS 扫进来的资源会自动建档,但作品不在关注范围
|
||||
await sql`insert into settings (key, value) values ('identity', ${JSON.stringify({ eagerTmdbLookup: true })}::jsonb)
|
||||
on conflict (key) do update set value = excluded.value`.execute(h.db);
|
||||
});
|
||||
afterAll(async () => {
|
||||
await h?.close();
|
||||
await tmdb?.close();
|
||||
});
|
||||
|
||||
const observe = async (title: string) => {
|
||||
const r = await h.db
|
||||
.insertInto("source_release")
|
||||
.values({ site_id: siteId, site_release_id: String(830000 + ++n), title, subtitle: "夜晚的水母不会游泳" })
|
||||
.returningAll()
|
||||
.executeTakeFirstOrThrow();
|
||||
const res = await resolveRelease(h.rt, { sourceReleaseId: r.id, trigger: "observed" });
|
||||
expect(res.outcome, JSON.stringify(res)).toBe("auto");
|
||||
return { id: r.id, entityId: res.mediaEntityId as string };
|
||||
};
|
||||
const method = async (id: string) =>
|
||||
(
|
||||
(
|
||||
await h.db
|
||||
.selectFrom("source_release")
|
||||
.select("episode_resolution")
|
||||
.where("id", "=", id)
|
||||
.executeTakeFirstOrThrow()
|
||||
).episode_resolution as { method: string }
|
||||
).method;
|
||||
const refreshJobs = () => h.engine.enqueued.filter((e) => e.task === "refreshMetadata");
|
||||
const episodeDecisions = () =>
|
||||
h.db
|
||||
.selectFrom("decision")
|
||||
.selectAll()
|
||||
.where("kind", "=", "episode_mapping")
|
||||
.where("status", "in", ["open", "investigating", "awaiting_user"])
|
||||
.execute();
|
||||
|
||||
let entityId: string;
|
||||
let r40: string;
|
||||
|
||||
it("RSS 扫进来自动建档:确定性换算照做,但不排元数据刷新、不开集数待确认", async () => {
|
||||
const a = await observe("[ANi] Yoru no Kurage wa Oyogenai - 03 [1080P][Baha][WEB-DL][AAC AVC][CHT][MP4]");
|
||||
entityId = a.entityId;
|
||||
expect(await method(a.id)).toBe("unverified");
|
||||
expect(await isEntityWatched(h.db, entityId)).toBe(false);
|
||||
const b = await observe("[Lilith-Raws] Yoru no Kurage wa Oyogenai - 40 [Baha][WEB-DL][1080p][AVC AAC][CHT][MP4]");
|
||||
r40 = b.id;
|
||||
expect(b.entityId).toBe(entityId);
|
||||
expect(refreshJobs()).toEqual([]);
|
||||
expect(await episodeDecisions()).toEqual([]);
|
||||
// 季详情一次都没拉(识别只用搜索与详情)
|
||||
expect(tmdb.callsTo(/\/season\//)).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("即使季表已知、资源换算不了,非关注作品也不开事项", async () => {
|
||||
// 直接刷新一次(模拟别处拉过季表),非关注作品的待确认资源照样只记在资源上
|
||||
expect((await refreshMetadata(h.rt, entityId, { full: true })).status).toBe("refreshed");
|
||||
expect(await method(r40)).toBe("unresolved");
|
||||
expect(await episodeDecisions()).toEqual([]);
|
||||
await h.db.updateTable("media_entity").set({ last_metadata_refresh_at: null }).where("id", "=", entityId).execute();
|
||||
});
|
||||
|
||||
it("进入关注范围(建订阅):补排一次 full 刷新;刷新重算后对换算不了的资源开内部事项", async () => {
|
||||
await h.db
|
||||
.insertInto("subscription")
|
||||
.values({ media_entity_id: entityId, media_kind: "series", status: "active" })
|
||||
.execute();
|
||||
expect(await isEntityWatched(h.db, entityId)).toBe(true);
|
||||
expect(await enqueueWatchedRefresh(h.ctx, entityId, "subscription")).toBe(true);
|
||||
expect(refreshJobs()).toEqual([
|
||||
{
|
||||
task: "refreshMetadata",
|
||||
input: { mediaEntityId: entityId, full: true },
|
||||
key: `refreshMetadata:watched:${entityId}`,
|
||||
},
|
||||
]);
|
||||
// 执行入队的刷新(同 refreshMetadata 任务处理器)
|
||||
expect((await refreshMetadata(h.rt, entityId, { full: true })).status).toBe("refreshed");
|
||||
const ds = await episodeDecisions();
|
||||
expect(ds).toHaveLength(1);
|
||||
expect((ds[0]!.evidence as { sourceReleaseIds: string[] }).sourceReleaseIds).toEqual([r40]);
|
||||
// 已刷新过的不再补排
|
||||
expect(await enqueueWatchedRefresh(h.ctx, entityId, "download")).toBe(false);
|
||||
});
|
||||
|
||||
it("用户搜索 / 手动识别出的新作品照常排刷新", async () => {
|
||||
const r = await h.db
|
||||
.insertInto("source_release")
|
||||
.values({ site_id: siteId, site_release_id: String(830000 + ++n), title: "Oppenheimer.2023.2160p.UHD.BluRay" })
|
||||
.returningAll()
|
||||
.executeTakeFirstOrThrow();
|
||||
const before = refreshJobs().length;
|
||||
const res = await resolveRelease(h.rt, { sourceReleaseId: r.id, trigger: "search" });
|
||||
expect(res.outcome, JSON.stringify(res)).toBe("auto");
|
||||
expect(await isEntityWatched(h.db, res.mediaEntityId as string)).toBe(true);
|
||||
expect(refreshJobs().length).toBe(before + 1);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user