diff --git a/apps/server/src/modules/sources-identity/capabilities.ts b/apps/server/src/modules/sources-identity/capabilities.ts index a0a0cd04..d33d8165 100644 --- a/apps/server/src/modules/sources-identity/capabilities.ts +++ b/apps/server/src/modules/sources-identity/capabilities.ts @@ -166,7 +166,8 @@ export function registerSourcesCapabilities(registry: CapabilityRegistry, rt: () decisionId: null, }; } - const res = await resolveRelease(r, { sourceReleaseId: srId, force: true }); + // Agent / 用户显式要求识别:按手动触发查 TMDB(惰性识别不拦) + const res = await resolveRelease(r, { sourceReleaseId: srId, force: true, trigger: "manual" }); const status: "resolved" | "proposed" | "ambiguous" | "unidentifiable" = res.outcome === "auto" || res.outcome === "skipped" || res.outcome === "skipped_locked" ? res.mediaEntityId diff --git a/apps/server/src/modules/sources-identity/defs.ts b/apps/server/src/modules/sources-identity/defs.ts index c6f81468..b9236ccc 100644 --- a/apps/server/src/modules/sources-identity/defs.ts +++ b/apps/server/src/modules/sources-identity/defs.ts @@ -96,6 +96,8 @@ const TE = ["media-entities"]; const TI = ["identity"]; const TD = ["decisions"]; +export const DecisionMinSeverity = z.enum(["info", "warning", "critical"]); + export const D = { sites: { list: { @@ -297,7 +299,12 @@ export const D = { "/decisions", "待处理事项(缺省只列未关闭)", Decision, - { status: DecisionStatus.optional(), kind: DecisionKind.optional() }, + { + status: DecisionStatus.optional(), + kind: DecisionKind.optional(), + /** 只看不低于此级别的事项;收件箱缺省传 warning 隐藏"仅信息"类 */ + minSeverity: DecisionMinSeverity.optional(), + }, TD, ), get: get("/decisions/{id}", "待处理事项详情", Decision, TD), diff --git a/apps/server/src/modules/sources-identity/routes/decisions.ts b/apps/server/src/modules/sources-identity/routes/decisions.ts index 3465e513..273ba903 100644 --- a/apps/server/src/modules/sources-identity/routes/decisions.ts +++ b/apps/server/src/modules/sources-identity/routes/decisions.ts @@ -4,12 +4,13 @@ import { sql } from "@umt/db"; import { z } from "zod"; import { ApiException } from "../../../http/errors.ts"; import type { ModuleApp } from "../../types.ts"; -import { D } from "../defs.ts"; +import { D, DecisionMinSeverity } from "../defs.ts"; import { route } from "../http.ts"; import { decodeCursor, jsonb, page, toApi, toApiList } from "../rows.ts"; import type { SourcesRuntime } from "../runtime.ts"; import { audit } from "../service/common.ts"; import { assignReleaseIdentity, dismissPoolEntry } from "../service/identity-admin.ts"; +import { groupMemberIds } from "../service/identity-decisions.ts"; // 待处理中心数据源:decision 列表/详情/聚合/处理。 // 身份类(歧义 / 待识别)由本模块直接执行所选候选或手动指定;其它类型只记录处理结果并发 decision.updated, @@ -17,6 +18,11 @@ import { assignReleaseIdentity, dismissPoolEntry } from "../service/identity-adm const OPEN = ["open", "investigating", "awaiting_user"] as const; const T = ["decisions"]; +const SEV_AT_LEAST = { + info: ["info", "warning", "critical"], + warning: ["warning", "critical"], + critical: ["critical"], +} as const; const SummaryGroup = z.object({ kind: DecisionKind, @@ -36,6 +42,7 @@ export function decisionRoutes(app: ModuleApp, rt: () => SourcesRuntime) { let q = db().selectFrom("decision").selectAll(); q = query.status ? q.where("status", "=", query.status) : q.where("status", "in", [...OPEN]); if (query.kind) q = q.where("kind", "=", query.kind); + if (query.minSeverity) q = q.where("severity", "in", [...SEV_AT_LEAST[query.minSeverity]]); const rows = await q .orderBy(sql`case severity when 'critical' then 0 when 'warning' then 1 else 2 end`) .orderBy("created_at", "desc") @@ -55,11 +62,12 @@ export function decisionRoutes(app: ModuleApp, rt: () => SourcesRuntime) { scope: "read", summary: "未关闭待处理事项按作品与问题类型聚合", tags: T, - query: z.object({ status: DecisionStatus.optional() }), + query: z.object({ status: DecisionStatus.optional(), minSeverity: DecisionMinSeverity.optional() }), response: z.object({ total: z.number().int(), groups: z.array(SummaryGroup) }), }, async ({ query }) => { const statuses = query.status ? [query.status] : [...OPEN]; + const severities: string[] = [...SEV_AT_LEAST[query.minSeverity ?? "info"]]; // 作品归属:evidence.mediaEntityId,或 subject 为 media_entity,或 subject 为已识别的 source_release const rows = await sql<{ kind: z.infer; @@ -75,7 +83,7 @@ export function decisionRoutes(app: ModuleApp, rt: () => SourcesRuntime) { case when d.subject_type = 'media_entity' then d.subject_id end, (select r.media_entity_id from source_release r where d.subject_type = 'source_release' and r.id = d.subject_id) ) as entity_id - from decision d where d.status = any(${statuses}) + from decision d where d.status = any(${statuses}) and d.severity = any(${severities}) ) select d.kind, d.entity_id, e.title as entity_title, count(*)::int as count, max(case d.severity when 'critical' then 2 when 'warning' then 1 else 0 end)::int as max_sev, @@ -104,6 +112,44 @@ export function decisionRoutes(app: ModuleApp, rt: () => SourcesRuntime) { return toApi(row); }); + /** 组内其它资源:忽略 → 各自池条目 dismissed;指定 → 关联到代表资源已确定的作品(同样加人工锁)。单条失败不影响其它 */ + async function applyToGroup( + decisionId: string, + subjectId: string, + others: string[], + dismissing: boolean, + note: string | null, + actor: Parameters[2], + ) { + const subject = await db() + .selectFrom("source_release") + .select(["media_entity_id"]) + .where("id", "=", subjectId) + .executeTakeFirst(); + if (!dismissing && !subject?.media_entity_id) return; + for (const srId of others) { + try { + if (dismissing) { + const pool = await db() + .selectFrom("identity_candidate") + .select(["id"]) + .where("source_release_id", "=", srId) + .executeTakeFirst(); + if (pool) await dismissPoolEntry(rt(), pool.id, actor, note); + } else { + await assignReleaseIdentity( + rt(), + srId, + { mediaEntityId: subject!.media_entity_id!, lock: true, reason: note ?? `decision ${decisionId}` }, + actor, + ); + } + } catch (e) { + rt().ctx.logger.warn({ err: e, srId, decisionId }, "合并事项组内资源处理失败"); + } + } + } + route(app, D.decisions.resolve, async ({ params, body, actor }) => { const d = await db().selectFrom("decision").selectAll().where("id", "=", params.id).executeTakeFirst(); if (!d) throw new ApiException("not_found", "待处理事项不存在"); @@ -118,7 +164,10 @@ export function decisionRoutes(app: ModuleApp, rt: () => SourcesRuntime) { const identityKind = d.kind === "identity_ambiguous" || d.kind === "identity_unresolved"; if (identityKind && d.subject_type === "source_release" && d.subject_id) { const requires = opt?.input?.requires; - if (body.dismiss || opt?.key === "dismiss") { + // 合并事项(同来源同标题):先处理代表资源,再把结果套到组内其它资源 + const others = d.dedupe_key.startsWith("identity-group:") ? await groupMemberIds(db(), d.id, d.subject_id) : []; + const dismissing = body.dismiss || opt?.key === "dismiss"; + if (dismissing) { const pool = await db() .selectFrom("identity_candidate") .select(["id"]) @@ -163,6 +212,7 @@ export function decisionRoutes(app: ModuleApp, rt: () => SourcesRuntime) { actor, ); } + if (others.length) await applyToGroup(d.id, d.subject_id, others, dismissing, body.note ?? null, actor); } // 站点登录需人工验证:"重试登录" = 入队刷新登录(结果另行更新账号状态与待处理) if (d.dedupe_key?.startsWith("site-login:") && opt?.key === "retry") { diff --git a/apps/server/src/modules/sources-identity/routes/identity.ts b/apps/server/src/modules/sources-identity/routes/identity.ts index dfc1f7f7..02589aee 100644 --- a/apps/server/src/modules/sources-identity/routes/identity.ts +++ b/apps/server/src/modules/sources-identity/routes/identity.ts @@ -628,7 +628,8 @@ export function identityRoutes(app: ModuleApp, rt: () => SourcesRuntime) { if (!pool) throw new ApiException("not_found", "待识别条目不存在"); const r = await rt().ctx.engine.enqueue( "resolveObservation", - { sourceReleaseId: pool.source_release_id, force: true }, + // 用户点"重新识别" = 手动触发:即使不相关也查 TMDB(惰性识别的 c 路径) + { sourceReleaseId: pool.source_release_id, force: true, trigger: "manual" }, { idempotencyKey: `resolve:retry:${params.id}:${Math.floor(Date.now() / 10_000)}`, traceId: randomUUID() }, ); return { taskId: r.runId }; diff --git a/apps/server/src/modules/sources-identity/scheduler.ts b/apps/server/src/modules/sources-identity/scheduler.ts index faa3167c..29ef66b8 100644 --- a/apps/server/src/modules/sources-identity/scheduler.ts +++ b/apps/server/src/modules/sources-identity/scheduler.ts @@ -1,6 +1,7 @@ import { randomUUID } from "node:crypto"; import { sql } from "@umt/db"; import type { SourcesRuntime } from "./runtime.ts"; +import { archiveStalePool } from "./service/identity-decisions.ts"; // worker 后台调度:按 Feed 轮询策略发 feed.due(outbox → ingestFeed)、按站点快照周期入队 snapshotSite、 // 按待识别池 next_review_at 入队复查。幂等键按时间槽生成,多 worker 并发也不会重复。 @@ -86,13 +87,17 @@ export async function tickSnapshots(rt: SourcesRuntime): Promise { return sites.rows.length; } -/** 待识别池复查:到期的先顺延一个租约,避免下个 tick 重复入队 */ +/** + * 待识别池复查:到期的先顺延一个租约,避免下个 tick 重复入队;顺带(每小时至多一次)归档过期的 unmatched 条目。 + * 幂等键带上到期时间:别名触发器唤醒 / 证据版本变化后计数归 1 时,不会撞上旧的复查键而被吞掉。 + */ export async function tickReviews(rt: SourcesRuntime): Promise { const { ctx } = rt; const now = rt.now(); + await archiveStalePool(rt); const reviews = await ctx.db.transaction().execute(async (trx) => { - const rows = await sql<{ id: string; source_release_id: string; attempts: number }>` - select id, source_release_id, attempts from identity_candidate + const rows = await sql<{ id: string; source_release_id: string; attempts: number; due: Date }>` + select id, source_release_id, attempts, next_review_at as due from identity_candidate where status = 'pending' and next_review_at is not null and next_review_at <= ${now} order by next_review_at limit 50 for update skip locked`.execute(trx); for (const r of rows.rows) { @@ -108,7 +113,7 @@ export async function tickReviews(rt: SourcesRuntime): Promise { await ctx.engine.enqueue( "resolveObservation", { sourceReleaseId: r.source_release_id, force: true }, - { idempotencyKey: `review:${r.id}:${r.attempts}`, traceId: randomUUID() }, + { idempotencyKey: `review:${r.id}:${r.attempts}:${new Date(r.due).getTime()}`, traceId: randomUUID() }, ); } return reviews.length; diff --git a/apps/server/src/modules/sources-identity/service/identity-admin.ts b/apps/server/src/modules/sources-identity/service/identity-admin.ts index f555448a..d8bc7000 100644 --- a/apps/server/src/modules/sources-identity/service/identity-admin.ts +++ b/apps/server/src/modules/sources-identity/service/identity-admin.ts @@ -12,7 +12,7 @@ import { toParsedColumns } from "@umt/mapping"; import { ApiException } from "../../../http/errors.ts"; import { jsonb } from "../rows.ts"; import type { SourcesRuntime } from "../runtime.ts"; -import { type Actor, audit, closeDecisions, type Exec } from "./common.ts"; +import { type Actor, audit, type Exec } from "./common.ts"; import { acceptIdentity, createEntityFromTmdb, @@ -22,6 +22,7 @@ import { looksLikeVariety, resolveEntityId, } from "./entities.ts"; +import { detachFromIdentityDecisions } from "./identity-decisions.ts"; import { recordAttempt } from "./resolve.ts"; import { backfillRecognizeCache } from "./tmdb-recognize-cache.ts"; @@ -149,10 +150,10 @@ export async function assignReleaseIdentity( }) .where("source_release_id", "=", srId) .execute(); - await closeDecisions( + await detachFromIdentityDecisions( ctx, trx, - `identity:${srId}`, + srId, "resolved", { mediaEntityId: entityId, by: "manual" }, actorName(actor), @@ -468,7 +469,7 @@ export async function dismissPoolEntry(rt: SourcesRuntime, poolId: string, actor .where("id", "=", row.source_release_id) .where("resolution_locked", "=", false) .execute(); - await closeDecisions(ctx, trx, `identity:${row.source_release_id}`, "dismissed", { note }, actorName(actor)); + await detachFromIdentityDecisions(ctx, trx, row.source_release_id, "dismissed", { note }, actorName(actor)); await audit(trx, actor, "identity.pool_dismissed", { type: "identity_candidate", id: poolId }, { note }); return row; }); diff --git a/apps/server/src/modules/sources-identity/service/identity-decisions.ts b/apps/server/src/modules/sources-identity/service/identity-decisions.ts new file mode 100644 index 00000000..aa850a39 --- /dev/null +++ b/apps/server/src/modules/sources-identity/service/identity-decisions.ts @@ -0,0 +1,228 @@ +import type { DecisionOption, PoolCandidate } from "@umt/contracts"; +import { sql } from "@umt/db"; +import type { ServerContext } from "../../types.ts"; +import { jsonb } from "../rows.ts"; +import type { SourcesRuntime } from "../runtime.ts"; +import { closeDecisions, type Exec, openDecision } from "./common.ts"; + +// DEC-1 身份类待处理的合并与池治理(REQ §7.5、§13): +// - 只有与订阅 / 本地作品相关(本地别名命中)或已查过 TMDB 仍有歧义的资源才开事项; +// - 同一来源 + 同一归一化标题合并为一条事项(dedupeKey identity-group::),成员 = 池里 +// decision_id 指向它且仍 pending 的条目;事项的 subject 与候选选项取自代表资源(最近一次进来的那条); +// - 成员被识别 / 指定 / 忽略时从组里摘除,组空了才关闭事项; +// - unmatched 池条目超过 identity.poolArchiveDays 没有新证据就归档。 + +const OPEN = ["open", "investigating", "awaiting_user"] as const; +/** 事项证据里最多列多少条成员资源(完整列表在待识别池按标题检索) */ +export const GROUP_LIST_LIMIT = 50; + +export const groupDedupeKey = (siteId: string, groupKey: string) => `identity-group:${siteId}:${groupKey}`; + +/** 身份类事项的手动指定选项;resolve 时按 input.requires 取 DecisionResolveInput.identity 的对应字段 */ +function manualIdentityOptions(sourceReleaseId: string): DecisionOption[] { + return [ + { + key: "manual_tmdb", + label: "手动指定 TMDB ID", + description: "按 TMDB 电影或剧集 ID 建档并关联,加人工锁", + input: { sourceReleaseId, requires: "tmdb" }, + }, + { + key: "local_entity", + label: "选本地作品", + description: "关联到已有的本地作品,加人工锁", + input: { sourceReleaseId, requires: "mediaEntityId" }, + }, + ]; +} + +export function identityOptions(sourceReleaseId: string, candidates: readonly PoolCandidate[]): DecisionOption[] { + return [ + ...candidates.slice(0, 5).map((c, i) => ({ + key: `c${i}`, + label: `${c.title}${c.year ? ` (${c.year})` : ""} · ${c.localEntityId ? "本地作品" : `${c.provider}:${c.entityType}/${c.externalId}`}`, + input: { sourceReleaseId, candidateIndex: i }, + })), + ...manualIdentityOptions(sourceReleaseId), + { key: "dismiss", label: "忽略此资源" }, + ]; +} + +interface Member { + source_release_id: string; + title: string; + reason: string; + candidates: unknown; + updated_at: Date; +} + +async function members(exec: Exec, decisionId: string, exclude: string | null): Promise { + const rows = await sql` + select p.source_release_id, r.title, p.reason, p.candidates, p.updated_at + from identity_candidate p join source_release r on r.id = p.source_release_id + where p.decision_id = ${decisionId} and p.status = 'pending' + and (${exclude}::uuid is null or p.source_release_id <> ${exclude}::uuid) + order by p.updated_at desc, p.source_release_id`.execute(exec); + return rows.rows; +} + +/** 合并事项的成员资源 ID(不含 subject 本身时传 exclude) */ +export async function groupMemberIds(exec: Exec, decisionId: string, exclude: string | null = null) { + return (await members(exec, decisionId, exclude)).map((m) => m.source_release_id); +} + +function titleFor( + kind: "identity_ambiguous" | "identity_unresolved", + siteName: string, + n: number, + one: string, + work: string | null, +) { + const what = kind === "identity_ambiguous" ? "待确认身份" : "待识别"; + if (n <= 1) return `${what}:${one}`; + return `来源「${siteName}」的 ${n} 条资源${what}${work ? `:${work}` : ""}`; +} + +/** + * 按当前成员重写合并事项的标题 / 摘要 / 成员清单;代表资源 = representative(缺省取最近更新的成员)。 + * 组内没有成员时关闭事项并返回 false。 + */ +export async function refreshGroupDecision( + ctx: ServerContext, + trx: Exec, + decisionId: string, + opts: { + representative?: string | null; + exclude?: string | null; + closeAs?: { status: "resolved" | "dismissed"; resolution: Record; by: string }; + } = {}, +): Promise { + const d = await trx + .selectFrom("decision") + .selectAll() + .where("id", "=", decisionId) + .where("status", "in", [...OPEN]) + .executeTakeFirst(); + if (!d) return false; + const list = await members(trx, decisionId, opts.exclude ?? null); + if (!list.length) { + const close = opts.closeAs ?? { status: "resolved" as const, resolution: { by: "group_empty" }, by: "system" }; + await closeDecisions(ctx, trx, d.dedupe_key, close.status, close.resolution, close.by); + return false; + } + const rep = list.find((m) => m.source_release_id === opts.representative) ?? list[0]!; + const ev = (d.evidence ?? {}) as Record; + const group = (ev.group ?? {}) as { siteName?: string; title?: string | null }; + const kind = (d.kind === "identity_ambiguous" ? "identity_ambiguous" : "identity_unresolved") as + | "identity_ambiguous" + | "identity_unresolved"; + const candidates = (Array.isArray(rep.candidates) ? rep.candidates : []) as PoolCandidate[]; + const title = titleFor(kind, group.siteName ?? "未知来源", list.length, rep.title, group.title ?? null); + await trx + .updateTable("decision") + .set({ + title, + subject_type: "source_release", + subject_id: rep.source_release_id, + options: jsonb(identityOptions(rep.source_release_id, candidates)), + evidence: jsonb({ + ...ev, + sourceReleaseId: rep.source_release_id, + count: list.length, + releases: list + .slice(0, GROUP_LIST_LIMIT) + .map((m) => ({ sourceReleaseId: m.source_release_id, title: m.title })), + }), + }) + .where("id", "=", decisionId) + .execute(); + return true; +} + +export interface GroupDecisionInput { + kind: "identity_ambiguous" | "identity_unresolved"; + siteId: string; + /** 分组键:归一化标题;解析不出标题时用资源 ID(不合并) */ + groupKey: string; + /** 分组展示用的作品标题(解析标题) */ + workTitle: string | null; + sourceReleaseId: string; + releaseTitle: string; + summary: string; + candidates: PoolCandidate[]; + evidence: Record; + nextReviewAt: Date | null; +} + +/** 开 / 并入合并事项;返回事项 ID(调用方随后把池条目 decision_id 指向它,再 refreshGroupDecision) */ +export async function openGroupDecision(ctx: ServerContext, trx: Exec, g: GroupDecisionInput): Promise { + const site = await trx.selectFrom("site").select(["name"]).where("id", "=", g.siteId).executeTakeFirst(); + const d = await openDecision(ctx, trx, { + kind: g.kind, + // 与订阅 / 本地作品相关的身份问题会卡住追更,不是"仅信息" + severity: "warning", + title: titleFor(g.kind, site?.name ?? "未知来源", 1, g.releaseTitle, g.workTitle), + summary: g.summary, + subjectType: "source_release", + subjectId: g.sourceReleaseId, + dedupeKey: groupDedupeKey(g.siteId, g.groupKey), + evidence: { + ...g.evidence, + group: { siteId: g.siteId, siteName: site?.name ?? null, key: g.groupKey, title: g.workTitle }, + }, + options: identityOptions(g.sourceReleaseId, g.candidates), + nextReviewAt: g.nextReviewAt, + }); + return d.id; +} + +/** + * 资源已识别 / 已指定 / 已忽略:关闭它的旧式单条事项(identity:),并从合并事项里摘除;组空了才关闭合并事项。 + */ +export async function detachFromIdentityDecisions( + ctx: ServerContext, + trx: Exec, + srId: string, + status: "resolved" | "dismissed", + resolution: Record, + by: string, +): Promise { + await closeDecisions(ctx, trx, `identity:${srId}`, status, resolution, by); + const pool = await trx + .selectFrom("identity_candidate") + .select(["decision_id"]) + .where("source_release_id", "=", srId) + .executeTakeFirst(); + if (!pool?.decision_id) return; + const d = await trx + .selectFrom("decision") + .select(["id", "dedupe_key"]) + .where("id", "=", pool.decision_id) + .where("status", "in", [...OPEN]) + .executeTakeFirst(); + if (!d?.dedupe_key.startsWith("identity-group:")) return; + await refreshGroupDecision(ctx, trx, d.id, { exclude: srId, closeAs: { status, resolution, by } }); +} + +const ARCHIVE_EVERY_MS = 3600_000; +let lastArchiveAt = 0; + +/** + * 过期归档:unmatched 池条目超过 identity.poolArchiveDays 没有新证据 → archived(仍可按状态筛出、可手动指定; + * 同一资源再被观察到或新增相近别名时回到 pending)。每进程每小时最多执行一次。 + */ +export async function archiveStalePool(rt: SourcesRuntime, opts: { force?: boolean } = {}): Promise { + const now = rt.now(); + if (!opts.force && now.getTime() - lastArchiveAt < ARCHIVE_EVERY_MS) return 0; + lastArchiveAt = now.getTime(); + const { poolArchiveDays } = await rt.ctx.settings.get("identity"); + const cutoff = new Date(now.getTime() - poolArchiveDays * 86_400_000); + const r = await rt.ctx.db + .updateTable("identity_candidate") + .set({ status: "archived", next_review_at: null }) + .where("status", "=", "pending") + .where("reason", "=", "unmatched") + .where((eb) => eb(eb.fn.coalesce("last_attempt_at", "created_at"), "<", cutoff)) + .executeTakeFirst(); + return Number(r.numUpdatedRows ?? 0); +} diff --git a/apps/server/src/modules/sources-identity/service/resolve.ts b/apps/server/src/modules/sources-identity/service/resolve.ts index fcb4ebf4..d877c064 100644 --- a/apps/server/src/modules/sources-identity/service/resolve.ts +++ b/apps/server/src/modules/sources-identity/service/resolve.ts @@ -22,7 +22,7 @@ import { } from "@umt/mapping"; import { jsonb } from "../rows.ts"; import type { SourcesRuntime } from "../runtime.ts"; -import { type Actor, closeDecisions, type Exec, openDecision, SYSTEM } from "./common.ts"; +import { type Actor, closeDecisions, type Exec, SYSTEM } from "./common.ts"; import { addAliasIfAbsent, createEntityFromTmdb, @@ -32,12 +32,21 @@ import { looksLikeVariety, resolveEntityId, } from "./entities.ts"; +import { detachFromIdentityDecisions, openGroupDecision, refreshGroupDecision } from "./identity-decisions.ts"; import { loadRecognizeCache, type RecognizeMeta, recognizeMetaOf, saveRecognizeCache } from "./tmdb-recognize-cache.ts"; -// resolveObservation:parseRelease → 本地召回(别名 pg_trgm + 已有外部身份 + 人工锁)→ TMDB 候选 → domain.resolveIdentity。 +// resolveObservation:parseRelease → 本地召回(别名 pg_trgm + 已有外部身份 + 人工锁)→ [按需] TMDB 候选 → domain.resolveIdentity。 // auto 才建/关联正式作品;ambiguous / no_match / 查询失败进待识别池(候选只存 JSON),有界退避复查。 +// +// DEC-1 惰性识别(对照 MP:RSS / 站点资源只在匹配订阅或用户搜索时识别,识别失败不告警):TMDB 只在 lookupTrigger +// 给出理由时查询——命中本地作品别名(订阅 / 追踪 / 已登记的作品)、已发布的新作品收录规则前置条件通过、用户手动识别、 +// 搜索结果、池条目此前已查过 TMDB(复查延续),或设置 identity.eagerTmdbLookup 打开。其余只进池(reason=unmatched), +// 不开待处理。待处理只收"相关"的:与本地作品相关却无法识别,或查过 TMDB 仍有歧义;同来源同标题合并为一条。 const MAX_REVIEW_ATTEMPTS = 8; +/** unmatched(只做了本地召回)的周期复查:每天一次、最多 3 次;新增相近别名时由触发器立即唤醒,不受此上限 */ +const UNMATCHED_REVIEW_MS = 24 * 3600_000; +const MAX_UNMATCHED_REVIEWS = 3; const TRGM_THRESHOLD = 0.45; export interface ResolveInput { @@ -45,10 +54,26 @@ export interface ResolveInput { sourceReleaseId?: string | undefined; /** 复查/手动重试:忽略"已识别且证据未变"的短路 */ force?: boolean | undefined; + /** + * 谁触发的识别,决定是否查 TMDB(惰性识别):observed = 新观察到的资源(缺省);review = 池周期复查(force 缺省); + * manual = 用户在资源列表 / 待识别池点开或重新识别、Agent 调 identity.resolve;search = 搜索结果。 + * manual / search 总是查 TMDB;observed / review 只在命中本地别名、新作品收录规则或设置打开时查。 + */ + trigger?: ResolveTrigger | undefined; } +export type ResolveTrigger = "observed" | "review" | "manual" | "search"; + export interface ResolveResult { - outcome: "auto" | "proposed" | "ambiguous" | "no_match" | "lookup_failed" | "skipped_locked" | "skipped"; + outcome: + | "auto" + | "proposed" + | "ambiguous" + | "no_match" + | "lookup_failed" + | "unmatched" + | "skipped_locked" + | "skipped"; sourceReleaseId: string | null; mediaEntityId: string | null; contentUnitIds: string[]; @@ -251,12 +276,22 @@ export async function resolveRelease(rt: SourcesRuntime, input: ResolveInput): P } const parsed = parseRelease(release.title, release.subtitle ? { subtitle: release.subtitle } : {}); const cols = toParsedColumns(parsed); + const trigger: ResolveTrigger = input.trigger ?? (input.force ? "review" : "observed"); if (!parsed.title || parsed.auxiliaryOnly) { - return toPool(rt, release, observationId, parsed, null, [], "no_match", "unparseable_title", null); + return toPool(rt, release, observationId, parsed, null, [], "unmatched", "unparseable_title", null, { + tracked: false, + lookup: null, + }); } - // 召回:本地优先,再 TMDB;TMDB 失败不等于无结果 + // 召回:本地优先;TMDB 只在有理由时查(惰性识别),查询失败不等于无结果 const local = await localCandidates(ctx.db, parsed, release); + const lookup = await lookupReason(rt, parsed, trigger, local.length > 0, pool?.reason ?? null); + const scope = { tracked: local.length > 0, lookup }; + if (!lookup) { + // 不相关的资源:只进池(可检索、可手动指定),不查 TMDB、不开待处理 + return toPool(rt, release, observationId, parsed, null, [], "unmatched", "no_local_match", null, scope); + } let remote: MetadataCandidate[] = []; let lookupError: string | null = null; let recognize: { cacheHit: boolean; meta: RecognizeMeta | null } = { cacheHit: false, meta: null }; @@ -321,7 +356,7 @@ export async function resolveRelease(rt: SourcesRuntime, input: ResolveInput): P } } } - return bindAuto(rt, release, observationId, parsed, cols, resolution, version, detailed ?? null); + return bindAuto(rt, release, observationId, parsed, cols, resolution, version, detailed ?? null, lookup); } if (resolution.status === "auto" && !autoAccept) { @@ -335,6 +370,7 @@ export async function resolveRelease(rt: SourcesRuntime, input: ResolveInput): P "proposed", "automation_off", version, + scope, ); } if (lookupError && resolution.status !== "ambiguous") { @@ -348,6 +384,7 @@ export async function resolveRelease(rt: SourcesRuntime, input: ResolveInput): P "lookup_failed", lookupError, version, + scope, ); } return toPool( @@ -360,14 +397,60 @@ export async function resolveRelease(rt: SourcesRuntime, input: ResolveInput): P resolution.status === "ambiguous" ? "ambiguous" : "no_match", resolution.reason, version, + scope, ); } +/** 为什么要查 TMDB;null = 不查(惰性识别) */ +export type LookupReason = "tracked_alias" | "manual" | "search" | "previously_looked_up" | "eager" | "new_title_rule"; + +async function lookupReason( + rt: SourcesRuntime, + parsed: ParsedRelease, + trigger: ResolveTrigger, + tracked: boolean, + poolReason: string | null, +): Promise { + if (trigger === "manual") return "manual"; + if (trigger === "search") return "search"; + if (tracked) return "tracked_alias"; + // 池条目此前已查过 TMDB(旧行为 / 手动 / 规则触发):复查延续,仍受识别缓存与有界退避约束 + if (trigger === "review" && poolReason && poolReason !== "unmatched") return "previously_looked_up"; + if ((await rt.ctx.settings.get("identity")).eagerTmdbLookup) return "eager"; + if (await passesNewTitlePrefilter(rt, parsed)) return "new_title_rule"; + return null; +} + +/** + * 新作品收录规则的前置硬条件:有已发布且开启 newTitle 的获取策略(并有有效持续授权),资源的类型落在规则的 + * mediaKinds 内,标题年份(若有)不早于窗口能覆盖的年份。通过才值得查 TMDB 拿首播 / 上映日期交给评估判断。 + */ +async function passesNewTitlePrefilter(rt: SourcesRuntime, parsed: ParsedRelease): Promise { + const rows = await sql<{ rules: unknown }>` + select p.rules from policy_revision p + where p.kind = 'acquisition' and p.status = 'published' + and coalesce((p.rules->'newTitle'->>'enabled')::boolean, false) + and exists (select 1 from automation_grant g where g.policy_revision_id = p.id and g.status = 'active')`.execute( + rt.ctx.db, + ); + if (!rows.rows.length) return false; + const tvLike = parsed.mediaTypeHint === "tv" || parsed.season !== undefined || parsed.episodes !== undefined; + const nowYear = rt.now().getUTCFullYear(); + return rows.rows.some((r) => { + const nt = ((r.rules ?? {}) as { newTitle?: { windowDays?: number; mediaKinds?: string[] } }).newTitle ?? {}; + const kinds = nt.mediaKinds ?? []; + if (kinds.length && !kinds.some((k) => (tvLike ? k === "series" || k === "variety" : k === "movie"))) return false; + // 与评估同一缺省(acquisition-organize ports.newTitleWindowDays = 30) + const windowYears = Math.ceil((nt.windowDays ?? 30) / 365); + return parsed.year === undefined || parsed.year >= nowYear - windowYears; + }); +} + async function recordAttempt( exec: Exec, srId: string, observationId: string | null, - outcome: "auto" | "proposed" | "ambiguous" | "no_match" | "lookup_failed" | "manual" | "skipped_locked", + outcome: "auto" | "proposed" | "ambiguous" | "no_match" | "lookup_failed" | "unmatched" | "manual" | "skipped_locked", reason: string | null, version: string, count: number, @@ -407,6 +490,7 @@ async function bindAuto( resolution: IdentityResolution, version: string, tmdbDetail: MetadataCandidate | null, + lookup: LookupReason, ): Promise { const { ctx } = rt; const best = resolution.best!; @@ -458,6 +542,7 @@ async function bindAuto( evidenceVersion: version, best: rankedSummary(best), candidateCount: resolution.ranked.length, + lookup, }; await trx .updateTable("source_release") @@ -484,10 +569,10 @@ async function bindAuto( .where("source_release_id", "=", release.id) .where("status", "=", "pending") .execute(); - await closeDecisions( + await detachFromIdentityDecisions( ctx, trx, - `identity:${release.id}`, + release.id, "resolved", { mediaEntityId: entityId, by: "auto" }, "system", @@ -532,29 +617,12 @@ async function bindAuto( } function reviewDelayMs(reason: string, attempts: number): number | null { + if (reason === "unmatched") return attempts >= MAX_UNMATCHED_REVIEWS ? null : UNMATCHED_REVIEW_MS; if (attempts >= MAX_REVIEW_ATTEMPTS) return null; const base = reason === "lookup_failed" ? 15 * 60_000 : 3600_000; return Math.min(base * 2 ** Math.max(0, attempts - 1), 7 * 24 * 3600_000); } -/** 身份类事项的手动指定选项;resolve 时按 input.requires 取 DecisionResolveInput.identity 的对应字段 */ -function manualIdentityOptions(sourceReleaseId: string) { - return [ - { - key: "manual_tmdb", - label: "手动指定 TMDB ID", - description: "按 TMDB 电影或剧集 ID 建档并关联,加人工锁", - input: { sourceReleaseId, requires: "tmdb" }, - }, - { - key: "local_entity", - label: "选本地作品", - description: "关联到已有的本地作品,加人工锁", - input: { sourceReleaseId, requires: "mediaEntityId" }, - }, - ]; -} - /** 事项摘要里给人看的原因;原始代码仍记在 resolution_evidence */ const DETAIL_TEXT: Record = { no_viable_candidate: "没有可信候选", @@ -562,9 +630,18 @@ const DETAIL_TEXT: Record = { weak_evidence: "证据不足", multiple_candidates_same_level: "多个候选同级", automation_off: "自动化已关闭", + no_local_match: "没有命中订阅或本地作品,未查询 TMDB", }; const detailText = (d: string) => DETAIL_TEXT[d] ?? d; +type PoolReason = "ambiguous" | "no_match" | "proposed" | "lookup_failed" | "unmatched"; + +/** 这次识别的相关性:tracked = 本地别名命中(订阅 / 追踪 / 已登记作品);lookup = 查 TMDB 的理由(null = 未查) */ +interface ResolveScope { + tracked: boolean; + lookup: LookupReason | null; +} + async function toPool( rt: SourcesRuntime, release: SourceReleaseRow, @@ -572,9 +649,10 @@ async function toPool( parsed: ParsedRelease, resolution: IdentityResolution | null, ranked: RankedCandidate[], - reason: "ambiguous" | "no_match" | "proposed" | "lookup_failed", + reason: PoolReason, detail: string, versionIn: string | null, + scope: ResolveScope, ): Promise { const { ctx } = rt; const version = versionIn ?? evidenceVersion(parsed, ranked); @@ -584,14 +662,16 @@ async function toPool( .slice(0, 10) .map(poolCandidate); const now = rt.now(); + const normalized = parsed.title ? normalizeTitle(parsed.title) || null : null; const query = { title: parsed.title ?? null, titleCn: parsed.titleCn ?? null, titleEn: parsed.titleEn ?? null, - normalized: parsed.title ? normalizeTitle(parsed.title) : null, + normalized, year: parsed.year ?? null, season: parsed.season ?? null, episodes: parsed.episodes ?? null, + lookup: scope.lookup, }; const contractStatus = reason === "proposed" @@ -603,7 +683,7 @@ async function toPool( return ctx.db.transaction().execute(async (trx) => { const prev = await trx .selectFrom("identity_candidate") - .select(["id", "attempts", "evidence_version"]) + .select(["id", "attempts", "evidence_version", "decision_id"]) .where("source_release_id", "=", release.id) .executeTakeFirst(); // 证据版本变化 = 新证据:从头计数;同一证据反复复查按退避递增,达上限停止自动复查(仍可检索) @@ -611,7 +691,6 @@ async function toPool( const delay = reviewDelayMs(reason, attempts); const nextReviewAt = delay === null ? null : new Date(now.getTime() + delay); - let decisionId: string | null = null; const evidence = { sourceReleaseId: release.id, siteId: release.site_id, @@ -620,64 +699,54 @@ async function toPool( evidenceVersion: version, query, }; - if ((reason === "ambiguous" || reason === "proposed") && candidates.length > 0) { - const d = await openDecision(ctx, trx, { - kind: "identity_ambiguous", - severity: "info", - title: `待确认身份:${release.title}`, - summary: `${candidates.length} 个候选证据等级相近(${detailText(detail)})。选择一个或忽略`, - subjectType: "source_release", - subjectId: release.id, - dedupeKey: `identity:${release.id}`, + // 只收相关的:查过 TMDB 仍有歧义(有候选可选),或与本地作品相关却无结果 / 查询失败且复查已停 + const wantsDecision = + reason === "ambiguous" || reason === "proposed" + ? candidates.length > 0 + : scope.tracked && (reason === "no_match" || (reason === "lookup_failed" && delay === null)); + let decisionId: string | null = null; + if (wantsDecision) { + const ambiguous = (reason === "ambiguous" || reason === "proposed") && candidates.length > 0; + decisionId = await openGroupDecision(ctx, trx, { + kind: ambiguous ? "identity_ambiguous" : "identity_unresolved", + siteId: release.site_id, + groupKey: normalized ?? release.id, + workTitle: parsed.title ?? null, + sourceReleaseId: release.id, + releaseTitle: release.title, + summary: ambiguous + ? `${candidates.length} 个候选证据等级相近(${detailText(detail)})。选择一个或忽略;同组资源一并处理` + : reason === "lookup_failed" + ? `元数据查询多次失败(${detail}),已停止自动复查。手动指定作品或忽略;同组资源一并处理` + : `标题与本地作品相关,但没有匹配的作品(${detailText(detail)})。手动指定 TMDB ID、选本地作品或忽略;同组资源一并处理`, + candidates, evidence, - options: [ - ...candidates.slice(0, 5).map((c, i) => ({ - key: `c${i}`, - label: `${c.title}${c.year ? ` (${c.year})` : ""} · ${c.localEntityId ? "本地作品" : `${c.provider}:${c.entityType}/${c.externalId}`}`, - input: { sourceReleaseId: release.id, candidateIndex: i }, - })), - ...manualIdentityOptions(release.id), - { key: "dismiss", label: "忽略此资源" }, - ], nextReviewAt, }); - decisionId = d.id; - } else if (reason !== "lookup_failed" || delay === null) { - // 没有可选候选(无结果 / 无法解析 / 查询失败且自动复查已停):只能人工指定,同样进待处理 - const d = await openDecision(ctx, trx, { - kind: "identity_unresolved", - severity: "info", - title: `待识别:${release.title}`, - summary: - reason === "lookup_failed" - ? `元数据查询多次失败(${detail}),已停止自动复查。手动指定作品或忽略` - : `没有匹配的作品(${detailText(detail)})。手动指定 TMDB ID、选本地作品或忽略`, - subjectType: "source_release", - subjectId: release.id, - dedupeKey: `identity:${release.id}`, - evidence, - options: [ - ...manualIdentityOptions(release.id), - { - key: "dismiss", - label: "忽略此资源", - }, - ], - nextReviewAt, - }); - decisionId = d.id; + } else if (prev?.decision_id) { + // 不再需要人工处理(例如别名删除后变得不相关):从原合并事项里摘除 + await detachFromIdentityDecisions(ctx, trx, release.id, "resolved", { by: "no_longer_relevant" }, "system"); } await sql` - insert into identity_candidate (source_release_id, status, reason, query, candidates, evidence_version, attempts, - last_attempt_at, next_review_at, decision_id) - values (${release.id}, 'pending', ${reason}, ${jsonb(query)}::jsonb, ${jsonb(candidates)}::jsonb, ${version}, ${attempts}, - ${now}, ${nextReviewAt}, ${decisionId}) + insert into identity_candidate (source_release_id, status, reason, query, normalized_title, candidates, evidence_version, + attempts, last_attempt_at, next_review_at, decision_id) + values (${release.id}, 'pending', ${reason}, ${jsonb(query)}::jsonb, ${normalized}, ${jsonb(candidates)}::jsonb, ${version}, + ${attempts}, ${now}, ${nextReviewAt}, ${decisionId}) on conflict (source_release_id) do update set - status = 'pending', reason = excluded.reason, query = excluded.query, candidates = excluded.candidates, - evidence_version = excluded.evidence_version, attempts = excluded.attempts, last_attempt_at = excluded.last_attempt_at, - next_review_at = excluded.next_review_at, decision_id = coalesce(excluded.decision_id, identity_candidate.decision_id), + status = 'pending', reason = excluded.reason, query = excluded.query, normalized_title = excluded.normalized_title, + candidates = excluded.candidates, evidence_version = excluded.evidence_version, attempts = excluded.attempts, + last_attempt_at = excluded.last_attempt_at, next_review_at = excluded.next_review_at, + decision_id = case when ${wantsDecision} then excluded.decision_id else null end, resolved_media_entity_id = null, resolved_by = null, resolved_at = null`.execute(trx); + if (decisionId) { + await refreshGroupDecision(ctx, trx, decisionId, { representative: release.id }); + // 旧式单条事项(identity:)并入合并事项;换了组(标题更新)的,原组重算成员 + await closeDecisions(ctx, trx, `identity:${release.id}`, "resolved", { mergedInto: decisionId }, "system"); + if (prev?.decision_id && prev.decision_id !== decisionId) { + await refreshGroupDecision(ctx, trx, prev.decision_id); + } + } await trx .updateTable("source_release") @@ -685,7 +754,13 @@ async function toPool( parser_version: cols.parserVersion, parsed: jsonb(cols.parsed), resolution_status: toReleaseResolutionStatus(contractStatus), - resolution_evidence: jsonb({ reason, detail, evidenceVersion: version, candidateCount: candidates.length }), + resolution_evidence: jsonb({ + reason, + detail, + evidenceVersion: version, + candidateCount: candidates.length, + lookup: scope.lookup, + }), }) .where("id", "=", release.id) .where("resolution_locked", "=", false) diff --git a/apps/server/src/modules/sources-identity/tasks.ts b/apps/server/src/modules/sources-identity/tasks.ts index 08112ff6..facd2b06 100644 --- a/apps/server/src/modules/sources-identity/tasks.ts +++ b/apps/server/src/modules/sources-identity/tasks.ts @@ -36,7 +36,13 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu name: "resolveObservation", version: 1, inputSchema: z - .object({ observationId: id.nullish(), sourceReleaseId: id.nullish(), force: z.boolean().optional() }) + .object({ + observationId: id.nullish(), + sourceReleaseId: id.nullish(), + force: z.boolean().optional(), + // 惰性识别(DEC-1):manual / search 总是查 TMDB;缺省按 force 视为 review / observed + trigger: z.enum(["observed", "review", "manual", "search"]).optional(), + }) .refine((v) => !!v.observationId || !!v.sourceReleaseId, { message: "需要 observationId 或 sourceReleaseId" }), retryPolicy: NETWORK_RETRY, handler: (t) => @@ -45,6 +51,7 @@ export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRu observationId: t.input.observationId ?? undefined, sourceReleaseId: t.input.sourceReleaseId ?? undefined, force: t.input.force, + trigger: t.input.trigger, }), ), }), diff --git a/apps/server/test/e2e/e5-auto-investigate.test.ts b/apps/server/test/e2e/e5-auto-investigate.test.ts index bdec6ed1..664093ad 100644 --- a/apps/server/test/e2e/e5-auto-investigate.test.ts +++ b/apps/server/test/e2e/e5-auto-investigate.test.ts @@ -47,7 +47,8 @@ describe.skipIf(!hasPg)("E5 后台唤醒有界调查", () => { .executeTakeFirstOrThrow(); await e.booted.ctx.engine.enqueue( "resolveObservation", - { sourceReleaseId: rel.id }, + // 惰性识别(DEC-1):不相关的资源不查 TMDB;按搜索结果触发才会查到两个同名候选 + { sourceReleaseId: rel.id, trigger: "search" }, { idempotencyKey: `e5:resolve:${rel.id}`, traceId: rel.id }, ); const decision = await waitFor( diff --git a/apps/server/test/sources-identity/lazy-identity.test.ts b/apps/server/test/sources-identity/lazy-identity.test.ts new file mode 100644 index 00000000..3941c8c4 --- /dev/null +++ b/apps/server/test/sources-identity/lazy-identity.test.ts @@ -0,0 +1,352 @@ +import { sql } from "@umt/db"; +import { normalizeTitle } from "@umt/domain"; +import { createMockTmdb, type MockTmdb, TMDB_FIXTURES, type TmdbFixture } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { down, up } from "../../../../packages/db/migrations/0047_identity_pool.ts"; +import { resolveRelease } from "../../src/modules/sources-identity/service/resolve.ts"; +import { createHarness, type Harness, json, pgAvailable, seedSite } from "./harness.ts"; + +// DEC-1 惰性识别(对照 MP:RSS / 站点资源只在匹配订阅或用户搜索时识别,识别失败不告警)与待识别池治理。 + +const pg = await pgAvailable(); + +const twin = (id: number, en = "Twins", zh = "龙兄鼠弟", date = "1988-12-09"): TmdbFixture => ({ + type: "movie", + id, + original_title: en, + original_language: "en", + origin_country: ["US"], + date, + titles: { "zh-CN": zh, "en-US": en }, + popularity: 10 + id / 1000, + alternative_titles: [], + external_ids: {}, +}); + +describe.skipIf(!pg)("DEC-1 惰性识别 + 待识别池", () => { + let h: Harness; + let tmdb: MockTmdb; + let siteId: string; + let n = 0; + + beforeAll(async () => { + tmdb = await createMockTmdb({ + fixtures: [ + ...TMDB_FIXTURES, + twin(111), + twin(112), + twin(211, "Nightfall", "夜幕", "1995-03-01"), + twin(212, "Nightfall", "夜幕", "1995-03-01"), + ], + }); + h = await createHarness("lazyid", { tmdbUrl: tmdb.url }); + siteId = (await seedSite(h, { baseUrl: "https://hhan.example/" })).site.id; + }); + afterAll(async () => { + await h?.close(); + await tmdb?.close(); + }); + + const release = async (title: string) => + h.db + .insertInto("source_release") + .values({ site_id: siteId, site_release_id: String(900000 + ++n), title }) + .returningAll() + .executeTakeFirstOrThrow(); + const tmdbCalls = () => tmdb.callsTo(/\/3\//).length; + const identityDecisions = () => + h.db + .selectFrom("decision") + .selectAll() + .where("kind", "in", ["identity_ambiguous", "identity_unresolved"]) + .where("status", "in", ["open", "investigating", "awaiting_user"]) + .execute(); + + it("RSS 100 条都不相关 → 0 条待处理、100 条池条目(unmatched)、0 次 TMDB 请求", async () => { + const calls = tmdbCalls(); + const ids: string[] = []; + for (let i = 0; i < 100; i++) { + const r = await release(`真仙觉醒,我透视万物${i} 1080p WEB-DL H265 AAC-HHWEB`); + ids.push(r.id); + const res = await resolveRelease(h.rt, { sourceReleaseId: r.id }); + expect(res).toMatchObject({ outcome: "unmatched", decisionId: null, reason: "no_local_match" }); + } + expect(tmdbCalls()).toBe(calls); + expect(await identityDecisions()).toHaveLength(0); + const pool = await h.db + .selectFrom("identity_candidate") + .selectAll() + .where("source_release_id", "in", ids) + .execute(); + expect(pool).toHaveLength(100); + expect(pool.every((p) => p.status === "pending" && p.reason === "unmatched" && p.decision_id === null)).toBe(true); + // 有界复查:每天一次本地召回 + const next = pool[0]!.next_review_at!.getTime() - Date.now(); + expect(next).toBeGreaterThan(23 * 3600_000); + // 池可检索 + const list = await json(await h.req("/identity-pool?reason=unmatched&q=真仙觉醒&limit=200")); + expect(list.status).toBe(200); + expect(list.body.data.items.length).toBe(100); + expect(list.body.data.items[0].normalizedTitle).toBeTruthy(); + const rel = await h.db + .selectFrom("source_release") + .select(["resolution_status", "resolution_evidence"]) + .where("id", "=", ids[0]!) + .executeTakeFirstOrThrow(); + expect(rel.resolution_status).toBe("unidentifiable"); + expect((rel.resolution_evidence as { lookup: unknown }).lookup).toBeNull(); + }); + + it("复查次数有上限:unmatched 3 次后不再自动复查", async () => { + const r = await release("完全无关的短剧 1080p WEB-DL-HHWEB"); + for (let i = 0; i < 3; i++) await resolveRelease(h.rt, { sourceReleaseId: r.id, force: true }); + const p = await h.db + .selectFrom("identity_candidate") + .selectAll() + .where("source_release_id", "=", r.id) + .executeTakeFirstOrThrow(); + expect(p.attempts).toBe(3); + expect(p.next_review_at).toBeNull(); + }); + + let oppId: string; + it("手动识别(池里点重新识别 / Agent)总是查 TMDB;命中订阅作品别名的新资源查 TMDB 并 resolved", async () => { + const seed = await release("Oppenheimer.2023.2160p.UHD.BluRay.REMUX.HDR.HEVC-FGT"); + const lazy = await resolveRelease(h.rt, { sourceReleaseId: seed.id }); + expect(lazy.outcome).toBe("unmatched"); + const manual = await resolveRelease(h.rt, { sourceReleaseId: seed.id, force: true, trigger: "manual" }); + expect(manual.outcome).toBe("auto"); + oppId = manual.mediaEntityId!; + const pool = await h.db + .selectFrom("identity_candidate") + .select(["status"]) + .where("source_release_id", "=", seed.id) + .executeTakeFirstOrThrow(); + expect(pool.status).toBe("resolved"); + + // 同一作品的新 RSS 条目:标题命中本地别名 → 相关 → 查 TMDB → 自动识别 + const r = await release("奥本海默.Oppenheimer.2023.1080p.WEB-DL.H264-GRP"); + const res = await resolveRelease(h.rt, { sourceReleaseId: r.id }); + expect(res).toMatchObject({ outcome: "auto", mediaEntityId: oppId }); + const rel = await h.db + .selectFrom("source_release") + .select(["resolution_evidence", "resolution_status"]) + .where("id", "=", r.id) + .executeTakeFirstOrThrow(); + expect(rel.resolution_status).toBe("resolved"); + expect((rel.resolution_evidence as { lookup: string }).lookup).toBe("tracked_alias"); + }); + + it("新增相近别名(新证据)立即唤醒 unmatched 池条目;复查命中别名后查 TMDB", async () => { + const r = await release("The.Bear.S03E02.2160p.DSNP.WEB-DL.DDP5.1.H.265-FLUX"); + expect((await resolveRelease(h.rt, { sourceReleaseId: r.id })).outcome).toBe("unmatched"); + const before = await h.db + .selectFrom("identity_candidate") + .select(["next_review_at"]) + .where("source_release_id", "=", r.id) + .executeTakeFirstOrThrow(); + expect(before.next_review_at!.getTime()).toBeGreaterThan(Date.now() + 3600_000); + const created = await json(await h.req("/media-entities", { body: { kind: "series", title: "The Bear" } })); + expect(created.status).toBe(200); + const woke = await h.db + .selectFrom("identity_candidate") + .select(["next_review_at"]) + .where("source_release_id", "=", r.id) + .executeTakeFirstOrThrow(); + expect(woke.next_review_at!.getTime()).toBeLessThanOrEqual(Date.now()); + const res = await resolveRelease(h.rt, { sourceReleaseId: r.id, force: true }); + expect(res.outcome).not.toBe("unmatched"); + expect(["auto", "ambiguous", "proposed"]).toContain(res.outcome); + }); + + it("歧义 → 同来源同标题合并为 1 条事项;选候选对整组生效", async () => { + const titles = [ + "Twins.1988.1080p.BluRay.x264-GRP", + "Twins.1988.720p.BluRay.x264-OTHER", + "Twins.1988.2160p.UHD.BluRay.REMUX-THIRD", + ]; + const ids: string[] = []; + for (const t of titles) { + const r = await release(t); + ids.push(r.id); + const res = await resolveRelease(h.rt, { sourceReleaseId: r.id, trigger: "search" }); + expect(res.outcome).toBe("ambiguous"); + } + const twinsGroup = `identity-group:${siteId}:twins`; + const ds = (await identityDecisions()).filter((d) => d.dedupe_key === twinsGroup); + expect(ds).toHaveLength(1); + const d = ds[0]!; + expect(d.severity).toBe("warning"); + expect(d.title).toMatch(/^来源「.+」的 3 条资源待确认身份:Twins$/); + const ev = d.evidence as { count: number; releases: { sourceReleaseId: string }[] }; + expect(ev.count).toBe(3); + expect(ev.releases.map((x) => x.sourceReleaseId).sort()).toEqual([...ids].sort()); + const pool = await h.db + .selectFrom("identity_candidate") + .select(["decision_id"]) + .where("source_release_id", "in", ids) + .execute(); + expect(pool.every((p) => p.decision_id === d.id)).toBe(true); + + const ok = await json(await h.req(`/decisions/${d.id}/resolve`, { body: { optionKey: "c0" } })); + expect(ok.status).toBe(200); + expect(ok.body.data.status).toBe("resolved"); + const rels = await h.db + .selectFrom("source_release") + .select(["media_entity_id", "resolution_locked"]) + .where("id", "in", ids) + .execute(); + expect(new Set(rels.map((r) => r.media_entity_id)).size).toBe(1); + expect(rels.every((r) => r.media_entity_id && r.resolution_locked)).toBe(true); + expect((await identityDecisions()).filter((x) => x.dedupe_key === twinsGroup)).toHaveLength(0); + }); + + it("组内一条被单独指定 → 只从组里摘除;最后一条处理完才关闭事项", async () => { + const ids: string[] = []; + for (const t of ["Nightfall.1995.1080p.WEB-DL-AAA", "Nightfall.1995.720p.WEB-DL-BBB"]) { + const r = await release(t); + ids.push(r.id); + expect((await resolveRelease(h.rt, { sourceReleaseId: r.id, trigger: "search" })).outcome).toBe("ambiguous"); + } + const open = (await identityDecisions()).filter((x) => x.dedupe_key === `identity-group:${siteId}:nightfall`); + expect(open).toHaveLength(1); + const d = open[0]!; + const pool = await h.db + .selectFrom("identity_candidate") + .select(["id", "source_release_id"]) + .where("decision_id", "=", d.id) + .where("status", "=", "pending") + .execute(); + const first = pool[0]!; + expect( + (await json(await h.req(`/identity-pool/${first.id}/assign`, { body: { mediaEntityId: oppId } }))).status, + ).toBe(200); + expect(pool).toHaveLength(2); + const after = await h.db.selectFrom("decision").selectAll().where("id", "=", d.id).executeTakeFirstOrThrow(); + expect(after.status).toBe("open"); + expect((after.evidence as { count: number }).count).toBe(1); + expect(after.subject_id).toBe(pool[1]!.source_release_id); + expect(after.title).toMatch(/^待确认身份:Nightfall\.1995\./); + // 最后一条忽略 → 事项关闭 + const last = await h.db + .selectFrom("identity_candidate") + .select(["id"]) + .where("source_release_id", "=", pool[1]!.source_release_id) + .executeTakeFirstOrThrow(); + expect((await h.req(`/identity-pool/${last.id}/dismiss`, { body: {} })).status).toBe(200); + const closed = await h.db + .selectFrom("decision") + .select(["status"]) + .where("id", "=", d.id) + .executeTakeFirstOrThrow(); + expect(closed.status).toBe("dismissed"); + }); + + it("收件箱级别筛选:minSeverity=warning 不含仅信息类", async () => { + await h.db + .insertInto("decision") + .values({ kind: "other", severity: "info", title: "巡检报告", dedupe_key: "test:info" }) + .execute(); + const all = await json(await h.req("/decisions/summary")); + const warn = await json(await h.req("/decisions/summary?minSeverity=warning")); + expect(all.body.data.total - warn.body.data.total).toBe(1); + const list = await json(await h.req("/decisions?minSeverity=warning")); + expect(list.body.data.items.every((d: { severity: string }) => d.severity !== "info")).toBe(true); + }); + + it("过期归档:unmatched 超过 poolArchiveDays 没有新证据 → archived", async () => { + const { archiveStalePool } = await import("../../src/modules/sources-identity/service/identity-decisions.ts"); + const r = await release("很久以前的短剧 1080p WEB-DL-HHWEB"); + await resolveRelease(h.rt, { sourceReleaseId: r.id }); + await h.db + .updateTable("identity_candidate") + .set({ last_attempt_at: new Date(Date.now() - 31 * 86_400_000) }) + .where("source_release_id", "=", r.id) + .execute(); + expect(await archiveStalePool(h.rt, { force: true })).toBeGreaterThanOrEqual(1); + const p = await h.db + .selectFrom("identity_candidate") + .select(["status", "next_review_at"]) + .where("source_release_id", "=", r.id) + .executeTakeFirstOrThrow(); + expect(p).toMatchObject({ status: "archived", next_review_at: null }); + }); +}); + +describe.skipIf(!pg)("迁移 0047:存量 identity_unresolved 清理", () => { + let h: Harness; + beforeAll(async () => { + h = await createHarness("dec1mig"); + }); + afterAll(async () => { + await h?.close(); + }); + + it("无关联的 open identity_unresolved → dismissed(moved_to_pool),池条目改 unmatched;相关的保留", async () => { + const siteId = (await seedSite(h, { baseUrl: "https://hhan.example/" })).site.id; + await down(h.db); + const mk = async (title: string, parsedTitle: string) => { + const r = await h.db + .insertInto("source_release") + .values({ site_id: siteId, site_release_id: title, title, resolution_status: "unidentifiable" }) + .returningAll() + .executeTakeFirstOrThrow(); + const d = await h.db + .insertInto("decision") + .values({ + kind: "identity_unresolved", + severity: "info", + title: `待识别:${title}`, + subject_type: "source_release", + subject_id: r.id, + dedupe_key: `identity:${r.id}`, + }) + .returning("id") + .executeTakeFirstOrThrow(); + await sql`insert into identity_candidate (source_release_id, reason, query, evidence_version, attempts, decision_id, + next_review_at) + values (${r.id}, 'no_match', ${JSON.stringify({ title: parsedTitle, normalized: normalizeTitle(parsedTitle) })}::jsonb, + 'v1', 1, ${d.id}, now() + interval '1 hour')`.execute(h.db); + return { releaseId: r.id, decisionId: d.id }; + }; + const noise = await mk("真仙觉醒,我透视万物 1080p WEB-DL H265 AAC-HHWEB", "真仙觉醒,我透视万物"); + const related = await mk("龙兄鼠弟.1988.1080p.WEB-DL", "龙兄鼠弟"); + const e = await h.db + .insertInto("media_entity") + .values({ kind: "movie", title: "龙兄鼠弟" }) + .returning("id") + .executeTakeFirstOrThrow(); + await sql`insert into title_alias (media_entity_id, alias, normalized, kind, source) + values (${e.id}, '龙兄鼠弟', ${normalizeTitle("龙兄鼠弟")}, 'translated', 'user')`.execute(h.db); + + await up(h.db); + + const d1 = await h.db + .selectFrom("decision") + .selectAll() + .where("id", "=", noise.decisionId) + .executeTakeFirstOrThrow(); + expect(d1.status).toBe("dismissed"); + expect(d1.resolution).toMatchObject({ resolution: "moved_to_pool" }); + const p1 = await h.db + .selectFrom("identity_candidate") + .selectAll() + .where("source_release_id", "=", noise.releaseId) + .executeTakeFirstOrThrow(); + expect(p1).toMatchObject({ reason: "unmatched", decision_id: null, next_review_at: null, status: "pending" }); + expect(p1.normalized_title).toBe(normalizeTitle("真仙觉醒,我透视万物")); + + const d2 = await h.db + .selectFrom("decision") + .select(["status"]) + .where("id", "=", related.decisionId) + .executeTakeFirstOrThrow(); + expect(d2.status).toBe("open"); + const p2 = await h.db + .selectFrom("identity_candidate") + .select(["reason", "decision_id"]) + .where("source_release_id", "=", related.releaseId) + .executeTakeFirstOrThrow(); + expect(p2).toMatchObject({ reason: "no_match", decision_id: related.decisionId }); + }); +}); diff --git a/apps/server/test/sources-identity/resolve.test.ts b/apps/server/test/sources-identity/resolve.test.ts index ea6e4109..1ccbc8fa 100644 --- a/apps/server/test/sources-identity/resolve.test.ts +++ b/apps/server/test/sources-identity/resolve.test.ts @@ -31,6 +31,11 @@ describe.skipIf(!up)("resolveObservation / 身份与元数据", () => { tmdb = await createMockTmdb({ fixtures: [...TMDB_FIXTURES, twin(111), twin(112)] }); h = await createHarness("resolve", { tmdbUrl: tmdb.url }); siteId = (await seedSite(h, { baseUrl: "https://audiences.example/" })).site.id; + // 本组覆盖 TMDB 召回本身:打开"每条新资源都查 TMDB"(旧行为);惰性识别见 lazy-identity.test.ts + await h.db + .insertInto("settings") + .values({ key: "identity", value: JSON.stringify({ eagerTmdbLookup: true }) } as never) + .execute(); }); afterAll(async () => { await h?.close(); @@ -120,7 +125,7 @@ describe.skipIf(!up)("resolveObservation / 身份与元数据", () => { }); let twinsRelease: string; - it("ambiguous:进待识别池 + 待处理事项,不建正式作品", async () => { + it("ambiguous:进待识别池 + 待处理事项(合并事项,warning),不建正式作品", async () => { const before = await h.db.selectFrom("media_entity").select("id").execute(); const r = await release("Twins.1988.1080p.BluRay.x264-GRP"); twinsRelease = r.id; @@ -136,7 +141,8 @@ describe.skipIf(!up)("resolveObservation / 身份与元数据", () => { expect((pool.candidates as unknown[]).length).toBe(2); expect(pool.next_review_at!.getTime()).toBeGreaterThan(Date.now()); const d = await h.db.selectFrom("decision").selectAll().where("id", "=", res.decisionId!).executeTakeFirstOrThrow(); - expect(d).toMatchObject({ kind: "identity_ambiguous", status: "open", subject_id: r.id }); + expect(d).toMatchObject({ kind: "identity_ambiguous", status: "open", subject_id: r.id, severity: "warning" }); + expect(d.dedupe_key).toBe(`identity-group:${siteId}:twins`); expect((d.options as { key: string }[]).map((o) => o.key)).toEqual([ "c0", "c1", @@ -164,7 +170,11 @@ describe.skipIf(!up)("resolveObservation / 身份与元数据", () => { expect(pool2.attempts).toBe(2); expect(pool2.next_review_at!.getTime()).toBeGreaterThan(pool.next_review_at!.getTime()); expect( - await h.db.selectFrom("decision").select("id").where("dedupe_key", "=", `identity:${r.id}`).execute(), + await h.db + .selectFrom("decision") + .select("id") + .where("dedupe_key", "=", `identity-group:${siteId}:twins`) + .execute(), ).toHaveLength(1); }); @@ -214,38 +224,32 @@ describe.skipIf(!up)("resolveObservation / 身份与元数据", () => { expect(after.media_entity_id).toBe(rel.media_entity_id); }); - it("no_match 进池并开未识别事项;手动指定 TMDB id / 选本地作品即人工指定加锁;TMDB 失败记为 lookup_failed", async () => { + it("no_match(与本地作品无关)只进池不开事项;池条目上手动指定 TMDB id / 选本地作品即人工指定加锁;TMDB 失败记为 lookup_failed", async () => { const r1 = await release("Zzqxv.Nonexistent.2019.1080p.WEB-DL-NOPE"); const a = await resolveRelease(h.rt, { sourceReleaseId: r1.id }); expect(a.outcome).toBe("no_match"); - expect(a.decisionId).not.toBeNull(); + expect(a.decisionId).toBeNull(); const rel = await h.db .selectFrom("source_release") .select(["resolution_status"]) .where("id", "=", r1.id) .executeTakeFirstOrThrow(); expect(rel.resolution_status).toBe("unidentifiable"); - const d = await h.db.selectFrom("decision").selectAll().where("id", "=", a.decisionId!).executeTakeFirstOrThrow(); - expect(d).toMatchObject({ kind: "identity_unresolved", status: "open", subject_id: r1.id }); - expect((d.options as { key: string }[]).map((o) => o.key)).toEqual(["manual_tmdb", "local_entity", "dismiss"]); const pool1 = await h.db .selectFrom("identity_candidate") - .select(["decision_id"]) + .select(["id", "decision_id", "reason", "normalized_title"]) .where("source_release_id", "=", r1.id) .executeTakeFirstOrThrow(); - expect(pool1.decision_id).toBe(d.id); - // 缺少输入 → 400;给 TMDB id → 人工指定加锁,事项关闭 - expect((await json(await h.req(`/decisions/${d.id}/resolve`, { body: { optionKey: "manual_tmdb" } }))).status).toBe( - 400, - ); + expect(pool1).toMatchObject({ decision_id: null, reason: "no_match", normalized_title: "zzqxvnonexistent" }); + // 缺少输入 → 422(校验失败);给 TMDB id → 人工指定加锁 + expect((await json(await h.req(`/identity-pool/${pool1.id}/assign`, { body: {} }))).status).toBe(422); const fx = TMDB_FIXTURES.find((f) => f.type === "movie")!; const ok = await json( - await h.req(`/decisions/${d.id}/resolve`, { - body: { optionKey: "manual_tmdb", identity: { tmdb: { type: "movie", id: String(fx.id) } } }, + await h.req(`/identity-pool/${pool1.id}/assign`, { + body: { external: { provider: "tmdb", entityType: "movie", externalId: String(fx.id) } }, }), ); expect(ok.status).toBe(200); - expect(ok.body.data.status).toBe("resolved"); const rel1 = await h.db.selectFrom("source_release").selectAll().where("id", "=", r1.id).executeTakeFirstOrThrow(); expect(rel1).toMatchObject({ resolution_status: "resolved", resolution_locked: true }); const ident = await h.db @@ -260,11 +264,12 @@ describe.skipIf(!up)("resolveObservation / 身份与元数据", () => { const r3 = await release("Xyzzy.Plugh.2018.1080p.WEB-DL-NOPE"); const c = await resolveRelease(h.rt, { sourceReleaseId: r3.id }); expect(c.outcome).toBe("no_match"); - const local = await json( - await h.req(`/decisions/${c.decisionId}/resolve`, { - body: { optionKey: "local_entity", identity: { mediaEntityId: oppId } }, - }), - ); + const pool3 = await h.db + .selectFrom("identity_candidate") + .select(["id"]) + .where("source_release_id", "=", r3.id) + .executeTakeFirstOrThrow(); + const local = await json(await h.req(`/identity-pool/${pool3.id}/assign`, { body: { mediaEntityId: oppId } })); expect(local.status).toBe(200); const rel3 = await h.db.selectFrom("source_release").selectAll().where("id", "=", r3.id).executeTakeFirstOrThrow(); expect(rel3).toMatchObject({ media_entity_id: oppId, resolution_locked: true }); diff --git a/apps/web/app/api/decisions.ts b/apps/web/app/api/decisions.ts index 2897e227..25b0157b 100644 --- a/apps/web/app/api/decisions.ts +++ b/apps/web/app/api/decisions.ts @@ -15,10 +15,13 @@ import { getParsed, parseOrThrow } from "./library.ts"; export const decisionKeys = { all: ["decisions"] as const, list: (f: DecisionFilters) => ["decisions", "list", f] as const, - summary: (status?: DecisionStatus) => ["decisions", "summary", status ?? "open"] as const, + summary: (status?: DecisionStatus, minSeverity?: MinSeverity) => + ["decisions", "summary", status ?? "open", minSeverity ?? "info"] as const, }; -export type DecisionFilters = { status?: DecisionStatus; kind?: DecisionKind }; +/** 只看不低于此级别的事项;收件箱缺省 warning(隐藏"仅信息"类) */ +export type MinSeverity = "info" | "warning" | "critical"; +export type DecisionFilters = { status?: DecisionStatus; kind?: DecisionKind; minSeverity?: MinSeverity }; /** /decisions/summary(合同 DecisionSummary) */ export { DecisionSummary }; @@ -37,10 +40,10 @@ export function useDecisions(filters: DecisionFilters, enabled = true) { }); } -export function useDecisionSummary(status?: DecisionStatus) { +export function useDecisionSummary(status?: DecisionStatus, minSeverity?: MinSeverity) { return useQuery({ - queryKey: decisionKeys.summary(status), - queryFn: ({ signal }) => getParsed(DecisionSummary, "/decisions/summary", { status }, signal), + queryKey: decisionKeys.summary(status, minSeverity), + queryFn: ({ signal }) => getParsed(DecisionSummary, "/decisions/summary", { status, minSeverity }, signal), }); } diff --git a/apps/web/app/components/console/inbox-bell.tsx b/apps/web/app/components/console/inbox-bell.tsx index 57a3cce5..f03a951c 100644 --- a/apps/web/app/components/console/inbox-bell.tsx +++ b/apps/web/app/components/console/inbox-bell.tsx @@ -12,6 +12,8 @@ import { useSse } from "~/lib/sse.ts"; import { cn } from "~/lib/utils.ts"; // 顶栏收件箱(IA-3):待处理不进侧栏,铃铛 + 未决角标;下拉按问题类型列最近事项,点开进待处理页。 +// 角标只算需要你处理的事项(警告及以上):"仅信息"类(巡检报告、旧版本待处置等)不计入,在待处理页可展开查看。 +// 身份类事项已按来源 + 标题合并,一条事项可能代表多条资源(标题里写明 N 条)。 const RECENT = 8; const DOT = { info: "bg-muted-foreground/60", warning: "bg-warning", critical: "bg-destructive" } as const; @@ -27,7 +29,7 @@ export function useLiveDecisions() { void qc.invalidateQueries({ queryKey: decisionKeys.all }); }, 800); }); - return useDecisionSummary(); + return useDecisionSummary(undefined, "warning"); } export const inboxPath = (kind?: string) => (kind ? `/console/inbox?kind=${kind}` : "/console/inbox"); @@ -35,7 +37,7 @@ export const inboxPath = (kind?: string) => (kind ? `/console/inbox?kind=${kind} export function InboxBell() { const summary = useLiveDecisions(); const [open, setOpen] = useState(false); - const list = useDecisions({}, open); + const list = useDecisions({ minSeverity: "warning" }, open); const navigate = useNavigate(); const total = summary.data?.total ?? 0; diff --git a/apps/web/app/components/library/identity-pool-sheet.tsx b/apps/web/app/components/library/identity-pool-sheet.tsx index 26e6ef0a..c9a96deb 100644 --- a/apps/web/app/components/library/identity-pool-sheet.tsx +++ b/apps/web/app/components/library/identity-pool-sheet.tsx @@ -1,21 +1,27 @@ import type { IdentityPoolItem, PoolCandidate } from "@umt/contracts"; -import { ArrowRight, ExternalLink, LoaderCircle, RotateCw } from "lucide-react"; +import { ArrowRight, BellOff, ExternalLink, LoaderCircle, RotateCw } from "lucide-react"; +import { useId, useState } from "react"; import { Link } from "react-router"; import { toast } from "sonner"; -import { useRetryPool } from "~/api/identity.ts"; +import { useAssignPool, useDismissPool, useRetryPool } from "~/api/identity.ts"; import { DetailSection, DetailSheet } from "~/components/console/detail-sheet.tsx"; +import { FormField } from "~/components/console/form-field.tsx"; import { JsonView } from "~/components/console/json-view.tsx"; import { KvList } from "~/components/console/kv-list.tsx"; import { consolePath } from "~/components/console/nav.ts"; +import { EntityPicker, type PickedEntity } from "~/components/library/identity-entity-picker.tsx"; import { Button } from "~/components/ui/button.tsx"; +import { Input } from "~/components/ui/input.tsx"; +import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "~/components/ui/select.tsx"; import { errorMessage } from "~/lib/api.ts"; import { formatDateTime } from "~/lib/utils.ts"; import { EVIDENCE_LEVEL, POOL_REASON, POOL_STATUS } from "./labels.ts"; import { Chip, Tone } from "./parts.tsx"; -// 待识别资源详情(只读):候选证据 + 重新识别。选候选、忽略在待处理里做(判定只在待处理与作品详情)。 +// 待识别资源详情:候选证据 + 重新识别 + 手动指定。进了待处理的(与订阅相关的歧义 / 无结果)在待处理里选候选; +// 没进待处理的(unmatched:未命中订阅,惰性识别没查 TMDB)在这里手动指定 TMDB ID / 本地作品,或忽略。 -/** 多个候选 / 待确认的资源开"待确认身份"事项;没有候选的(无结果、复查耗尽)开"待识别"事项 */ +/** 与订阅 / 本地作品相关的歧义开"待确认身份"事项,相关却无结果开"待识别"事项(同来源同标题合并);unmatched 不进待处理 */ export function inInbox(item: Pick): boolean { return item.status === "pending" && (item.decisionId !== null || hasCandidates(item)); } @@ -85,6 +91,97 @@ function CandidateRow({ c, index }: { c: PoolCandidate; index: number }) { ); } +/** 手动指定(复用 /identity-pool/{id}/assign,加人工锁)与忽略 */ +function ManualAssign({ item }: { item: IdentityPoolItem }) { + const uid = useId(); + const assign = useAssignPool(); + const dismiss = useDismissPool(); + const [tmdbType, setTmdbType] = useState<"movie" | "tv">(item.query.season != null ? "tv" : "movie"); + const [tmdbId, setTmdbId] = useState(""); + const [entity, setEntity] = useState(null); + const tmdbOk = /^\d{1,10}$/.test(tmdbId.trim()); + const run = async (body: Parameters[0], ok: string) => { + try { + await assign.mutateAsync(body); + toast.success(ok); + } catch (err) { + toast.error(errorMessage(err, "指定失败")); + } + }; + return ( + +
+ + + + + setTmdbId(e.target.value)} + aria-invalid={tmdbId.trim() !== "" && !tmdbOk} + /> + + +
+
+ + + + +
+ +
+ ); +} + export function PoolSheet({ item, onClose }: { item: IdentityPoolItem | null; onClose: () => void }) { const retry = useRetryPool(); if (!item) return null; @@ -144,6 +241,8 @@ export function PoolSheet({ item, onClose }: { item: IdentityPoolItem | null; on ]} /> + {item.status !== "resolved" && !queued ? : null} + {item.candidates.length ? (
    @@ -152,7 +251,11 @@ export function PoolSheet({ item, onClose }: { item: IdentityPoolItem | null; on ))}
) : ( -

没有候选。有新证据时自动重试。

+

+ {item.reason === "unmatched" + ? "没有命中订阅或本地作品,未查询 TMDB。点「重新识别」会查一次 TMDB;新增相近别名时自动重试。" + : "没有候选。有新证据时自动重试。"} +

)}
diff --git a/apps/web/app/components/library/identity-views.tsx b/apps/web/app/components/library/identity-views.tsx index 70b67f81..c0afcc9d 100644 --- a/apps/web/app/components/library/identity-views.tsx +++ b/apps/web/app/components/library/identity-views.tsx @@ -149,7 +149,7 @@ export function PoolTab() { empty={{ icon: SearchX, title: filtered ? "没有符合条件的待识别资源" : "没有待识别资源", - description: "证据不足或冲突的资源会出现在这里,有新证据时自动重试。", + description: "未命中订阅、证据不足或冲突的资源会出现在这里;新增相近别名时自动重试,也可手动指定。", }} hasMore={pool.hasNextPage} onLoadMore={() => pool.fetchNextPage()} diff --git a/apps/web/app/components/library/labels.ts b/apps/web/app/components/library/labels.ts index 9a4c1340..7e50d528 100644 --- a/apps/web/app/components/library/labels.ts +++ b/apps/web/app/components/library/labels.ts @@ -164,6 +164,7 @@ export const POOL_REASON: Record = { no_match: { label: "没有匹配", tone: "neutral" }, proposed: { label: "待确认", tone: "warning" }, lookup_failed: { label: "查询失败", tone: "destructive" }, + unmatched: { label: "未命中订阅", tone: "neutral" }, }; export const POOL_STATUS: Record = { diff --git a/apps/web/app/components/settings/metadata.tsx b/apps/web/app/components/settings/metadata.tsx index 091a8e5c..17880b83 100644 --- a/apps/web/app/components/settings/metadata.tsx +++ b/apps/web/app/components/settings/metadata.tsx @@ -180,6 +180,39 @@ function ImageCacheStatus() { ); } +// 身份识别(DEC-1 惰性识别,对照 MP:RSS / 列表资源只在匹配订阅或用户搜索时识别) +const IDENTITY_CARD: CardSpec = { + anchor: "identity", + title: "资源识别", + description: + "缺省只在资源命中订阅或本地作品别名、新作品收录规则前置条件通过、你手动识别或搜索时查 TMDB;其余进待识别池,不进待处理。", + fields: [ + { + name: "identity.eagerTmdbLookup", + label: "每条新资源都查 TMDB", + kind: "switch", + hint: "旧行为。每条资源约 1–2 次搜索 + 命中时 1 次详情(识别缓存命中为 0);每小时 100 条的 RSS 约每天 2400–7200 次请求,站点独有的短剧 / 综艺多半查不到", + wide: true, + }, + { + name: "identity.poolArchiveDays", + label: "待识别池归档(天)", + kind: "number", + hint: "未命中订阅的条目超过这么多天没有新证据就归档,仍可筛出并手动指定", + }, + ], + fromSettings: (s) => ({ + "identity.eagerTmdbLookup": s.identity.eagerTmdbLookup, + "identity.poolArchiveDays": n(s.identity.poolArchiveDays), + }), + toPatch: (v) => ({ + identity: { + eagerTmdbLookup: v["identity.eagerTmdbLookup"] === true, + poolArchiveDays: Number(v["identity.poolArchiveDays"]), + }, + }), +}; + export function MetadataSection() { return ( @@ -187,6 +220,7 @@ export function MetadataSection() { <> + )} diff --git a/apps/web/app/routes/console/inbox.tsx b/apps/web/app/routes/console/inbox.tsx index 2611eeee..a3195355 100644 --- a/apps/web/app/routes/console/inbox.tsx +++ b/apps/web/app/routes/console/inbox.tsx @@ -1,6 +1,6 @@ import { useQueryClient } from "@tanstack/react-query"; import type { Decision, DecisionKind, DecisionOption, DecisionStatus } from "@umt/contracts"; -import { BellOff, Bot, CheckCheck, ChevronRight, Inbox, Radio, RotateCw } from "lucide-react"; +import { BellOff, Bot, CheckCheck, ChevronDown, ChevronRight, Inbox, Radio, RotateCw } from "lucide-react"; import { type ReactNode, useId, useMemo, useRef, useState } from "react"; import { Link, useSearchParams } from "react-router"; import { toast } from "sonner"; @@ -26,8 +26,10 @@ import { actorText } from "~/lib/labels.ts"; import { useSse, useSseStatus } from "~/lib/sse.ts"; import { formatDateTime } from "~/lib/utils.ts"; -// 待处理(REQ §13.1):按作品与问题类型合并;每项给证据、建议、影响范围与预览; +// 待处理(REQ §13.1):按问题类型 + 作品 / 来源合并并可折叠;每项给证据、建议、影响范围与预览; // 处理选项来自后端 decision.options;SSE 收到 decision.* 事件即刷新。 +// 缺省只看需要你处理的事项(警告及以上),"仅信息"类在"级别"筛选里展开。身份类事项后端已按来源 + 标题合并, +// 卡片里列出组内资源,选候选 / 指定 / 忽略对整组生效。 const OPEN: DecisionStatus[] = ["open", "investigating", "awaiting_user"]; const STATUS_FILTER = [ @@ -36,9 +38,55 @@ const STATUS_FILTER = [ { value: "expired", label: "已过期" }, ]; const SEV_RANK = { info: 0, warning: 1, critical: 2 } as const; +const SEV_FILTER = [{ value: "all", label: "含仅信息" }]; +/** 缺省展开的组数;其余折叠 */ +const OPEN_GROUPS = 3; /** 证据里约定俗成的字段(Agent/后端写入);其余原样放进"原始证据" */ const str = (v: unknown) => (typeof v === "string" && v.trim() ? v : null); +/** 身份类合并事项的来源分组(后端 evidence.group) */ +function sourceGroupOf(d: Decision): { siteId: string; siteName: string | null } | null { + const g = d.evidence.group as { siteId?: unknown; siteName?: unknown } | undefined; + return g && typeof g.siteId === "string" ? { siteId: g.siteId, siteName: str(g.siteName) } : null; +} + +/** 合并事项的组内资源 */ +function GroupReleases({ d }: { d: Decision }) { + const list = Array.isArray(d.evidence.releases) + ? (d.evidence.releases as { sourceReleaseId?: string; title?: string }[]) + : []; + const count = typeof d.evidence.count === "number" ? d.evidence.count : list.length; + const [open, setOpen] = useState(false); + if (count <= 1) return null; + return ( +
+ + {open ? ( +
    + {list.map((r) => ( +
  • + {r.title} +
  • + ))} + {count > list.length ? ( +
  • + 另有 {count - list.length} 条,可在媒体概览 → 待识别按标题检索 +
  • + ) : null} +
+ ) : null} +
+ ); +} + function entityOf(d: Decision): string | null { const m = d.evidence.mediaEntityId; if (typeof m === "string" && /^[0-9a-f-]{36}$/.test(m)) return m; @@ -240,6 +288,7 @@ function DecisionCard({ d }: { d: Decision }) {

{d.title}

{d.summary ?

{d.summary}

: null} +
@@ -272,12 +321,20 @@ function DecisionCard({ d }: { d: Decision }) { ); } -type Group = { key: string; entityId: string | null; kind: DecisionKind; items: Decision[] }; +type Group = { + key: string; + entityId: string | null; + source: { siteId: string; siteName: string | null } | null; + kind: DecisionKind; + items: Decision[]; +}; export default function InboxPage() { const [sp, setSp] = useSearchParams(); const status = (sp.get("status") ?? "") as DecisionStatus | ""; const kind = (sp.get("kind") ?? "") as DecisionKind | ""; + const sev = sp.get("sev") ?? ""; + const minSeverity = sev === "all" ? undefined : "warning"; const set = (k: string, v: string) => setSp( (prev) => { @@ -289,8 +346,8 @@ export default function InboxPage() { { replace: true }, ); - const summary = useDecisionSummary(status || undefined); - const list = useDecisions({ status: status || undefined, kind: kind || undefined }); + const summary = useDecisionSummary(status || undefined, minSeverity); + const list = useDecisions({ status: status || undefined, kind: kind || undefined, minSeverity }); const rows = list.data?.pages.flatMap((p) => p.items); // SSE:decision.created / decision.updated → 刷新(合并 500ms 内的多条事件) @@ -324,13 +381,14 @@ export default function InboxPage() { return [...m.entries()].sort((a, b) => SEV_RANK[b[1].sev] - SEV_RANK[a[1].sev] || b[1].count - a[1].count); }, [summary.data]); - // 按作品 + 问题类型合并;组内按严重度、时间 + // 按问题类型 + 作品 / 来源合并;组内按严重度、时间 const groups = useMemo(() => { const m = new Map(); for (const d of rows ?? []) { const entityId = entityOf(d); - const key = `${entityId ?? `subject:${d.subjectType ?? "none"}`}|${d.kind}`; - const g = m.get(key) ?? { key, entityId, kind: d.kind, items: [] }; + const source = entityId ? null : sourceGroupOf(d); + const key = `${entityId ?? (source ? `site:${source.siteId}` : `subject:${d.subjectType ?? "none"}`)}|${d.kind}`; + const g = m.get(key) ?? { key, entityId, source, kind: d.kind, items: [] }; g.items.push(d); m.set(key, g); } @@ -339,6 +397,9 @@ export default function InboxPage() { }, [rows]); const isOpenView = !status; + // 折叠状态:缺省前 OPEN_GROUPS 组展开;用户点过的组按用户的来 + const [toggled, setToggled] = useState>({}); + const isExpanded = (key: string, index: number) => toggled[key] ?? index < OPEN_GROUPS; return ( <> )} - setSp(new URLSearchParams(), { replace: true }) : undefined}> + setSp(new URLSearchParams(), { replace: true }) : undefined}> set("kind", v)} options={opts(DECISION_KIND)} /> + set("sev", v)} options={SEV_FILTER} allLabel="需处理" /> {list.isPending ? ( @@ -413,15 +475,27 @@ export default function InboxPage() { /> ) : (
- {groups.map((g) => { + {groups.map((g, index) => { const title = g.entityId ? titles.get(g.entityId) : null; + const expanded = isExpanded(g.key, index); const maxSev = g.items.reduce( (a, d) => (SEV_RANK[d.severity] > SEV_RANK[a] ? d.severity : a), "info", ); return (
-
+
+ {DECISION_KIND[g.kind]} {g.entityId ? ( {title ?? "关联作品"} + ) : g.source ? ( + + 来源「{g.source.siteName ?? "未知站点"}」 + ) : ( {g.kind === "identity_ambiguous" || g.kind === "identity_unresolved" ? "尚未归属作品" : "系统级"} @@ -437,9 +515,7 @@ export default function InboxPage() { )} {g.items.length} 项
- {g.items.map((d) => ( - - ))} + {expanded ? g.items.map((d) => ) : null}
); })} diff --git a/docs/runbook/identity.md b/docs/runbook/identity.md new file mode 100644 index 00000000..ebb47f4f --- /dev/null +++ b/docs/runbook/identity.md @@ -0,0 +1,91 @@ +# 资源识别与待识别池运行手册(DEC-1) + +新观察到的资源(RSS、站点列表快照、搜索结果)先进待识别池,只在有理由时才查 TMDB。只有和订阅或本地作品相关、 +又需要你决定的资源才进待处理。对照 MoviePilot:RSS / 站点资源只在匹配订阅或用户搜索时才识别,识别失败只写日志、不告警 +(`app/chain/subscribe/match.py` `_prepare_match_torrents`:未识别的种子最多重试 3 次,之后跳过)。 + +需求依据:REQ §4.2–4.3(观察不等于事实)、§6.3(新作品收录 / 追更补缺 / 升级都由规则触发)、§7.5(待识别池可检索、 +不污染正式库、有界复查、不留死数据)、§13(待处理按作品与问题类型合并)。 + +## 识别流程(resolveObservation) + +1. 解析标题(release-parser),本地召回:显式外部 ID、`title_alias` 精确 / pg_trgm ≥ 0.45(本地作品 = 订阅、追踪、已登记的作品)。 +2. 判断是否查 TMDB(`lookup`,记在 `source_release.resolution_evidence.lookup` 和池条目 `query.lookup`): + + | lookup | 条件 | + |---|---| + | `manual` | 任务输入 `trigger: "manual"`:池里点「重新识别」、Agent / 能力 `identity.resolve` | + | `search` | 任务输入 `trigger: "search"`:搜索结果 | + | `tracked_alias` | 本地召回有候选(标题命中订阅 / 本地作品的别名) | + | `previously_looked_up` | 周期复查,且这条池条目以前查过 TMDB(旧数据、手动、规则触发) | + | `eager` | 设置 `identity.eagerTmdbLookup = true` | + | `new_title_rule` | 有已发布、开启 `newTitle` 且有有效持续授权的获取策略,资源类型在规则的 `mediaKinds` 内,标题年份(若有)不早于窗口能覆盖的年份 | + + 都不满足:只进池,`reason = unmatched`,不查 TMDB,不开待处理。 +3. 查了 TMDB 的,走原来的 `domain.resolveIdentity`:auto 建档 / 关联并发 `release.resolved`;否则进池。 + +任务输入 `trigger` 缺省:`force` 为 true 视为 `review`(池复查),否则 `observed`。 +搜索模块想让搜索结果直接识别,入队 `resolveObservation` 时传 `trigger: "search"`(目前搜索结果经 `release.observed` 进来, +按 observed 处理;命中订阅别名的照样会查)。 + +## 待处理只收相关的 + +| 结果 | 开事项? | +|---|---| +| `unmatched` | 否 | +| `ambiguous` / `proposed`(查过 TMDB,有候选) | 是,`identity_ambiguous`,级别 warning | +| `no_match` / 复查已停的 `lookup_failed` | 只有本地召回有候选(与本地作品相关)时开 `identity_unresolved`,级别 warning | + +合并:dedupeKey `identity-group::<归一化标题>`,同来源同标题的资源并成一条。成员 = `identity_candidate.decision_id` +指向该事项且仍 `pending` 的条目。标题 1 条时写资源原标题,多条时写「来源「X」的 N 条资源待识别 / 待确认身份:<标题>」, +`evidence.releases` 列前 50 条,`evidence.count` 为总数。 + +- 处理(选候选 / 手动 TMDB ID / 选本地作品):先处理代表资源(subject),再把确定的作品套到组内其它资源(都加人工锁)。 +- 忽略:组内每条池条目都 `dismissed`。 +- 组内某条被单独识别 / 指定 / 忽略:从组里摘除,代表资源和候选选项换成剩下的最近一条;组空了才关事项。 +- 旧式单条事项(`identity:`)在该资源下次复查时并入合并事项并关闭。 + +收件箱:顶栏角标和待处理页缺省只算 warning 及以上(`minSeverity=warning`),"仅信息"类在待处理页「级别 → 含仅信息」里看。 +待处理页按问题类型 + 作品 / 来源分组,前 3 组展开,其余折叠;合并事项卡片可展开组内资源。 + +## 池的治理 + +- 复查有上限:`unmatched` 每天一次本地召回,最多 3 次;查过 TMDB 的按原退避(无结果 1h 起、查询失败 15min 起,翻倍到 7 天,最多 8 次)。 +- 新证据即重试:`title_alias` 插入触发器 `title_alias_wake_identity_pool`,把归一化标题相同或相似度 ≥ 0.45 的 + `unmatched` / `no_match` 池条目的 `next_review_at` 设为现在。建档、手动加别名、合并作品都会触发。 +- 过期归档:`identity.poolArchiveDays`(缺省 30)天没有新尝试的 `unmatched` 条目改为 `archived`,在池复查定时任务里做, + 每进程每小时最多一次。归档不删资源,仍可按状态筛出、手动指定。 +- 检索:媒体概览 → 待识别,按原始标题 / 解析标题检索,原因筛「未命中订阅」即 unmatched。条目详情里可「重新识别」 + (手动触发,会查 TMDB)、手动指定 TMDB ID / 本地作品(复用 `/identity-pool/{id}/assign`,加人工锁)或忽略。 + +## 设置 + +设置 → 元数据 → 资源识别: + +- `identity.eagerTmdbLookup`(缺省关):打开即旧行为,每条新资源都查 TMDB。每条约 1–2 次搜索 + 命中时 1 次详情 + (识别缓存命中为 0);每小时 100 条的 RSS 约每天 2400–7200 次请求,站点独有的短剧 / 综艺多半查不到。 + 打开后无结果的资源仍只进池,不开事项。 +- `identity.poolArchiveDays`(缺省 30)。 + +## 存量清理(迁移 0047_identity_pool) + +未关闭的 `identity_unresolved` 事项中,资源未关联作品、池里没有候选、归一化标题也不命中任何本地别名的: +事项改 `dismissed`,`resolution = {"resolution":"moved_to_pool","by":"migration:0047_identity_pool"}`; +池条目改 `reason = unmatched`、清 `decision_id`、停自动复查。相关的保留原样,下次复查时并入合并事项。 + +需要重建镜像 / 重启 api + worker 才生效(迁移随启动执行)。 + +## 排查 + +```sql +-- 池分布 +select reason, status, count(*) from identity_candidate group by 1, 2 order by 3 desc; +-- 某条资源为什么没查 TMDB +select resolution_status, resolution_evidence->>'lookup' as lookup, resolution_evidence->>'detail' as detail + from source_release where id = ''; +-- 某个合并事项的成员 +select p.source_release_id, r.title, p.reason from identity_candidate p join source_release r on r.id = p.source_release_id + where p.decision_id = '' and p.status = 'pending'; +-- 迁移搬走的事项 +select count(*) from decision where resolution->>'resolution' = 'moved_to_pool'; +``` diff --git a/packages/contracts/src/identity.ts b/packages/contracts/src/identity.ts index 6dd22815..8d8d7d1d 100644 --- a/packages/contracts/src/identity.ts +++ b/packages/contracts/src/identity.ts @@ -126,8 +126,11 @@ export type IdentityRevisionCreate = z.infer; export const IdentityCandidateStatus = z.enum(["pending", "resolved", "dismissed", "archived"]); export type IdentityCandidateStatus = z.infer; -/** proposed = 证据足够但自动化关闭/只读,待确认 */ -export const IdentityCandidateReason = z.enum(["ambiguous", "no_match", "proposed", "lookup_failed"]); +/** + * proposed = 证据足够但自动化关闭/只读,待确认; + * unmatched = 惰性识别:本地召回无结果且没有触发 TMDB 查询(不相关的 RSS / 列表资源),只进池、不开待处理 + */ +export const IdentityCandidateReason = z.enum(["ambiguous", "no_match", "proposed", "lookup_failed", "unmatched"]); export type IdentityCandidateReason = z.infer; /** 待识别池里的一个候选(只存 JSON,不建正式作品) */ @@ -151,6 +154,8 @@ export const IdentityCandidate = EntityBase.extend({ reason: IdentityCandidateReason, /** 解析出的检索条件(标题/年份/季集) */ query: JsonObject, + /** 归一化标题(与 title_alias.normalized 同口径):新增别名时按它唤醒复查,待处理按"来源 + 它"合并 */ + normalizedTitle: z.string().nullable(), candidates: z.array(PoolCandidate), evidenceVersion: z.string(), attempts: z.number().int().nonnegative(), @@ -171,6 +176,7 @@ export const ResolutionOutcome = z.enum([ "lookup_failed", "manual", "skipped_locked", + "unmatched", ]); export type ResolutionOutcome = z.infer; diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index c749b393..d3ba7e26 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -357,6 +357,19 @@ export const BudgetSettings = z.object({ }); export type BudgetSettings = z.infer; +/** + * 身份识别(DEC-1 惰性识别,对照 MP:RSS / 列表资源只在匹配订阅或用户搜索时识别)。 + * eagerTmdbLookup = true 恢复旧行为:每条新观察到的资源都查 TMDB。每条资源约 1–2 次搜索 + 命中时 1 次详情 + * (识别缓存命中则 0 次);一个每小时 100 条的 RSS 约为每天 2400–7200 次 TMDB 请求,多数是站点独有的短剧 / 综艺,查不到。 + * 关闭(缺省)时只在命中订阅 / 本地作品别名、已发布的新作品收录规则前置条件通过、用户手动识别或搜索时查询。 + */ +export const IdentitySettings = z.object({ + eagerTmdbLookup: z.boolean(), + /** 待识别池里 unmatched 条目多少天没有新证据就归档(仍可按状态筛出、可手动指定) */ + poolArchiveDays: z.number().int().min(1).max(3650), +}); +export type IdentitySettings = z.infer; + export const GlobalSettings = z.object({ automation: AutomationSettings, cache: CacheSettings, @@ -368,6 +381,7 @@ export const GlobalSettings = z.object({ metadata: MetadataSettings, sites: SitesSettings, naming: NamingSettings, + identity: IdentitySettings, }); export type GlobalSettings = z.infer; @@ -438,6 +452,7 @@ export const DEFAULT_SETTINGS: GlobalSettings = { userDataMaxSeedingPages: 10, }, naming: { ...DEFAULT_NAMING_TEMPLATES, lockFileNameForBound: false }, + identity: { eagerTmdbLookup: false, poolArchiveDays: 30 }, }; /** PATCH /settings:按分组部分更新;服务端与现值合并后用 GlobalSettings 整体再校验 */ @@ -486,6 +501,7 @@ export const GlobalSettingsUpdate = z.object({ .optional(), sites: SitesSettings.partial().optional(), naming: NamingSettings.partial().optional(), + identity: IdentitySettings.partial().optional(), }); export type GlobalSettingsUpdate = z.infer; @@ -510,6 +526,7 @@ export function mergeSettings(current: GlobalSettings, patch: GlobalSettingsUpda metadata: { ...current.metadata, ...metadata }, sites: { ...(current.sites ?? DEFAULT_SETTINGS.sites), ...patch.sites }, naming: { ...(current.naming ?? DEFAULT_SETTINGS.naming), ...patch.naming }, + identity: { ...(current.identity ?? DEFAULT_SETTINGS.identity), ...patch.identity }, }; } @@ -666,11 +683,12 @@ export const SETTINGS_SECTIONS: readonly SettingsSectionDef[] = [ { id: "metadata", label: "元数据", - description: "TMDB API Key 与地址、TMDB 缓存、图片代理", + description: "TMDB API Key 与地址、TMDB 缓存、资源识别、图片代理", legacyTabs: [], anchors: [ { id: "tmdb", label: "TMDB" }, { id: "tmdb-cache", label: "TMDB 缓存" }, + { id: "identity", label: "资源识别" }, { id: "images", label: "图片代理与缓存" }, ], fields: [ @@ -679,6 +697,8 @@ export const SETTINGS_SECTIONS: readonly SettingsSectionDef[] = [ "cache.tmdbTtlHours", "cache.tmdbMaxSize", "cache.emptyResultTtlSec", + "identity.eagerTmdbLookup", + "identity.poolArchiveDays", "cache.globalImageCache", "cache.globalImageCacheDays", ], diff --git a/packages/db/migrations/0047_identity_pool.ts b/packages/db/migrations/0047_identity_pool.ts new file mode 100644 index 00000000..a640d0e4 --- /dev/null +++ b/packages/db/migrations/0047_identity_pool.ts @@ -0,0 +1,102 @@ +import type { Kysely } from "kysely"; +import { execDdl, oneOf } from "../src/migration-helpers.ts"; + +// DEC-1 惰性识别 + 待识别池治理(REQ §4.2–4.3、§7.5、§13;对照 MP:RSS 资源只在匹配订阅 / 用户搜索时识别,识别失败不告警): +// - identity_candidate.reason 新增 unmatched:本地召回无结果、也不属于任何需要 TMDB 查询的触发(订阅别名 / 新作品收录规则 / +// 手动 / 搜索),只进池、不查 TMDB、不开待处理; +// - resolution_attempt.outcome 同步新增 unmatched; +// - identity_candidate.normalized_title:归一化标题(与 title_alias.normalized 同一口径),也是待处理合并分组键 +// (来源 + 归一化标题)的一部分;title_alias 插入触发器按它唤醒相近的池条目复查(新证据到来即重试); +// - 清理存量:未关闭的 identity_unresolved 事项里,资源未关联作品、池内无候选、标题也不命中任何本地别名的, +// 事项关闭为 dismissed(resolution.resolution = 'moved_to_pool'),池条目改为 unmatched 并解除事项关联。 + +const OLD_REASONS = ["ambiguous", "no_match", "proposed", "lookup_failed"] as const; +const OLD_OUTCOMES = [ + "auto", + "proposed", + "ambiguous", + "no_match", + "lookup_failed", + "manual", + "skipped_locked", +] as const; + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl( + db, + ` +alter table identity_candidate drop constraint identity_candidate_reason_check; +alter table identity_candidate add constraint identity_candidate_reason_check + ${oneOf("reason", [...OLD_REASONS, "unmatched"])}; +alter table resolution_attempt drop constraint resolution_attempt_outcome_check; +alter table resolution_attempt add constraint resolution_attempt_outcome_check + ${oneOf("outcome", [...OLD_OUTCOMES, "unmatched"])}; + +alter table identity_candidate add column normalized_title text; +update identity_candidate set normalized_title = nullif(query->>'normalized', ''); +create index identity_candidate_normalized_trgm_idx on identity_candidate + using gin (normalized_title gin_trgm_ops) where status = 'pending'; + +-- 新证据唤醒:任何路径新增别名(建档 / 手动加别名 / 合并)后,归一化标题相同或足够相近(与本地召回同阈值 0.45)的 +-- unmatched / no_match 池条目立即到期复查(复查仍是本地召回优先,命中别名才查 TMDB) +create function identity_pool_wake_on_alias() returns trigger language plpgsql as $$ +begin + if new.normalized is null or new.normalized = '' then return null; end if; + update identity_candidate set next_review_at = now() + where status = 'pending' and reason in ('unmatched', 'no_match') and normalized_title is not null + and (next_review_at is null or next_review_at > now()) + and (normalized_title = new.normalized + or (length(new.normalized) >= 3 and normalized_title % new.normalized + and similarity(normalized_title, new.normalized) >= 0.45)); + return null; +end $$; +create trigger title_alias_wake_identity_pool after insert on title_alias + for each row execute function identity_pool_wake_on_alias(); + +with moved as ( + select p.id as pool_id, d.id as decision_id + from identity_candidate p + join source_release r on r.id = p.source_release_id + join decision d on d.id = p.decision_id or d.dedupe_key = 'identity:' || p.source_release_id::text + where d.kind = 'identity_unresolved' + and d.status in ('open', 'investigating', 'awaiting_user') + and p.status = 'pending' + and r.media_entity_id is null + and jsonb_array_length(p.candidates) = 0 + and not exists ( + select 1 from title_alias a + where p.normalized_title is not null + and (a.normalized = p.normalized_title + or (length(p.normalized_title) >= 3 and similarity(a.normalized, p.normalized_title) >= 0.45))) +), closed as ( + update decision set status = 'dismissed', + resolution = jsonb_build_object('resolution', 'moved_to_pool', 'by', 'migration:0047_identity_pool'), + resolved_by = 'system:identity-pool', resolved_at = now() + where id in (select decision_id from moved) + returning id +) +update identity_candidate set reason = 'unmatched', decision_id = null, next_review_at = null + where id in (select pool_id from moved); +`, + ); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl( + db, + ` +update identity_candidate set reason = 'no_match' where reason = 'unmatched'; +update resolution_attempt set outcome = 'no_match' where outcome = 'unmatched'; +drop trigger if exists title_alias_wake_identity_pool on title_alias; +drop function if exists identity_pool_wake_on_alias(); +drop index if exists identity_candidate_normalized_trgm_idx; +alter table identity_candidate drop column normalized_title; +alter table identity_candidate drop constraint identity_candidate_reason_check; +alter table identity_candidate add constraint identity_candidate_reason_check ${oneOf("reason", OLD_REASONS)}; +alter table resolution_attempt drop constraint resolution_attempt_outcome_check; +alter table resolution_attempt add constraint resolution_attempt_outcome_check ${oneOf("outcome", OLD_OUTCOMES)}; +`, + ); +} diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index d785e19d..20808a33 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -139,7 +139,7 @@ export type FeedSourceHealthStatus = "unknown" | "healthy" | "degraded" | "faili export type FieldOverrideSource = "tmdb" | "user" | "agent" | "nfo" | "site" | "import"; export type FieldOverrideActorType = "user" | "agent" | "system" | "api_key"; export type IdentityCandidateStatus = "pending" | "resolved" | "dismissed" | "archived"; -export type IdentityCandidateReason = "ambiguous" | "no_match" | "proposed" | "lookup_failed"; +export type IdentityCandidateReason = "ambiguous" | "no_match" | "proposed" | "lookup_failed" | "unmatched"; export type IdentityRedirectActorType = "user" | "agent" | "system" | "api_key"; export type IdentityRevisionAction = | "created" @@ -210,7 +210,8 @@ export type ResolutionAttemptOutcome = | "no_match" | "lookup_failed" | "manual" - | "skipped_locked"; + | "skipped_locked" + | "unmatched"; export type ResolutionAttemptActorType = "user" | "agent" | "system" | "api_key"; export type RuleGroupKind = "acquisition" | "upgrade" | "search"; export type RuleGroupSource = "manual" | "chat" | "agent"; @@ -787,6 +788,7 @@ export interface IdentityCandidateTable { status: Generated; reason: IdentityCandidateReason; query: Generated; + normalized_title: string | null; candidates: Generated; evidence_version: string; attempts: Generated;