From fcd0d8f223ea3da7defbcf6010bff1e5fe9c0e45 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 3 Oct 2026 23:22:16 +0800 Subject: [PATCH] =?UTF-8?q?fix(acquisition):=20=E8=AE=A2=E9=98=85=E6=80=BB?= =?UTF-8?q?=E5=B8=A6=E8=87=AA=E5=B7=B1=E7=9A=84=E6=94=B6=E5=BD=95=E7=89=88?= =?UTF-8?q?=E6=9C=AC=E3=80=81=E6=B4=97=E7=89=88=E5=88=B0=E6=9C=80=E9=AB=98?= =?UTF-8?q?=E6=A1=A3=E6=89=8D=E5=AE=8C=E6=88=90=E3=80=81=E6=8E=A5=E7=BB=AD?= =?UTF-8?q?=E5=90=8E=E7=BB=AD=E5=AD=A3=E5=8A=A0=E5=AD=A3=E4=B8=8B=E7=95=8C?= =?UTF-8?q?=E3=80=81=E9=87=8D=E8=AF=84=E6=8B=A9=E4=BC=98=E5=B9=B6=E6=94=BE?= =?UTF-8?q?=E5=AE=BD=E8=8C=83=E5=9B=B4=E3=80=81=E6=89=8B=E5=8A=A8=E4=B8=8B?= =?UTF-8?q?=E8=BD=BD=E8=AE=A1=E5=85=A5=E5=9C=A8=E9=80=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - H1:新建 / 编辑订阅时没有自己的收录版本就按策略模板建,没有模板建隐式「默认版本」(质量预设随后写入); 旧订阅留下的同名版本改名加序号;已有的 0 收录版本订阅在首次评估 / 订阅重评时补建 - H2:收录版本有升级目标(规则组 / 阶梯按 pri 洗版)时,订阅完成要求覆盖齐且每个已覆盖单元到目标 pri; slotComplete 多返回 atTarget,接收即计入与入库后两处完成判定统一用 completesOnCoverage + slotDone - H4:「接续后续季」评估只看不早于订阅季的资源,与覆盖矩阵同口径(seasonInScope) - M4:按收录版本 / 订阅重评时先用 rankForSlot 按主收录版本择优排序再逐条预留;订阅重评只评该订阅的收录版本 - M5:重评取最近 500 条(原 50 条),先按订阅季范围、已覆盖单元在内存里过滤再逐条评估 - M6:同作品手动下载中的计划(slot_id 为空)并入在途单元,冲突延后时排退避重评 - 资源解析 / 站点上下文挪到 release-facts.ts,评估与完成判定共用,避免循环引用 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- .../routes/subscriptions.ts | 37 +- .../services/acquisition.ts | 368 +++++++++++------- .../services/release-facts.ts | 70 ++++ .../services/subscriptions.ts | 333 +++++++++++++++- .../upgrade-completion.test.ts | 155 ++++++++ .../subscription-scope-reevaluate.test.ts | 222 +++++++++++ 6 files changed, 1000 insertions(+), 185 deletions(-) create mode 100644 apps/server/src/modules/acquisition-organize/services/release-facts.ts create mode 100644 apps/server/test/acquisition-organize/upgrade-completion.test.ts create mode 100644 apps/server/test/subscriptions/subscription-scope-reevaluate.test.ts diff --git a/apps/server/src/modules/acquisition-organize/routes/subscriptions.ts b/apps/server/src/modules/acquisition-organize/routes/subscriptions.ts index 46385d26..4aed224c 100644 --- a/apps/server/src/modules/acquisition-organize/routes/subscriptions.ts +++ b/apps/server/src/modules/acquisition-organize/routes/subscriptions.ts @@ -2,7 +2,6 @@ import { type AcquisitionSlot, Api, DEFAULT_SEARCH_INTERVAL_HOURS, - type SlotTemplate, type Subscription, type SubscriptionGroup, type SubscriptionHistory, @@ -16,7 +15,7 @@ import type { Context } from "hono"; import type { AuthEnv } from "../../../http/auth.ts"; import { ApiException } from "../../../http/errors.ts"; import type { ModuleApp } from "../../types.ts"; -import { createSlotsFromTemplates, publishedAcquisitionPolicies } from "../services/acquisition.ts"; +import { ensureSubscriptionSlot } from "../services/acquisition.ts"; import type { AoDeps, Exec } from "../services/common.ts"; import { buildSubscriptionModels, @@ -107,9 +106,6 @@ async function assertNoDuplicate( if (hit) conflict(`该作品同一季已有订阅：${hit.id}`); } -/** 没有收录策略模板、表单给了质量预设时,新订阅按这个模板建一个收录版本(预设随后写入) */ -const DEFAULT_SLOT_TEMPLATE: SlotTemplate = { name: "默认版本" }; - /** * 保存后应用表单的派生写入:包含 / 排除 → 订阅私有规则组;质量预设(MP resolution / quality / effect)→ * 该订阅自己的收录版本;MP total_episode → 季总集数;best_version → 收录版本升级策略。 @@ -376,33 +372,8 @@ export function registerSubscriptionRoutes(app: ModuleApp, d: AoDeps) { }) .returningAll() .executeTakeFirstOrThrow(); - if (r.media_entity_id) { - const has = await trx - .selectFrom("acquisition_slot") - .select("id") - .where("media_entity_id", "=", r.media_entity_id) - .executeTakeFirst(); - if (!has) { - const pols = await publishedAcquisitionPolicies(trx); - const pol = - (body.policyRevisionId ? pols.find((p) => p.row.id === body.policyRevisionId) : undefined) ?? - pols.find((p) => p.rules.defaultSlots?.length); - // 策略没有模板但表单给了质量预设(MP resolution / quality / effect):建一个"默认版本"承载预设 - const preset = !!(norm.params.resolution || norm.params.source || norm.params.effect); - const templates = pol?.rules.defaultSlots?.length - ? pol.rules.defaultSlots - : preset - ? [DEFAULT_SLOT_TEMPLATE] - : []; - if (templates.length) - await createSlotsFromTemplates(trx, { - entityId: r.media_entity_id, - subscriptionId: r.id, - policyRevisionId: pol?.rules.defaultSlots?.length ? pol.row.id : null, - templates, - }); - } - } + // 订阅总有自己的收录版本:策略模板,没有就建一个默认版本(质量预设随后由 applyDerived 写入) + await ensureSubscriptionSlot(trx, r); await applyDerived(trx, r, norm, now); return r; }); @@ -458,6 +429,8 @@ export function registerSubscriptionRoutes(app: ModuleApp, d: AoDeps) { .where("id", "=", params.id) .returningAll() .executeTakeFirstOrThrow(); + // 旧订阅没有自己的收录版本:编辑时补建(质量预设 / 升级策略才有地方写) + if (r.status !== "cancelled") await ensureSubscriptionSlot(trx, r); if (norm) await applyDerived(trx, r, norm, d.ports.now(), subParams(cur)); else if (body.status !== undefined) await syncPrivateRuleGroup(trx, r, d.ports.now()); return r; diff --git a/apps/server/src/modules/acquisition-organize/services/acquisition.ts b/apps/server/src/modules/acquisition-organize/services/acquisition.ts index a9a6f36f..edebcf38 100644 --- a/apps/server/src/modules/acquisition-organize/services/acquisition.ts +++ b/apps/server/src/modules/acquisition-organize/services/acquisition.ts @@ -1,10 +1,4 @@ -import type { - AcquisitionPolicyRules, - AcquisitionPriority, - AcquisitionSlot, - CandidateEvaluation, - SlotTemplate, -} from "@umt/contracts"; +import type { AcquisitionPolicyRules, AcquisitionPriority, CandidateEvaluation, SlotTemplate } from "@umt/contracts"; import type { AcquisitionSlotRow, MediaEntityRow, SourceReleaseRow, SubscriptionRow } from "@umt/db"; import { sql } from "@umt/db"; import { @@ -12,15 +6,14 @@ import { checkBudget, computeUnitGaps, determineMode, - type EvaluationContext, effectiveLadder, episodeKey, evaluateCandidate, improvableUnits, type ParsedRelease, - parseRelease, priorityOf, type QualityFacts, + rankForSlot, ruleChainOf, type SlotEvaluation, type SlotSpec, @@ -29,8 +22,8 @@ import { transition, unitsOverlap, } from "@umt/domain"; -import { fromParsedColumns, slotSpecFromRow, toContractSlotEvaluation } from "@umt/mapping"; -import { hashJson, jsonb, startOfDay, toApi } from "../util.ts"; +import { toContractSlotEvaluation } from "@umt/mapping"; +import { hashJson, jsonb, startOfDay } from "../util.ts"; import { type AoDeps, automationBlock, @@ -45,24 +38,32 @@ import { recordCandidate, } from "./common.ts"; import { autoFillCheck, type ForceFill } from "./evaluate-autofill.ts"; -import { entityOriginalLanguage, loadRuleCatalog, mergeHard, type RuleCatalog, resolveRuleChain } from "./rules.ts"; +import { contextOf, episodeResolutionOf, parsedOf, unitKeysOf } from "./release-facts.ts"; +import { entityOriginalLanguage, loadRuleCatalog, type RuleCatalog } from "./rules.ts"; import { clearDeferral, clearReleaseDeferrals, clearSubscriptionDeferrals, gateRetryDelayMs, + manualInflightPlans, rangeOf, recordDeferral, scheduleSlotReevaluate, scheduleSlotRetry, + seasonInScope, seasonLedgers, slotCoveredUnits, slotInflightPlans, + slotSpecWithRules, subParams, + subSeasonOf, TRACKING_SUB_STATUSES, } from "./subscriptions.ts"; +import { applySlotPreset } from "./subscriptions-params.ts"; import { decideVolume, loadVolumes } from "./volumes.ts"; +export { contextOf, parsedOf, unitKeysOf } from "./release-facts.ts"; + // evaluateAcquisition:release.resolved → 匹配订阅/Slot → 模式 → 确定性评估 → 预算/缓存卷 → 冻结计划并预留。 // 拒绝与被挡住的原因都写成候选评估记录(artifact),供作品详情"为什么选/不选"。 @@ -98,34 +99,6 @@ const PRIORITY: Record, AcquisitionPriority> = { const BUSY: readonly string[] = ["reserved", "downloading", "verifying"]; -interface StoredEpisodeResolution { - method?: string; - special?: string; - canonical?: { season: number; episodes: { start: number; end: number } } | null; -} - -const episodeResolutionOf = (r: Pick): StoredEpisodeResolution | null => - r.episode_resolution && typeof r.episode_resolution === "object" - ? (r.episode_resolution as StoredEpisodeResolution) - : null; - -/** - * source_release → ParsedRelease(解析版本过期时按原标题重解析;站点体积覆盖解析体积)。 - * 重解析时把集数解析的规范季集叠加回去(parsed 列本来就是规范值,重解析不能丢映射)。 - */ -export function parsedOf(r: SourceReleaseRow): ParsedRelease { - let p = fromParsedColumns({ parsed: (r.parsed ?? {}) as Record, parserVersion: r.parser_version }); - if (!p) { - p = parseRelease(r.title, r.subtitle ? { subtitle: r.subtitle } : {}); - const c = episodeResolutionOf(r)?.canonical; - if (c) { - const { seasonEnd: _drop, ...rest } = p; - p = { ...rest, season: c.season, episodes: { ...c.episodes }, seasonInferred: false }; - } - } - return r.size_bytes !== null ? { ...p, sizeBytes: r.size_bytes } : p; -} - /** 集数待确认(无法换算 / 规则失效)或映射为特别篇 / 花絮 / 忽略:不参与订阅评估 */ export function episodeUnresolved(r: Pick): boolean { const res = episodeResolutionOf(r); @@ -145,16 +118,6 @@ export async function currentFacts(exec: Exec, assetId: string | null): Promise< return { ...q, ...(a.size_bytes !== null ? { sizeBytes: a.size_bytes } : {}) }; } -/** 单元键:有 content_unit 用其 id,没有时用 s{季}e{集} 合成键(追更判定两侧同口径) */ -export function unitKeysOf(p: Pick): string[] { - if (p.airDate) return [`d${p.airDate}`]; - if (!p.episodes) return []; - const s = p.season ?? 1; - const out: string[] = []; - for (let e = p.episodes.start; e <= p.episodes.end && out.length < 500; e++) out.push(`s${s}e${e}`); - return out; -} - async function releaseUnits( exec: Exec, entityId: string, @@ -227,9 +190,24 @@ export async function createSlotsFromTemplates( policyRevisionId: string | null; templates: readonly SlotTemplate[]; }, -): Promise { +): Promise { + // 收录版本名在作品内唯一(media_entity_id, content_unit_id, name):旧订阅留下的同名版本不挡新订阅,改名加序号 + const taken = new Set( + ( + await exec + .selectFrom("acquisition_slot") + .select("name") + .where("media_entity_id", "=", input.entityId) + .where("content_unit_id", "is", null) + .execute() + ).map((r) => r.name), + ); + let n = 0; for (const [i, t] of input.templates.entries()) { - await exec + let name = t.name; + for (let k = 2; taken.has(name) && k < 1000; k++) name = `${t.name} ${k}`; + taken.add(name); + const row = await exec .insertInto("acquisition_slot") .values({ // 同一事务建的收录版本 created_at 相同:按模板顺序给 position,第一个为主版本 @@ -237,7 +215,7 @@ export async function createSlotsFromTemplates( media_entity_id: input.entityId, subscription_id: input.subscriptionId, policy_revision_id: input.policyRevisionId, - name: t.name, + name, hard_conditions: jsonb(t.hardConditions), soft_preferences: jsonb(t.softPreferences), fallback: jsonb(t.fallback), @@ -246,8 +224,58 @@ export async function createSlotsFromTemplates( rule_group_ids: JSON.stringify(t.ruleGroupIds ?? []), }) .onConflict((oc) => oc.doNothing()) - .execute(); + .returning("id") + .executeTakeFirst(); + if (row) n++; } + return n; +} + +/** 追踪中的订阅里没有自己收录版本的,逐个补建;返回是否建了 */ +async function backfillSubscriptionSlots( + exec: Exec, + subs: readonly SubscriptionRow[], + slots: readonly Pick[], +): Promise { + let made = 0; + for (const s of subs) + if (s.media_entity_id && !slots.some((x) => x.subscription_id === s.id)) + made += await ensureSubscriptionSlot(exec, s); + return made > 0; +} + +/** 订阅没有收录策略模板时的隐式缺省收录版本:无硬条件,表单的质量预设 / 包含排除随后由订阅参数写入 */ +export const DEFAULT_SLOT_TEMPLATE: SlotTemplate = { name: "默认版本" }; + +/** + * 订阅总有属于自己的收录版本(H1):没有时按收录策略模板(指定的或第一个带 defaultSlots 的已发布策略)建, + * 没有模板就建一个隐式「默认版本」,再把订阅表单的质量预设写进去。作品已有别的订阅(旧季、已取消)的 + * 收录版本不算——评估时订阅只用自己的收录版本。返回新建个数(已有则 0)。 + */ +export async function ensureSubscriptionSlot( + exec: Exec, + sub: Pick, +): Promise { + if (!sub.media_entity_id) return 0; + const own = await exec + .selectFrom("acquisition_slot") + .select("id") + .where("subscription_id", "=", sub.id) + .executeTakeFirst(); + if (own) return 0; + const pols = await publishedAcquisitionPolicies(exec); + const pol = + (sub.policy_revision_id ? pols.find((p) => p.row.id === sub.policy_revision_id) : undefined) ?? + pols.find((p) => p.rules.defaultSlots?.length); + const fromPolicy = !!pol?.rules.defaultSlots?.length; + const n = await createSlotsFromTemplates(exec, { + entityId: sub.media_entity_id, + subscriptionId: sub.id, + policyRevisionId: fromPolicy ? pol!.row.id : null, + templates: fromPolicy ? pol!.rules.defaultSlots! : [DEFAULT_SLOT_TEMPLATE], + }); + if (n > 0) await applySlotPreset(exec, { subscriptionId: sub.id, prev: {}, next: subParams(sub) }); + return n; } /** @@ -291,6 +319,13 @@ interface Ctx { forceFill?: ForceFill | undefined; } +/** 作品原语言(规则组 CNSUB/CNVOI 放行),一次评估内缓存 */ +function entityLang(c: Ctx, entityId: string): Promise { + c.langs ??= new Map(); + if (!c.langs.has(entityId)) c.langs.set(entityId, entityOriginalLanguage(c.d.ctx.db, entityId)); + return c.langs.get(entityId)!; +} + async function note(c: Ctx, base: Omit) { await recordCandidate(c.d.ctx.db, { ...base, evaluatedAt: c.now.toISOString() }); } @@ -314,32 +349,49 @@ export async function evaluateAcquisition(d: AoDeps, input: EvaluateInput): Prom c.out.skipped.push({ sourceReleaseId: input.sourceReleaseId, reason: "release_not_found" }); return c.out; } - await evaluateRelease(c, rel, input.contentUnitIds ?? [], input.acquisitionSlotId); + await evaluateRelease(c, rel, input.contentUnitIds ?? [], { slotId: input.acquisitionSlotId }); return c.out; } - // 按 Slot / 订阅重评:取该作品最近已识别的发布 - let entityId: string | null = null; + return reevaluate(c, input); +} + +/** + * 按收录版本 / 订阅重评时最多取的已识别资源条数(按首次出现倒序)。原来只看最近 50 条,长剧、多发布组时早期集 + * 进不了评估(M5);放宽到 500 后先在内存里按订阅季范围、已覆盖单元过滤,只把可能有用的逐条评估 + * (每条评估约十几次查询,500 条全过一遍也在几秒量级)。 + */ +export const REEVALUATE_MAX_RELEASES = 500; + +/** + * 按收录版本 / 订阅重评:延后的资源 + 作品已识别资源(按范围过滤),先用 rankForSlot 按主收录版本择优排序 + * (pri → 来源 / 分辨率等级 → 软偏好),再逐条评估预留(M4:同一单元多个合格资源时 pri 最高的先占位, + * 其余延后 / 洗版判断)。逐条到来的 RSS 资源仍走 evaluateRelease,不受影响。 + */ +async function reevaluate(c: Ctx, input: EvaluateInput): Promise { + const db = c.d.ctx.db; + let sub: SubscriptionRow | undefined; + let focus: AcquisitionSlotRow[] = []; if (input.acquisitionSlotId) { - entityId = - ( - await db - .selectFrom("acquisition_slot") - .select("media_entity_id") - .where("id", "=", input.acquisitionSlotId) - .executeTakeFirst() - )?.media_entity_id ?? null; + focus = await db.selectFrom("acquisition_slot").selectAll().where("id", "=", input.acquisitionSlotId).execute(); + const sid = focus[0]?.subscription_id; + sub = sid ? await db.selectFrom("subscription").selectAll().where("id", "=", sid).executeTakeFirst() : undefined; } else if (input.subscriptionId) { - entityId = - ( - await db - .selectFrom("subscription") - .select("media_entity_id") - .where("id", "=", input.subscriptionId) - .executeTakeFirst() - )?.media_entity_id ?? null; + sub = await db.selectFrom("subscription").selectAll().where("id", "=", input.subscriptionId).executeTakeFirst(); + if (sub && (TRACKING_SUB_STATUSES as readonly string[]).includes(sub.status)) await ensureSubscriptionSlot(db, sub); + if (sub) + focus = await db + .selectFrom("acquisition_slot") + .selectAll() + .where("subscription_id", "=", sub.id) + .orderBy("position") + .orderBy("created_at") + .execute(); } + const entityId = focus[0]?.media_entity_id ?? sub?.media_entity_id ?? null; if (!entityId) return c.out; - // 先评该 Slot 延后的发布(按集号升序,连续更新的集依次补上),再评最近已识别的发布 + const entity = await resolveEntity(db, entityId); + if (!entity) return c.out; + // 先评该 Slot 延后的发布(按集号升序,连续更新的集依次补上),再评作品其它已识别的发布 const deferred = input.acquisitionSlotId ? await db .selectFrom("slot_deferral as sd") @@ -362,14 +414,93 @@ export async function evaluateAcquisition(d: AoDeps, input: EvaluateInput): Prom .where("resolution_status", "=", "resolved") .where("availability", "=", "active") .orderBy("first_seen_at", "desc") - .limit(50) + .limit(REEVALUATE_MAX_RELEASES) .execute(); - for (const r of [...deferred, ...rels.filter((x) => !seen.has(x.id))]) - await evaluateRelease(c, r, [], input.acquisitionSlotId); + const fresh = await inScopeReleases( + c, + entity, + sub, + focus, + rels.filter((x) => !seen.has(x.id)), + ); + const ordered = await rankReleases(c, entity, sub, focus[0], [...deferred, ...fresh]); + const scope = input.acquisitionSlotId ? { slotId: input.acquisitionSlotId } : sub ? { subscriptionId: sub.id } : {}; + for (const r of ordered) await evaluateRelease(c, r, [], scope); return c.out; } -async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readonly string[], onlySlotId?: string) { +/** + * 重评候选的预过滤(M5):订阅季范围外的(与评估同口径)不评;剧集单集资源的单元在每个目标收录版本里都已覆盖、 + * 且这些收录版本都没开升级时不评。季包 / 综艺按日期 / 电影原样保留,交给逐条评估。 + */ +async function inScopeReleases( + c: Ctx, + entity: MediaEntityRow, + sub: SubscriptionRow | undefined, + focus: readonly AcquisitionSlotRow[], + rels: SourceReleaseRow[], +): Promise { + if (entity.kind === "movie") return rels; + const db = c.d.ctx.db; + const subSeason = sub ? await subSeasonOf(db, sub) : null; + const upgrading = focus.some((s) => !!(s.upgrade_threshold as { enabled?: boolean } | null)?.enabled); + const slotCount = focus.length + ? Number( + ( + await db + .selectFrom("acquisition_slot") + .select((eb) => eb.fn.countAll().as("n")) + .where("media_entity_id", "=", entity.id) + .executeTakeFirstOrThrow() + ).n, + ) + : 0; + const covered = + focus.length && !upgrading + ? await Promise.all(focus.map((s) => slotCoveredUnits(db, s, { entityWide: slotCount <= 1 }))) + : []; + return rels.filter((r) => { + const p = parsedOf(r); + const season = p.season ?? (p.episodes ? 1 : undefined); + if (sub && season !== undefined && !seasonInScope(sub, subSeason, season)) return false; + if (!covered.length || !p.episodes) return true; + const keys = unitKeysOf(p); + return !covered.every((cv) => keys.every((k) => cv.keys.has(k))); + }); +} + +/** 按主收录版本择优排序(rankForSlot);没有收录版本时保持原顺序 */ +async function rankReleases( + c: Ctx, + entity: MediaEntityRow, + sub: SubscriptionRow | undefined, + slot: AcquisitionSlotRow | undefined, + rels: SourceReleaseRow[], +): Promise { + if (!slot || rels.length < 2) return rels; + c.rules ??= loadRuleCatalog(c.d.ctx.db); + const spec = slotSpecWithRules(await c.rules, slot, sub, entity); + const lang = spec.rules ? await entityLang(c, entity.id) : undefined; + const byParsed = new Map(); + const cands = rels.map((r) => { + const p = parsedOf(r); + byParsed.set(p, r); + return p; + }); + return rankForSlot(spec, cands, null, (p) => ({ + ...contextOf(byParsed.get(p)!, c.now), + ...(lang ? { originalLanguage: lang } : {}), + })).map((x) => byParsed.get(x.candidate)!); +} + +/** 评估范围:只评某个收录版本,或只评某个订阅的收录版本;缺省评作品的全部收录版本 */ +interface EvalScope { + slotId?: string | undefined; + subscriptionId?: string | undefined; +} + +async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readonly string[], scope: EvalScope = {}) { + const onlySlotId = scope.slotId; const { d, now } = c; const db = d.ctx.db; const skip = (reason: string) => c.out.skipped.push({ sourceReleaseId: rel.id, reason }); @@ -495,12 +626,22 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon const activeSubs = new Map( subs.filter((s) => (TRACKING_SUB_STATUSES as readonly string[]).includes(s.status)).map((s) => [s.id, s]), ); + // 一次性补建:旧版本建的订阅可能没有自己的收录版本(H1),首次评估时补上 + if (await backfillSubscriptionSlots(db, [...activeSubs.values()], slots)) + slots = await db + .selectFrom("acquisition_slot") + .selectAll() + .where("media_entity_id", "=", entity.id) + .orderBy("position") + .orderBy("created_at") + .execute(); const tracked = activeSubs.size > 0 || slots.some((s) => s.subscription_id === null); const releaseSeason = parsed.season ?? (parsed.episodes ? 1 : undefined); let planned: string | null = null; for (const slot of slots) { if (onlySlotId && slot.id !== onlySlotId) continue; + if (scope.subscriptionId && slot.subscription_id !== scope.subscriptionId) continue; const sub = slot.subscription_id ? activeSubs.get(slot.subscription_id) : undefined; if (slot.subscription_id && !sub) { // 订阅已暂停/完成/取消:不评估,也不留延后行 @@ -523,15 +664,8 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon .executeTakeFirst(); if (!seasonOk) continue; } - // 仅本季:发布季号与订阅季不一致则跳过 - if (sub?.season_scope === "current_season" && sub.season_unit_id && releaseSeason !== undefined) { - const su = await db - .selectFrom("content_unit") - .select("season_number") - .where("id", "=", sub.season_unit_id) - .executeTakeFirst(); - if (su?.season_number !== null && su?.season_number !== undefined && su.season_number !== releaseSeason) continue; - } + // 订阅季范围:仅本季 = 同季,接续后续季 = 不早于订阅季(与覆盖矩阵同口径) + if (sub && releaseSeason !== undefined && !seasonInScope(sub, await subSeasonOf(db, sub), releaseSeason)) continue; c.out.evaluated++; const outcome = await evaluateForSlot(c, { rel, @@ -583,33 +717,6 @@ async function clearEpisodeUnresolvedNotes(c: Ctx, sourceReleaseId: string, enti and content->'reasons' ? 'episode_unresolved'`.execute(c.d.ctx.db); } -/** 站点字段 → 评估上下文(做种数、促销下载系数、副标题/标签、站点 id、发布后分钟数) */ -export function contextOf(rel: SourceReleaseRow, now: Date = new Date()): EvaluationContext { - const promo = (rel.promotion ?? {}) as { kind?: string; raw?: { downloadFactor?: number | null } }; - const factor = - typeof promo.raw?.downloadFactor === "number" - ? promo.raw.downloadFactor - : promo.kind === "free" || promo.kind === "free2up" - ? 0 - : promo.kind === "50pctdown" || promo.kind === "50pctdown2up" - ? 0.5 - : promo.kind === "30pctdown" - ? 0.3 - : promo.kind === "none" || promo.kind === "2up" - ? 1 - : undefined; - return { - siteId: rel.site_id, - ...(rel.subtitle ? { subtitle: rel.subtitle } : {}), - labels: rel.tags_raw ?? [], - ...(rel.seeders !== null && rel.seeders !== undefined ? { seeders: rel.seeders } : {}), - ...(factor !== undefined ? { downloadVolumeFactor: factor } : {}), - ...(rel.source_published_at - ? { publishedMinutesAgo: Math.max(0, (now.getTime() - rel.source_published_at.getTime()) / 60_000) } - : {}), - }; -} - interface SlotInput { rel: SourceReleaseRow; entity: MediaEntityRow; @@ -642,37 +749,26 @@ async function evaluateForSlot(c: Ctx, p: SlotInput): Promise<{ planId: string | await note(c, { ...base, decision: "skipped", mode: null, rejected: [], reasons: ["site_out_of_scope"] }); return done; } - const spec = slotSpecFromRow(toApi(slot)); - const ctx = contextOf(rel, now); - // 规则组:收录版本 ruleGroupIds → 订阅 ruleGroupIds → 订阅默认链;开升级叠加洗版默认链 + // 规则组:收录版本 ruleGroupIds → 订阅 ruleGroupIds → 订阅默认链;开升级叠加洗版默认链;私有组包含 / 排除并入硬条件 c.rules ??= loadRuleCatalog(db); - const chain = resolveRuleChain(await c.rules, { - kind: "acquisition", - slotRuleGroupIds: (slot.rule_group_ids ?? []) as string[], - subRuleGroupIds: params.ruleGroupIds, - upgrade: !!spec.upgrade?.enabled, - mediaKind: subKindOf(p.entity.kind), - entityId: p.entity.id, - siteId: rel.site_id, - subscriptionId: sub?.id, - }); - // 规则组的结构化硬条件(订阅私有组的包含 / 排除)并入收录版本硬条件 - spec.hard = mergeHard(spec.hard, chain.hard); - if (chain.groups.length) { - spec.rules = { groups: chain.groups, atoms: chain.atoms }; - c.langs ??= new Map(); - if (!c.langs.has(p.entity.id)) c.langs.set(p.entity.id, entityOriginalLanguage(db, p.entity.id)); - const lang = await c.langs.get(p.entity.id); + const spec = slotSpecWithRules(await c.rules, slot, sub, p.entity, rel.site_id); + const ctx = contextOf(rel, now); + if (spec.rules) { + const lang = await entityLang(c, p.entity.id); if (lang) ctx.originalLanguage = lang; } - const inflight = await slotInflightPlans(db, slot.id); + // 在途 = 本收录版本的计划 + 同作品手动下载中的计划(M6) + const manual = await manualInflightPlans(db, p.entity.id); + const inflight = [...(await slotInflightPlans(db, slot.id)), ...manual]; const inflightKeys = inflight.flatMap((x) => x.unitKeys); const deferWithRetry = async (slotId: string, releaseId: string, reason: string, unitKeys: string[]) => { const attempts = await recordDeferral(db, { slotId, sourceReleaseId: releaseId, reason, unitKeys }); await scheduleSlotRetry(db, slotId, new Date(now.getTime() + gateRetryDelayMs(attempts))); }; const defer = async (reason: string, unitKeys: string[], mode: AcquisitionMode | null = null) => { - await recordDeferral(db, { slotId: slot.id, sourceReleaseId: rel.id, reason, unitKeys }); + const attempts = await recordDeferral(db, { slotId: slot.id, sourceReleaseId: rel.id, reason, unitKeys }); + // 手动计划结束不会触发本收录版本重评:有手动在途时按次数退避排一次重评 + if (manual.length) await scheduleSlotRetry(db, slot.id, new Date(now.getTime() + gateRetryDelayMs(attempts))); await note(c, { ...base, decision: "deferred", diff --git a/apps/server/src/modules/acquisition-organize/services/release-facts.ts b/apps/server/src/modules/acquisition-organize/services/release-facts.ts new file mode 100644 index 00000000..208f4b7f --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/services/release-facts.ts @@ -0,0 +1,70 @@ +import type { SourceReleaseRow } from "@umt/db"; +import { type EvaluationContext, type ParsedRelease, parseRelease } from "@umt/domain"; +import { fromParsedColumns } from "@umt/mapping"; + +// 资源行 → 评估输入(解析结果、站点上下文)。评估(acquisition.ts)与完成判定(subscriptions.ts)共用,单独成文件免得两者循环引用。 + +interface StoredEpisodeResolution { + method?: string; + special?: string; + canonical?: { season: number; episodes: { start: number; end: number } } | null; +} + +export const episodeResolutionOf = (r: Pick): StoredEpisodeResolution | null => + r.episode_resolution && typeof r.episode_resolution === "object" + ? (r.episode_resolution as StoredEpisodeResolution) + : null; + +/** + * source_release → ParsedRelease(解析版本过期时按原标题重解析;站点体积覆盖解析体积)。 + * 重解析时把集数解析的规范季集叠加回去(parsed 列本来就是规范值,重解析不能丢映射)。 + */ +export function parsedOf(r: SourceReleaseRow): ParsedRelease { + let p = fromParsedColumns({ parsed: (r.parsed ?? {}) as Record, parserVersion: r.parser_version }); + if (!p) { + p = parseRelease(r.title, r.subtitle ? { subtitle: r.subtitle } : {}); + const c = episodeResolutionOf(r)?.canonical; + if (c) { + const { seasonEnd: _drop, ...rest } = p; + p = { ...rest, season: c.season, episodes: { ...c.episodes }, seasonInferred: false }; + } + } + return r.size_bytes !== null ? { ...p, sizeBytes: r.size_bytes } : p; +} + +/** 单元键:有 content_unit 用其 id,没有时用 s{季}e{集} 合成键(追更判定两侧同口径) */ +export function unitKeysOf(p: Pick): string[] { + if (p.airDate) return [`d${p.airDate}`]; + if (!p.episodes) return []; + const s = p.season ?? 1; + const out: string[] = []; + for (let e = p.episodes.start; e <= p.episodes.end && out.length < 500; e++) out.push(`s${s}e${e}`); + return out; +} + +/** 站点字段 → 评估上下文(做种数、促销下载系数、副标题/标签、站点 id、发布后分钟数) */ +export function contextOf(rel: SourceReleaseRow, now: Date = new Date()): EvaluationContext { + const promo = (rel.promotion ?? {}) as { kind?: string; raw?: { downloadFactor?: number | null } }; + const factor = + typeof promo.raw?.downloadFactor === "number" + ? promo.raw.downloadFactor + : promo.kind === "free" || promo.kind === "free2up" + ? 0 + : promo.kind === "50pctdown" || promo.kind === "50pctdown2up" + ? 0.5 + : promo.kind === "30pctdown" + ? 0.3 + : promo.kind === "none" || promo.kind === "2up" + ? 1 + : undefined; + return { + siteId: rel.site_id, + ...(rel.subtitle ? { subtitle: rel.subtitle } : {}), + labels: rel.tags_raw ?? [], + ...(rel.seeders !== null && rel.seeders !== undefined ? { seeders: rel.seeders } : {}), + ...(factor !== undefined ? { downloadVolumeFactor: factor } : {}), + ...(rel.source_published_at + ? { publishedMinutesAgo: Math.max(0, (now.getTime() - rel.source_published_at.getTime()) / 60_000) } + : {}), + }; +} diff --git a/apps/server/src/modules/acquisition-organize/services/subscriptions.ts b/apps/server/src/modules/acquisition-organize/services/subscriptions.ts index 16fda6df..c5c491cb 100644 --- a/apps/server/src/modules/acquisition-organize/services/subscriptions.ts +++ b/apps/server/src/modules/acquisition-organize/services/subscriptions.ts @@ -1,7 +1,9 @@ import type { AcquisitionSlot, SubscriptionParams, SubscriptionProgress, UnitProgressState } from "@umt/contracts"; -import type { AcquisitionSlotRow, SubscriptionRow } from "@umt/db"; +import type { AcquisitionSlotRow, SourceReleaseRow, SubscriptionRow } from "@umt/db"; import { sql } from "@umt/db"; import { + currentPri, + effectiveLadder, episodeKey, movieCompletion, type ParsedRelease, @@ -10,13 +12,18 @@ import { priorityLadder, type QualityFacts, resolveEpisodeTotal, + ruleChainOf, type SeasonProgress, + type SlotSpec, seasonProgress, + seasonWildcard, type UnitRange, } from "@umt/domain"; import { slotSpecFromRow } from "@umt/mapping"; import { errMessage, jsonb, toApi } from "../util.ts"; import { type AoDeps, type Exec, upsertDecision } from "./common.ts"; +import { contextOf, parsedOf, unitKeysOf } from "./release-facts.ts"; +import { entityOriginalLanguage, loadRuleCatalog, mergeHard, type RuleCatalog, resolveRuleChain } from "./rules.ts"; // W5-C 订阅业务:季总集数账本、缺集/在途、完成判定与历史、Slot 重评调度、分集质量账本、主动搜索调度。 // 对照 docs/porting/moviepilot-subscribe.md §3/§7/§8/§9。 @@ -39,6 +46,30 @@ export function rangeOf(sub: Pick | null | undefined) return { ...(r.start !== undefined ? { start: r.start } : {}), ...(r.end !== undefined ? { end: r.end } : {}) }; } +/** 订阅季号(订阅没有季单元或季单元没有季号时为 null) */ +export async function subSeasonOf(exec: Exec, sub: Pick): Promise { + if (!sub.season_unit_id) return null; + const su = await exec + .selectFrom("content_unit") + .select("season_number") + .where("id", "=", sub.season_unit_id) + .executeTakeFirst(); + return su?.season_number ?? null; +} + +/** + * 发布季号是否在订阅季范围内:仅本季 = 同季;接续后续季 = 不早于订阅季(H4,与覆盖矩阵 coverage.ts 同口径); + * 订阅没有季号时不限。 + */ +export function seasonInScope( + sub: Pick, + subSeason: number | null, + releaseSeason: number, +): boolean { + if (subSeason === null) return true; + return sub.season_scope === "following_seasons" ? releaseSeason >= subSeason : releaseSeason === subSeason; +} + // ── 季总集数账本 ── export interface SeasonLedger { @@ -268,15 +299,259 @@ export async function slotInflightPlans(exec: Exec, slotId: string, excludePlanI return [...byId.values()]; } +/** + * 同作品的手动下载计划(slot_id 为空)里仍在途的(M6):订阅评估把它们并入在途单元,手动下载中的集不再被订阅重复下。 + * 手动计划没记单元,按资源解析补:单集 / 多集 → 集键;季包 → 整季通配;电影 → 空(电影收录版本有在途即忙)。 + */ +export async function manualInflightPlans(exec: Exec, entityId: string): Promise { + const rows = await exec + .selectFrom("acquisition_plan as p") + .innerJoin("source_release as r", "r.id", "p.source_release_id") + .leftJoin("download_task as dt", (j) => + j.onRef("dt.acquisition_plan_id", "=", "p.id").on("dt.removed_at", "is", null), + ) + .selectAll("r") + .select(["p.id as plan_id", "p.status as plan_status", "p.unit_keys as plan_unit_keys", "dt.status as dt_status"]) + .where("p.slot_id", "is", null) + .where("p.status", "in", [...INFLIGHT_PLAN_STATUSES]) + .where((eb) => + eb.or([eb("r.media_entity_id", "=", entityId), eb(sql`p.evaluation->>'mediaEntityId'`, "=", entityId)]), + ) + .execute(); + const byId = new Map(); + for (const r of rows) { + let keys = r.plan_unit_keys; + if (!keys.length) { + const parsed = parsedOf(r); + keys = unitKeysOf(parsed); + if (!keys.length && parsed.season !== undefined) keys = [seasonWildcard(parsed.season)]; + } + const stage: InflightPlan["stage"] = + r.plan_status === "reserved" + ? "reserved" + : r.dt_status === "completed" || r.dt_status === "importing" + ? "verifying" + : "downloading"; + byId.set(r.plan_id, { id: r.plan_id, status: r.plan_status, unitKeys: keys, stage }); + } + return [...byId.values()]; +} + // ── 完成判定 ── export interface SlotCompletion { + /** 覆盖齐(Slot 静止态按它落 satisfied / empty) */ complete: boolean; + /** + * 有升级目标(规则组 / 阶梯按 pri 洗版)时,范围内每个已覆盖单元(电影 = 片子)都已到目标 pri; + * 没有升级目标恒为 true。订阅完成要求 complete && atTarget(H2)。 + */ + atTarget: boolean; seasons: SeasonProgress[]; /** 继续追踪时的疑似缺集(总集数未知,已见最高集以下的空洞) */ suspectedMissing: { season: number; episodes: number[] }[]; } +/** 订阅类型 → 规则组 scope 的媒体类型 */ +const ruleMediaKind = (k: string | null | undefined): "movie" | "series" | "variety" => + k === "movie" ? "movie" : k === "variety" ? "variety" : "series"; + +/** + * 收录版本的评估口径:行 → SlotSpec,并入规则组链(收录版本 ruleGroupIds → 订阅 ruleGroupIds → 订阅默认链; + * 开升级叠加洗版默认链;订阅私有组的包含 / 排除并入硬条件)。siteId 用于规则组的站点范围,不给则不按站点过滤。 + */ +export function slotSpecWithRules( + cat: RuleCatalog, + slot: AcquisitionSlotRow, + sub: Pick | null | undefined, + entity: { id: string; kind: string }, + siteId?: string, +): SlotSpec { + const spec = slotSpecFromRow(toApi(slot)); + const chain = resolveRuleChain(cat, { + kind: "acquisition", + slotRuleGroupIds: (slot.rule_group_ids ?? []) as string[], + subRuleGroupIds: subParams(sub).ruleGroupIds, + upgrade: !!spec.upgrade?.enabled, + mediaKind: ruleMediaKind(entity.kind), + entityId: entity.id, + siteId, + subscriptionId: sub?.id, + }); + spec.hard = mergeHard(spec.hard, chain.hard); + if (chain.groups.length) spec.rules = { groups: chain.groups, atoms: chain.atoms }; + return spec; +} + +/** + * 洗版目标 pri:开了升级且按 pri 口径比较(规则组 / 阶梯有层,或显式 basis=priority)时为 targetPri,缺省 100 + * (同 evaluateCandidate 的 at_ceiling);按来源等级比较的升级没有「最高档」,返回 null(覆盖齐即完成)。 + */ +export function upgradeTargetPri(spec: SlotSpec): number | null { + const up = spec.upgrade; + if (!up?.enabled) return null; + const tiered = !!spec.soft?.tiers?.length || !!spec.rules?.groups.some((g) => g.layers.length > 0); + return (up.basis ?? (tiered ? "priority" : "tier")) === "priority" ? (up.targetPri ?? 100) : null; +} + +/** 订阅是否「覆盖齐即完成」:一次性订阅,或仅本季(含电影);接续后续季的持续订阅不完成,等待新季 */ +export const completesOnCoverage = (sub: Pick): boolean => + sub.mode === "once" || sub.season_scope === "current_season"; + +/** 收录版本是否算完成(订阅完成用):覆盖齐且到洗版目标 */ +export const slotDone = (c: Pick): boolean => c.complete && c.atTarget; + +interface PriVersion { + /** 覆盖的单元:季 → 集号集合;null = 整季(季包通配) */ + units: Map | null>; + pri: number; +} + +/** + * 洗版到顶判定(H2):收集该收录版本的现有版本(资产 + 已接收计划),按当前规则组重算每个版本的 pri + * (有来源资源按其标题 / 站点重算,同评估时 rulePriOfAsset;否则取计划评估时记下的 pri;再否则按交付特征算), + * 电影任一版本到目标即到顶;剧集要求范围内每个已覆盖单元都有到目标的版本。 + */ +async function upgradeReached( + exec: Exec, + input: { + slot: AcquisitionSlotRow; + sub: SubscriptionRow | null; + mediaType: "movie" | "tv"; + target: number; + cat: RuleCatalog; + newAssets: readonly string[]; + acceptedPlanIds: readonly string[]; + entityWide: boolean; + /** 剧集:范围内已覆盖的单元(季 → 集号) */ + covered: Map; + }, +): Promise { + const { slot, sub, cat } = input; + const entity = { + id: slot.media_entity_id, + kind: sub?.media_kind ?? (input.mediaType === "movie" ? "movie" : "series"), + }; + const lang = await entityOriginalLanguage(exec, slot.media_entity_id); + const relPri = (rel: SourceReleaseRow) => { + const spec = slotSpecWithRules(cat, slot, sub, entity, rel.site_id); + const parsed = parsedOf(rel); + const ctx = { ...contextOf(rel), ...(lang ? { originalLanguage: lang } : {}) }; + return effectiveLadder(spec, parsed, ctx, ruleChainOf(spec, parsed, ctx)).pri; + }; + const baseSpec = slotSpecWithRules(cat, slot, sub, entity); + const versions: PriVersion[] = []; + const assetIds = [...new Set([...input.newAssets, ...(slot.current_asset_id ? [slot.current_asset_id] : [])])]; + const assets = await exec + .selectFrom("media_asset as a") + .leftJoin("source_release as r", "r.id", "a.source_release_id") + .leftJoin("download_task as dt", "dt.id", "a.download_task_id") + .leftJoin("acquisition_plan as p", "p.id", "dt.acquisition_plan_id") + .select(["a.id", "a.media_info", "a.source_release_id", "p.evaluation as plan_evaluation"]) + .where("a.status", "=", "active") + .where("a.parent_asset_id", "is", null) + .where((eb) => { + const ors = [ + eb("p.slot_id", "=", slot.id), + eb( + "a.id", + "in", + eb + .selectFrom("slot_unit") + .select("media_asset_id") + .where("slot_id", "=", slot.id) + .where("media_asset_id", "is not", null) + .$castTo(), + ), + ]; + if (assetIds.length) ors.push(eb("a.id", "in", assetIds)); + if (input.entityWide && input.mediaType === "tv") ors.push(eb("a.media_entity_id", "=", slot.media_entity_id)); + return eb.or(ors); + }) + .execute(); + const relIds = [...new Set(assets.map((x) => x.source_release_id).filter((x): x is string => !!x))]; + const plans = input.acceptedPlanIds.length + ? await exec + .selectFrom("acquisition_plan") + .select(["id", "unit_keys", "source_release_id", "evaluation"]) + .where("id", "in", [...input.acceptedPlanIds]) + .execute() + : []; + for (const p of plans) if (p.source_release_id) relIds.push(p.source_release_id); + const rels = new Map( + (relIds.length + ? await exec + .selectFrom("source_release") + .selectAll() + .where("id", "in", [...new Set(relIds)]) + .execute() + : [] + ).map((r) => [r.id, r]), + ); + const evalPri = (e: unknown) => { + const v = (e as { pri?: unknown } | null)?.pri; + return typeof v === "number" ? v : undefined; + }; + const cov = assets.length + ? await exec + .selectFrom("asset_unit_coverage as c") + .innerJoin("content_unit as u", "u.id", "c.content_unit_id") + .select(["c.media_asset_id", "u.season_number", "u.episode_number"]) + .where( + "c.media_asset_id", + "in", + assets.map((x) => x.id), + ) + .where("u.kind", "=", "episode") + .execute() + : []; + for (const x of assets) { + const rel = x.source_release_id ? rels.get(x.source_release_id) : undefined; + const pri = + (rel ? relPri(rel) : undefined) ?? + evalPri(x.plan_evaluation) ?? + currentPri( + baseSpec, + ((x.media_info as { _umt?: { quality?: QualityFacts } })?._umt?.quality ?? {}) as QualityFacts, + ); + const units = new Map | null>(); + for (const c of cov) + if (c.media_asset_id === x.id && c.episode_number !== null) { + const s = c.season_number ?? 1; + const set = units.get(s) ?? new Set(); + set.add(c.episode_number); + units.set(s, set); + } + versions.push({ units, pri }); + } + for (const p of plans) { + const rel = p.source_release_id ? rels.get(p.source_release_id) : undefined; + const pri = (rel ? relPri(rel) : undefined) ?? evalPri(p.evaluation); + if (pri === undefined) continue; + const units = new Map | null>(); + for (const k of p.unit_keys) { + const u = parseUnitKey(k); + if (!u) continue; + if (u.wildcard) units.set(u.season, null); + else if (u.episode !== undefined && units.get(u.season) !== null) + units.set(u.season, (units.get(u.season) ?? new Set()).add(u.episode)); + } + versions.push({ units, pri }); + } + if (input.mediaType === "movie") return versions.some((v) => v.pri >= input.target); + for (const [season, eps] of input.covered) + for (const e of eps) { + const best = Math.max( + -1, + ...versions + .filter((v) => v.units.has(season) && (v.units.get(season) === null || v.units.get(season)!.has(e))) + .map((v) => v.pri), + ); + if (best < input.target) return false; + } + return true; +} + /** 剧集 Slot 覆盖的季:Slot 限定季单元 > 订阅季单元 > 作品已知的所有正片季 */ async function scopeSeasons( exec: Exec, @@ -314,10 +589,7 @@ async function scopeSeasons( */ export async function slotComplete( exec: Exec, - slot: Pick< - AcquisitionSlotRow, - "id" | "media_entity_id" | "content_unit_id" | "completion" | "subscription_id" | "current_asset_id" - >, + slot: AcquisitionSlotRow, mediaType: "movie" | "tv", newAssets: readonly string[], opts: { includeSubmitted?: boolean } = {}, @@ -333,6 +605,32 @@ export async function slotComplete( .where("p.status_reason", "!=", "accepted_unconfirmed") .execute() : []; + const sub = slot.subscription_id + ? ((await exec.selectFrom("subscription").selectAll().where("id", "=", slot.subscription_id).executeTakeFirst()) ?? + null) + : null; + // 洗版目标(H2):有目标时另判是否已到顶;没有目标 atTarget 恒为 true + const cat = await loadRuleCatalog(exec); + const target = upgradeTargetPri( + slotSpecWithRules(cat, slot, sub, { + id: slot.media_entity_id, + kind: sub?.media_kind ?? (mediaType === "movie" ? "movie" : "series"), + }), + ); + const reached = (covered: Map, entityWide: boolean) => + target === null + ? Promise.resolve(true) + : upgradeReached(exec, { + slot, + sub, + mediaType, + target, + cat, + newAssets, + acceptedPlanIds: accepted.map((p) => p.plan_id), + entityWide, + covered, + }); if (mediaType === "movie") { const assets = newAssets.length ? await exec @@ -355,19 +653,16 @@ export async function slotComplete( : {}), })), ]).complete; - return { complete, seasons: [], suspectedMissing: [] }; + return { complete, atTarget: complete && (await reached(new Map(), false)), seasons: [], suspectedMissing: [] }; } - const sub = slot.subscription_id - ? ((await exec.selectFrom("subscription").selectAll().where("id", "=", slot.subscription_id).executeTakeFirst()) ?? - null) - : null; const slotCount = await exec .selectFrom("acquisition_slot") .select((eb) => eb.fn.countAll().as("n")) .where("media_entity_id", "=", slot.media_entity_id) .executeTakeFirstOrThrow(); // 单 Slot 作品:作品下所有资产都算该 Slot 的覆盖(手动整理/历史导入);多 Slot 只算本 Slot 的 - const covered = await slotCoveredUnits(exec, slot, { entityWide: Number(slotCount.n) <= 1 }); + const entityWide = Number(slotCount.n) <= 1; + const covered = await slotCoveredUnits(exec, slot, { entityWide }); const inflight = await slotInflightPlans(exec, slot.id); const inflightEps = new Map(); for (const p of inflight) @@ -396,8 +691,13 @@ export async function slotComplete( }); }); const complete = sub?.status !== "pending" && progress.length > 0 && progress.every((p) => p.complete); + // 到顶只看范围内(总集数以内)的已覆盖单元 + const inScope = new Map( + progress.map((p) => [p.season, p.total !== null ? p.covered.filter((e) => e <= p.total!) : p.covered] as const), + ); return { complete, + atTarget: complete && (await reached(inScope, entityWide)), seasons: progress, suspectedMissing: progress .filter((p) => p.suspectedMissing.length > 0) @@ -506,14 +806,13 @@ export async function afterSlotVerified( }); } - if (input.completion.complete && slot.subscription_id) { + if (slotDone(input.completion) && slot.subscription_id) { const sub = await trx .selectFrom("subscription") .selectAll() .where("id", "=", slot.subscription_id) .executeTakeFirst(); - const seasonDone = sub?.media_kind !== "movie" && sub?.season_scope === "current_season"; - if (sub && sub.status === "active" && (sub.mode === "once" || seasonDone)) { + if (sub && sub.status === "active" && completesOnCoverage(sub)) { const siblings = await trx .selectFrom("acquisition_slot") .selectAll() @@ -531,7 +830,7 @@ export async function afterSlotVerified( ), ), ); - if (others.every((p) => p.complete)) + if (others.every(slotDone)) await completeSubscription(d, trx, sub, "slot_satisfied", { seasons: input.completion.seasons }); } } @@ -961,7 +1260,7 @@ export async function afterDownloadAccepted(d: AoDeps, planId: string): Promise< .executeTakeFirst(); if (sub?.status !== "active") return; await touchProgress(trx, { subscriptionId: sub.id }, d.ports.now()); - if (sub.mode !== "once" && sub.season_scope !== "current_season") return; + if (!completesOnCoverage(sub)) return; const slots = await trx.selectFrom("acquisition_slot").selectAll().where("subscription_id", "=", sub.id).execute(); if (!slots.length) return; const progress = await Promise.all( @@ -975,7 +1274,7 @@ export async function afterDownloadAccepted(d: AoDeps, planId: string): Promise< ), ), ); - if (progress.every((p) => p.complete)) + if (progress.every(slotDone)) await completeSubscription(d, trx, sub, "download_accepted", { slots: progress, seasons: progress.flatMap((p) => p.seasons), diff --git a/apps/server/test/acquisition-organize/upgrade-completion.test.ts b/apps/server/test/acquisition-organize/upgrade-completion.test.ts new file mode 100644 index 00000000..119cb9e3 --- /dev/null +++ b/apps/server/test/acquisition-organize/upgrade-completion.test.ts @@ -0,0 +1,155 @@ +import { createMockQb, createTempTree, type MockQb, makeTorrent, type TempTree } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts"; +import { submitDownload, syncDownloads } from "../../src/modules/acquisition-organize/services/downloads.ts"; +import { executeRecord, planImport } from "../../src/modules/acquisition-organize/services/organize.ts"; +import { applyEpisodeTotal } from "../../src/modules/acquisition-organize/services/subscriptions.ts"; +import { createHarness, fakeDisk, type Harness, pgAvailable, seedBase, seedEntity, seedRelease } from "./harness.ts"; + +// 洗版完成口径(H2):收录版本有升级目标(规则组阶梯按 pri 洗版)时,覆盖齐还不够,要到最高档才完成订阅。 +// 电影洗版、按集洗版、全集洗版各一例;下载器接收即计入(afterDownloadAccepted)与入库后(afterSlotVerified)同口径。 + +const hasPg = await pgAvailable(); +const SIZE = 300_000; + +describe.skipIf(!hasPg)("洗版到最高档才完成(真实 PG + mock qB + 临时目录)", () => { + let h: Harness; + let qb: MockQb; + let lib: TempTree; + let cache: TempTree; + const disk = fakeDisk(10_000_000_000); + const torrents = new Map(); + let base: Awaited>; + let groupId: string; + + beforeAll(async () => { + qb = await createMockQb({ addVisibleAfterPolls: 0 }); + lib = await createTempTree({}, { sentinel: true, prefix: "umt-lib-" }); + cache = await createTempTree({}, { prefix: "umt-cache-" }); + disk.state.watch = [cache.root]; + disk.setPct(10); + h = await createHarness("upgradedone", { + diskUsage: disk.fn, + torrentSource: async ({ release }) => ({ kind: "file", bytes: torrents.get(release.id)! }), + }); + base = await seedBase(h.db, { libraryPath: lib.root, cachePath: cache.root, qbUrl: qb.url }); + const g = await h.api("POST", "/rules", { name: "洗版到 REMUX", ruleString: "REMUX > WEBDL" }); + expect(g.status, JSON.stringify(g.body)).toBe(201); + groupId = g.body.data.id; + }, 60_000); + + afterAll(async () => { + await h?.close(); + await qb?.close(); + await lib?.cleanup(); + await cache?.cleanup(); + }); + + const db = () => h.db; + + async function washSub(kind: "movie" | "series", title: string, mode: "movie" | "per_unit" | "full_pack") { + const entityId = await seedEntity(db(), { kind, title, year: 2024 }); + const sub = await db() + .insertInto("subscription") + .values({ media_entity_id: entityId, media_kind: kind, mode: "once", season_scope: "current_season" }) + .returning("id") + .executeTakeFirstOrThrow(); + await db() + .insertInto("acquisition_slot") + .values({ + media_entity_id: entityId, + subscription_id: sub.id, + name: "洗版", + upgrade_threshold: JSON.stringify({ enabled: true, mode }), + rule_group_ids: JSON.stringify([groupId]), + }) + .execute(); + return { entityId, subId: sub.id }; + } + + async function release(entityId: string, title: string, files?: { path: string; length: number }[]) { + const t = files + ? makeTorrent({ name: title, files, pieceSeed: Math.floor(Math.random() * 1e6) }) + : makeTorrent({ name: `${title}.mkv`, length: SIZE, pieceSeed: Math.floor(Math.random() * 1e6) }); + const id = await seedRelease(db(), { siteId: base.siteId, entityId, title, size: SIZE, hash: t.hash }); + torrents.set(id, t.bytes); + return { id, hash: t.hash }; + } + const subRow = (id: string) => + db().selectFrom("subscription").selectAll().where("id", "=", id).executeTakeFirstOrThrow(); + const planRow = (id: string) => + db().selectFrom("acquisition_plan").selectAll().where("id", "=", id).executeTakeFirstOrThrow(); + + async function planFor(relId: string) { + const ev = await evaluateAcquisition(h.d, { sourceReleaseId: relId }); + expect(ev.plans, JSON.stringify(ev)).toHaveLength(1); + return ev.plans[0]!; + } + async function submit(planId: string) { + const s = await submitDownload(h.d, planId); + expect(s.status).toBe("submitted"); + return s.downloadTaskId!; + } + async function completeAndImport(hash: string, dtId: string) { + await qb.complete(hash, { writeFiles: true }); + const sync = await syncDownloads(h.d); + expect(sync.completed).toContain(dtId); + const p = await planImport(h.d, { downloadTaskId: dtId, trigger: "auto", actorType: "system" }); + expect(p.status, JSON.stringify(p)).toBe("running"); + const ex = await executeRecord(h.d, p.recordId!); + expect(ex.status).toBe("succeeded"); + } + + it("电影洗版:WEB-DL(pri 99)接收、入库都不完成;REMUX(pri 100)接收即完成", async () => { + const { entityId, subId } = await washSub("movie", "Wash Film", "movie"); + const web = await release(entityId, "Wash.Film.2024.1080p.WEB-DL.x264-GRP"); + const webTask = await submit(await planFor(web.id)); + expect(await subRow(subId)).toMatchObject({ status: "active" }); + await completeAndImport(web.hash, webTask); + expect(await subRow(subId)).toMatchObject({ status: "active" }); + const remux = await release(entityId, "Wash.Film.2024.1080p.BluRay.REMUX.AVC-GRP"); + const up = await planFor(remux.id); + expect(await planRow(up)).toMatchObject({ purpose: "upgrade" }); + await submit(up); + expect(await subRow(subId)).toMatchObject({ status: "completed", status_reason: "download_accepted" }); + }, 60_000); + + it("按集洗版:全集已覆盖但不是最高档时不完成;逐集洗到 REMUX 后完成", async () => { + const { entityId, subId } = await washSub("series", "Wash Episodes", "per_unit"); + await db() + .transaction() + .execute((trx) => applyEpisodeTotal(trx, { entityId, season: 1, fresh: 1, now: new Date() })); + const web = await release(entityId, "Wash.Episodes.S01E01.1080p.WEB-DL.x264-GRP"); + const webTask = await submit(await planFor(web.id)); + expect(await subRow(subId)).toMatchObject({ status: "active" }); + await completeAndImport(web.hash, webTask); + expect(await subRow(subId)).toMatchObject({ status: "active" }); + const remux = await release(entityId, "Wash.Episodes.S01E01.1080p.BluRay.REMUX.AVC-GRP"); + const up = await planFor(remux.id); + expect(await planRow(up)).toMatchObject({ purpose: "upgrade", unit_keys: ["s1e1"] }); + await submit(up); + expect(await subRow(subId)).toMatchObject({ status: "completed", status_reason: "download_accepted" }); + }, 60_000); + + it("全集洗版:WEB-DL 季包覆盖全季不完成;单集 REMUX 不接;REMUX 季包后完成", async () => { + const { entityId, subId } = await washSub("series", "Wash Pack", "full_pack"); + await db() + .transaction() + .execute((trx) => applyEpisodeTotal(trx, { entityId, season: 1, fresh: 2, now: new Date() })); + const files = (tag: string) => [1, 2].map((e) => ({ path: `Wash.Pack.S01E0${e}.1080p.${tag}.mkv`, length: SIZE })); + const web = await release(entityId, "Wash.Pack.S01.1080p.WEB-DL.x264-GRP", files("WEB-DL.x264-GRP")); + const webPlan = await planFor(web.id); + expect((await planRow(webPlan)).unit_keys).toEqual(["s1e1", "s1e2"]); + const webTask = await submit(webPlan); + expect(await subRow(subId)).toMatchObject({ status: "active" }); + await completeAndImport(web.hash, webTask); + expect(await subRow(subId)).toMatchObject({ status: "active" }); + const single = await release(entityId, "Wash.Pack.S01E01.1080p.BluRay.REMUX.AVC-GRP"); + expect((await evaluateAcquisition(h.d, { sourceReleaseId: single.id })).plans).toHaveLength(0); + const remux = await release(entityId, "Wash.Pack.S01.1080p.BluRay.REMUX.AVC-GRP", files("BluRay.REMUX.AVC-GRP")); + const up = await planFor(remux.id); + expect(await planRow(up)).toMatchObject({ purpose: "upgrade" }); + await submit(up); + expect(await subRow(subId)).toMatchObject({ status: "completed", status_reason: "download_accepted" }); + }, 60_000); +}); diff --git a/apps/server/test/subscriptions/subscription-scope-reevaluate.test.ts b/apps/server/test/subscriptions/subscription-scope-reevaluate.test.ts new file mode 100644 index 00000000..65a8f396 --- /dev/null +++ b/apps/server/test/subscriptions/subscription-scope-reevaluate.test.ts @@ -0,0 +1,222 @@ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts"; +import { ensureSeasonUnit } from "../../src/modules/acquisition-organize/services/subscriptions.ts"; +import { + createHarness, + fakeDisk, + type Harness, + pgAvailable, + seedBase, + seedEntity, + seedRelease, +} from "../acquisition-organize/harness.ts"; + +// 获取正确性(T3):订阅总有自己的收录版本(H1)、接续后续季的季下界(H4)、重评择优(M4)、 +// 重评不止最近 50 条(M5)、手动下载中的集计入在途(M6)。只到建计划为止(假引擎,不提交下载器)。 + +const hasPg = await pgAvailable(); +const SIZE = 300_000; + +describe.skipIf(!hasPg)("订阅范围与重评(真实 PG)", () => { + let h: Harness; + const disk = fakeDisk(10_000_000_000); + let base: Awaited>; + + beforeAll(async () => { + disk.setPct(10); + h = await createHarness("subscope", { diskUsage: disk.fn }); + base = await seedBase(h.db, { + libraryPath: "/tmp/nonexistent-lib", + cachePath: "/tmp", + qbUrl: "http://127.0.0.1:9", + }); + // 计划不提交,预留会一直占并发名额:放宽并发上限 + await h.setSettings("budget", { maxConcurrentDownloads: 100 }); + }, 60_000); + afterAll(async () => { + await h?.close(); + }); + + const db = () => h.db; + let seq = 0; + const hash = () => (++seq).toString(16).padStart(40, "a"); + async function release(entityId: string, title: string, firstSeen?: Date) { + const id = await seedRelease(db(), { siteId: base.siteId, entityId, title, size: SIZE, hash: hash() }); + if (firstSeen) + await db().updateTable("source_release").set({ first_seen_at: firstSeen }).where("id", "=", id).execute(); + return id; + } + const slotsOf = (subId: string) => + db().selectFrom("acquisition_slot").selectAll().where("subscription_id", "=", subId).orderBy("position").execute(); + const planRow = (id: string) => + db().selectFrom("acquisition_plan").selectAll().where("id", "=", id).executeTakeFirstOrThrow(); + async function seasons(entityId: string, n: number) { + const ids: string[] = []; + for (let s = 1; s <= n; s++) + ids.push( + await db() + .transaction() + .execute((trx) => ensureSeasonUnit(trx, entityId, s)), + ); + return ids; + } + async function createSub(body: Record) { + const r = await h.api("POST", "/subscriptions", body); + expect(r.status, JSON.stringify(r.body)).toBe(201); + return r.body.data.id as string; + } + + it("H1:无策略模板、没填质量预设的订阅也有自己的默认收录版本,能下载", async () => { + const e = await seedEntity(db(), { kind: "series", title: "无预设剧", year: 2025 }); + const subId = await createSub({ mediaEntityId: e, mediaKind: "series" }); + const slots = await slotsOf(subId); + expect(slots).toHaveLength(1); + expect(slots[0]).toMatchObject({ name: "默认版本", hard_conditions: {} }); + const r = await release(e, "No.Preset.S01E01.1080p.WEB-DL.x264-GRP"); + const ev = await evaluateAcquisition(h.d, { sourceReleaseId: r }); + expect(ev.plans, JSON.stringify(ev)).toHaveLength(1); + expect((await planRow(ev.plans[0]!)).slot_id).toBe(slots[0]!.id); + }); + + it("H1:S1 旧订阅已有收录版本时再订 S2,S2 订阅建自己的收录版本(重名加序号)并能下载", async () => { + const e = await seedEntity(db(), { kind: "series", title: "两季剧", year: 2024 }); + const [s1, s2] = await seasons(e, 2); + const sub1 = await createSub({ + mediaEntityId: e, + mediaKind: "series", + seasonUnitId: s1, + seasonScope: "current_season", + params: { resolution: "1080p" }, + }); + expect(await slotsOf(sub1)).toHaveLength(1); + await db().updateTable("subscription").set({ status: "completed" }).where("id", "=", sub1).execute(); + const sub2 = await createSub({ + mediaEntityId: e, + mediaKind: "series", + seasonUnitId: s2, + seasonScope: "current_season", + params: { resolution: "1080p" }, + }); + const own = await slotsOf(sub2); + expect(own).toHaveLength(1); + expect(own[0]!.name).toBe("默认版本 2"); + // 质量预设写进了 S2 订阅自己的收录版本 + expect(own[0]!.hard_conditions).toMatchObject({ resolutions: ["1080p", "1080i"] }); + const r = await release(e, "Two.Seasons.S02E01.1080p.WEB-DL.x264-GRP"); + const ev = await evaluateAcquisition(h.d, { sourceReleaseId: r }); + expect(ev.plans, JSON.stringify(ev)).toHaveLength(1); + expect((await planRow(ev.plans[0]!)).slot_id).toBe(own[0]!.id); + }); + + it("H1:已有的 0 收录版本订阅首次评估时补建;编辑时也补建", async () => { + const e = await seedEntity(db(), { kind: "movie", title: "Legacy Film", year: 2023 }); + const sub = await db() + .insertInto("subscription") + .values({ media_entity_id: e, media_kind: "movie", mode: "once", season_scope: "current_season" }) + .returning("id") + .executeTakeFirstOrThrow(); + expect(await slotsOf(sub.id)).toHaveLength(0); + const r = await release(e, "Legacy.Film.2023.1080p.WEB-DL.x264-GRP"); + const ev = await evaluateAcquisition(h.d, { sourceReleaseId: r }); + expect(ev.plans, JSON.stringify(ev)).toHaveLength(1); + expect(await slotsOf(sub.id)).toHaveLength(1); + + const e2 = await seedEntity(db(), { kind: "movie", title: "Legacy Film 2", year: 2023 }); + const sub2 = await db() + .insertInto("subscription") + .values({ media_entity_id: e2, media_kind: "movie", mode: "once", season_scope: "current_season" }) + .returning("id") + .executeTakeFirstOrThrow(); + const p = await h.api("PATCH", `/subscriptions/${sub2.id}`, { params: { resolution: "2160p" } }); + expect(p.status, JSON.stringify(p.body)).toBe(200); + const [slot] = await slotsOf(sub2.id); + expect(slot?.hard_conditions).toMatchObject({ resolutions: ["2160p"] }); + }); + + it("H4:从 S3 订「接续后续季」不下 S1 / S2,S3 及以后照常", async () => { + const e = await seedEntity(db(), { kind: "series", title: "三季剧", year: 2020 }); + const [, , s3] = await seasons(e, 4); + const subId = await createSub({ + mediaEntityId: e, + mediaKind: "series", + seasonUnitId: s3, + seasonScope: "following_seasons", + params: { resolution: "1080p" }, + }); + const planned = async (title: string) => + (await evaluateAcquisition(h.d, { sourceReleaseId: await release(e, title) })).plans.length; + expect(await planned("Three.Seasons.S01E01.1080p.WEB-DL.x264-GRP")).toBe(0); + expect(await planned("Three.Seasons.S02E05.1080p.WEB-DL.x264-GRP")).toBe(0); + expect(await planned("Three.Seasons.S02.1080p.WEB-DL.x264-GRP")).toBe(0); + expect(await planned("Three.Seasons.S03E01.1080p.WEB-DL.x264-GRP")).toBe(1); + expect(await planned("Three.Seasons.S04E01.1080p.WEB-DL.x264-GRP")).toBe(1); + // 订阅重评同口径 + const re = await evaluateAcquisition(h.d, { subscriptionId: subId }); + expect(re.plans).toHaveLength(0); + }); + + it("M4:订阅重评时同一集多个合格资源选 pri 最高的,不是最新的", async () => { + const e = await seedEntity(db(), { kind: "series", title: "择优剧", year: 2024 }); + const g = await h.api("POST", "/rules", { name: "择优阶梯", ruleString: "REMUX > WEBDL" }); + expect(g.status, JSON.stringify(g.body)).toBe(201); + const t0 = Date.now(); + const remux = await release(e, "Pick.Best.S01E01.1080p.BluRay.REMUX.AVC-GRP", new Date(t0 - 3_600_000)); + await release(e, "Pick.Best.S01E01.1080p.WEB-DL.x264-NEW", new Date(t0 - 60_000)); + // 订阅建好后 kickEvaluate 只入队(假引擎),这里直接跑订阅重评 + const subId = await createSub({ + mediaEntityId: e, + mediaKind: "series", + params: { filter_groups: [g.body.data.id], resolution: "1080p" }, + }); + const ev = await evaluateAcquisition(h.d, { subscriptionId: subId }); + expect(ev.plans, JSON.stringify(ev)).toHaveLength(1); + expect(await planRow(ev.plans[0]!)).toMatchObject({ source_release_id: remux, unit_keys: ["s1e1"] }); + expect(ev.deferred.map((x) => x.reason)).toEqual(["units_inflight"]); + }); + + it("M5:超过 50 条较新的资源之后,更早的集仍能被重评选中", async () => { + const e = await seedEntity(db(), { kind: "series", title: "长剧", year: 2018 }); + const subId = await createSub({ mediaEntityId: e, mediaKind: "series", params: { resolution: "1080p" } }); + const t0 = Date.now(); + const old = await release(e, "Long.Show.S01E01.1080p.WEB-DL.x264-GRP", new Date(t0 - 86_400_000)); + // 55 条更新的 720p 资源(不合格)把最近 50 条占满 + for (let i = 2; i <= 56; i++) + await release(e, `Long.Show.S01E${String(i).padStart(2, "0")}.720p.WEB-DL.x264-GRP`, new Date(t0 - i * 1000)); + const ev = await evaluateAcquisition(h.d, { subscriptionId: subId }); + expect(ev.plans, JSON.stringify(ev.skipped.slice(0, 3))).toHaveLength(1); + expect(await planRow(ev.plans[0]!)).toMatchObject({ source_release_id: old, unit_keys: ["s1e1"] }); + }, 60_000); + + it("M6:手动下载中的集,订阅不再用别的资源重复下;没在手动下载的集照常", async () => { + const e = await seedEntity(db(), { kind: "series", title: "手动剧", year: 2024 }); + const subId = await createSub({ mediaEntityId: e, mediaKind: "series", params: { resolution: "1080p" } }); + const [slot] = await slotsOf(subId); + const manualRel = await release(e, "Manual.Show.S01E01.1080p.WEB-DL.x264-USER"); + await db() + .insertInto("acquisition_plan") + .values({ + slot_id: null, + source_release_id: manualRel, + status: "submitted", + priority: "immediate", + evaluation: JSON.stringify({ manual: true, mediaEntityId: e }), + payload_hash: "manual-test", + idempotency_key: `manual:test:${manualRel}`, + status_reason: "manual_selection", + }) + .execute(); + const other = await release(e, "Manual.Show.S01E01.1080p.WEB-DL.x265-OTHER"); + const ev = await evaluateAcquisition(h.d, { sourceReleaseId: other }); + expect(ev.plans).toHaveLength(0); + expect(ev.deferred).toEqual([{ sourceReleaseId: other, slotId: slot!.id, reason: "units_inflight" }]); + // 手动计划结束不会触发重评:延后时排了退避重评 + const s = await db() + .selectFrom("acquisition_slot") + .select("next_evaluate_at") + .where("id", "=", slot!.id) + .executeTakeFirstOrThrow(); + expect(s.next_evaluate_at).not.toBeNull(); + const e2 = await release(e, "Manual.Show.S01E02.1080p.WEB-DL.x264-OTHER"); + expect((await evaluateAcquisition(h.d, { sourceReleaseId: e2 })).plans).toHaveLength(1); + }); +});