pixel_pop/UHD_Media_Tool
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(acquisition): 订阅总带自己的收录版本、洗版到最高档才完成、接续后续季加季下界、重评择优并放宽范围、手动下载计入在途

Browse Source 
- 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 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
This commit is contained in:
rootandClaude Opus 5.5 committed 2026-10-03 23:22:16 +08:00
1 parent 0085e45ca7
commit fcd0d8f223
6 files changed
+1000 -185
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

apps/server/src/modules/acquisition-organize/routes/subscriptions.ts
+5 -32
View File
@@ -2,7 +2,6 @@ import {
type AcquisitionSlot, type AcquisitionSlot,
Api, Api,
DEFAULT_SEARCH_INTERVAL_HOURS, DEFAULT_SEARCH_INTERVAL_HOURS,
type SlotTemplate,
type Subscription, type Subscription,
type SubscriptionGroup, type SubscriptionGroup,
type SubscriptionHistory, type SubscriptionHistory,
@@ -16,7 +15,7 @@ import type { Context } from "hono";
import type { AuthEnv } from "../../../http/auth.ts"; import type { AuthEnv } from "../../../http/auth.ts";
import { ApiException } from "../../../http/errors.ts"; import { ApiException } from "../../../http/errors.ts";
import type { ModuleApp } from "../../types.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 type { AoDeps, Exec } from "../services/common.ts";
import { import {
buildSubscriptionModels, buildSubscriptionModels,
@@ -107,9 +106,6 @@ async function assertNoDuplicate(
if (hit) conflict(`该作品同一季已有订阅：${hit.id}`); if (hit) conflict(`该作品同一季已有订阅：${hit.id}`);
} }
/** 没有收录策略模板、表单给了质量预设时,新订阅按这个模板建一个收录版本(预设随后写入) */
const DEFAULT_SLOT_TEMPLATE: SlotTemplate = { name: "默认版本" };
/** /**
* 保存后应用表单的派生写入:包含 / 排除 → 订阅私有规则组;质量预设(MP resolution / quality / effect)→ * 保存后应用表单的派生写入:包含 / 排除 → 订阅私有规则组;质量预设(MP resolution / quality / effect)→
* 该订阅自己的收录版本;MP total_episode → 季总集数;best_version → 收录版本升级策略。 * 该订阅自己的收录版本;MP total_episode → 季总集数;best_version → 收录版本升级策略。
@@ -376,33 +372,8 @@ export function registerSubscriptionRoutes(app: ModuleApp, d: AoDeps) {
}) })
.returningAll() .returningAll()
.executeTakeFirstOrThrow(); .executeTakeFirstOrThrow();
if (r.media_entity_id) { // 订阅总有自己的收录版本:策略模板,没有就建一个默认版本(质量预设随后由 applyDerived 写入)
const has = await trx await ensureSubscriptionSlot(trx, r);
.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,
});
}
}
await applyDerived(trx, r, norm, now); await applyDerived(trx, r, norm, now);
return r; return r;
}); });
@@ -458,6 +429,8 @@ export function registerSubscriptionRoutes(app: ModuleApp, d: AoDeps) {
.where("id", "=", params.id) .where("id", "=", params.id)
.returningAll() .returningAll()
.executeTakeFirstOrThrow(); .executeTakeFirstOrThrow();
// 旧订阅没有自己的收录版本:编辑时补建(质量预设 / 升级策略才有地方写)
if (r.status !== "cancelled") await ensureSubscriptionSlot(trx, r);
if (norm) await applyDerived(trx, r, norm, d.ports.now(), subParams(cur)); if (norm) await applyDerived(trx, r, norm, d.ports.now(), subParams(cur));
else if (body.status !== undefined) await syncPrivateRuleGroup(trx, r, d.ports.now()); else if (body.status !== undefined) await syncPrivateRuleGroup(trx, r, d.ports.now());
return r; return r;
apps/server/src/modules/acquisition-organize/services/acquisition.ts
+232 -136
View File
@@ -1,10 +1,4 @@
import type { import type { AcquisitionPolicyRules, AcquisitionPriority, CandidateEvaluation, SlotTemplate } from "@umt/contracts";
AcquisitionPolicyRules,
AcquisitionPriority,
AcquisitionSlot,
CandidateEvaluation,
SlotTemplate,
} from "@umt/contracts";
import type { AcquisitionSlotRow, MediaEntityRow, SourceReleaseRow, SubscriptionRow } from "@umt/db"; import type { AcquisitionSlotRow, MediaEntityRow, SourceReleaseRow, SubscriptionRow } from "@umt/db";
import { sql } from "@umt/db"; import { sql } from "@umt/db";
import { import {
@@ -12,15 +6,14 @@ import {
checkBudget, checkBudget,
computeUnitGaps, computeUnitGaps,
determineMode, determineMode,
type EvaluationContext,
effectiveLadder, effectiveLadder,
episodeKey, episodeKey,
evaluateCandidate, evaluateCandidate,
improvableUnits, improvableUnits,
type ParsedRelease, type ParsedRelease,
parseRelease,
priorityOf, priorityOf,
type QualityFacts, type QualityFacts,
rankForSlot,
ruleChainOf, ruleChainOf,
type SlotEvaluation, type SlotEvaluation,
type SlotSpec, type SlotSpec,
@@ -29,8 +22,8 @@ import {
transition, transition,
unitsOverlap, unitsOverlap,
} from "@umt/domain"; } from "@umt/domain";
import { fromParsedColumns, slotSpecFromRow, toContractSlotEvaluation } from "@umt/mapping"; import { toContractSlotEvaluation } from "@umt/mapping";
import { hashJson, jsonb, startOfDay, toApi } from "../util.ts"; import { hashJson, jsonb, startOfDay } from "../util.ts";
import { import {
type AoDeps, type AoDeps,
automationBlock, automationBlock,
@@ -45,24 +38,32 @@ import {
recordCandidate, recordCandidate,
} from "./common.ts"; } from "./common.ts";
import { autoFillCheck, type ForceFill } from "./evaluate-autofill.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 { import {
clearDeferral, clearDeferral,
clearReleaseDeferrals, clearReleaseDeferrals,
clearSubscriptionDeferrals, clearSubscriptionDeferrals,
gateRetryDelayMs, gateRetryDelayMs,
manualInflightPlans,
rangeOf, rangeOf,
recordDeferral, recordDeferral,
scheduleSlotReevaluate, scheduleSlotReevaluate,
scheduleSlotRetry, scheduleSlotRetry,
seasonInScope,
seasonLedgers, seasonLedgers,
slotCoveredUnits, slotCoveredUnits,
slotInflightPlans, slotInflightPlans,
slotSpecWithRules,
subParams, subParams,
subSeasonOf,
TRACKING_SUB_STATUSES, TRACKING_SUB_STATUSES,
} from "./subscriptions.ts"; } from "./subscriptions.ts";
import { applySlotPreset } from "./subscriptions-params.ts";
import { decideVolume, loadVolumes } from "./volumes.ts"; import { decideVolume, loadVolumes } from "./volumes.ts";
export { contextOf, parsedOf, unitKeysOf } from "./release-facts.ts";
// evaluateAcquisition:release.resolved → 匹配订阅/Slot → 模式 → 确定性评估 → 预算/缓存卷 → 冻结计划并预留。 // evaluateAcquisition:release.resolved → 匹配订阅/Slot → 模式 → 确定性评估 → 预算/缓存卷 → 冻结计划并预留。
// 拒绝与被挡住的原因都写成候选评估记录(artifact),供作品详情"为什么选/不选"。 // 拒绝与被挡住的原因都写成候选评估记录(artifact),供作品详情"为什么选/不选"。
@@ -98,34 +99,6 @@ const PRIORITY: Record<ReturnType<typeof priorityOf>, AcquisitionPriority> = {
const BUSY: readonly string[] = ["reserved", "downloading", "verifying"]; 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<SourceReleaseRow, "episode_resolution">): 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<string, unknown>, 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<SourceReleaseRow, "episode_resolution">): boolean { export function episodeUnresolved(r: Pick<SourceReleaseRow, "episode_resolution">): boolean {
const res = episodeResolutionOf(r); 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 } : {}) }; return { ...q, ...(a.size_bytes !== null ? { sizeBytes: a.size_bytes } : {}) };
} }
/** 单元键:有 content_unit 用其 id,没有时用 s{季}e{集} 合成键(追更判定两侧同口径) */
export function unitKeysOf(p: Pick<ParsedRelease, "season" | "episodes" | "airDate">): 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( async function releaseUnits(
exec: Exec, exec: Exec,
entityId: string, entityId: string,
@@ -227,9 +190,24 @@ export async function createSlotsFromTemplates(
policyRevisionId: string | null; policyRevisionId: string | null;
templates: readonly SlotTemplate[]; templates: readonly SlotTemplate[];
}, },
): Promise<void> { ): Promise<number> {
// 收录版本名在作品内唯一(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()) { 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") .insertInto("acquisition_slot")
.values({ .values({
// 同一事务建的收录版本 created_at 相同:按模板顺序给 position,第一个为主版本 // 同一事务建的收录版本 created_at 相同:按模板顺序给 position,第一个为主版本
@@ -237,7 +215,7 @@ export async function createSlotsFromTemplates(
media_entity_id: input.entityId, media_entity_id: input.entityId,
subscription_id: input.subscriptionId, subscription_id: input.subscriptionId,
policy_revision_id: input.policyRevisionId, policy_revision_id: input.policyRevisionId,
name: t.name, name,
hard_conditions: jsonb(t.hardConditions), hard_conditions: jsonb(t.hardConditions),
soft_preferences: jsonb(t.softPreferences), soft_preferences: jsonb(t.softPreferences),
fallback: jsonb(t.fallback), fallback: jsonb(t.fallback),
@@ -246,8 +224,58 @@ export async function createSlotsFromTemplates(
rule_group_ids: JSON.stringify(t.ruleGroupIds ?? []), rule_group_ids: JSON.stringify(t.ruleGroupIds ?? []),
}) })
.onConflict((oc) => oc.doNothing()) .onConflict((oc) => oc.doNothing())
.execute(); .returning("id")
.executeTakeFirst();
if (row) n++;
} }
return n;
}
/** 追踪中的订阅里没有自己收录版本的,逐个补建;返回是否建了 */
async function backfillSubscriptionSlots(
exec: Exec,
subs: readonly SubscriptionRow[],
slots: readonly Pick<AcquisitionSlotRow, "subscription_id">[],
): Promise<boolean> {
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<SubscriptionRow, "id" | "media_entity_id" | "params" | "policy_revision_id">,
): Promise<number> {
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; forceFill?: ForceFill | undefined;
} }
/** 作品原语言(规则组 CNSUB/CNVOI 放行),一次评估内缓存 */
function entityLang(c: Ctx, entityId: string): Promise<string | undefined> {
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<CandidateEvaluation, "evaluatedAt">) { async function note(c: Ctx, base: Omit<CandidateEvaluation, "evaluatedAt">) {
await recordCandidate(c.d.ctx.db, { ...base, evaluatedAt: c.now.toISOString() }); 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" }); c.out.skipped.push({ sourceReleaseId: input.sourceReleaseId, reason: "release_not_found" });
return c.out; return c.out;
} }
await evaluateRelease(c, rel, input.contentUnitIds ?? [], input.acquisitionSlotId); await evaluateRelease(c, rel, input.contentUnitIds ?? [], { slotId: input.acquisitionSlotId });
return c.out; return c.out;
} }
// 按 Slot / 订阅重评:取该作品最近已识别的发布 return reevaluate(c, input);
let entityId: string | null = null; }
/**
* 按收录版本 / 订阅重评时最多取的已识别资源条数(按首次出现倒序)。原来只看最近 50 条,长剧、多发布组时早期集
* 进不了评估(M5);放宽到 500 后先在内存里按订阅季范围、已覆盖单元过滤,只把可能有用的逐条评估
* (每条评估约十几次查询,500 条全过一遍也在几秒量级)。
*/
export const REEVALUATE_MAX_RELEASES = 500;
/**
* 按收录版本 / 订阅重评:延后的资源 + 作品已识别资源(按范围过滤),先用 rankForSlot 按主收录版本择优排序
* (pri → 来源 / 分辨率等级 → 软偏好),再逐条评估预留(M4:同一单元多个合格资源时 pri 最高的先占位,
* 其余延后 / 洗版判断)。逐条到来的 RSS 资源仍走 evaluateRelease,不受影响。
*/
async function reevaluate(c: Ctx, input: EvaluateInput): Promise<EvaluateOutcome> {
const db = c.d.ctx.db;
let sub: SubscriptionRow | undefined;
let focus: AcquisitionSlotRow[] = [];
if (input.acquisitionSlotId) { if (input.acquisitionSlotId) {
entityId = focus = await db.selectFrom("acquisition_slot").selectAll().where("id", "=", input.acquisitionSlotId).execute();
( const sid = focus[0]?.subscription_id;
await db sub = sid ? await db.selectFrom("subscription").selectAll().where("id", "=", sid).executeTakeFirst() : undefined;
.selectFrom("acquisition_slot")
.select("media_entity_id")
.where("id", "=", input.acquisitionSlotId)
.executeTakeFirst()
)?.media_entity_id ?? null;
} else if (input.subscriptionId) { } else if (input.subscriptionId) {
entityId = 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);
await db if (sub)
.selectFrom("subscription") focus = await db
.select("media_entity_id") .selectFrom("acquisition_slot")
.where("id", "=", input.subscriptionId) .selectAll()
.executeTakeFirst() .where("subscription_id", "=", sub.id)
)?.media_entity_id ?? null; .orderBy("position")
.orderBy("created_at")
.execute();
} }
const entityId = focus[0]?.media_entity_id ?? sub?.media_entity_id ?? null;
if (!entityId) return c.out; if (!entityId) return c.out;
// 先评该 Slot 延后的发布(按集号升序,连续更新的集依次补上),再评最近已识别的发布 const entity = await resolveEntity(db, entityId);
if (!entity) return c.out;
// 先评该 Slot 延后的发布(按集号升序,连续更新的集依次补上),再评作品其它已识别的发布
const deferred = input.acquisitionSlotId const deferred = input.acquisitionSlotId
? await db ? await db
.selectFrom("slot_deferral as sd") .selectFrom("slot_deferral as sd")
@@ -362,14 +414,93 @@ export async function evaluateAcquisition(d: AoDeps, input: EvaluateInput): Prom
.where("resolution_status", "=", "resolved") .where("resolution_status", "=", "resolved")
.where("availability", "=", "active") .where("availability", "=", "active")
.orderBy("first_seen_at", "desc") .orderBy("first_seen_at", "desc")
.limit(50) .limit(REEVALUATE_MAX_RELEASES)
.execute(); .execute();
for (const r of [...deferred, ...rels.filter((x) => !seen.has(x.id))]) const fresh = await inScopeReleases(
await evaluateRelease(c, r, [], input.acquisitionSlotId); 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; 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<SourceReleaseRow[]> {
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<number>().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<SourceReleaseRow[]> {
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<ParsedRelease, SourceReleaseRow>();
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 { d, now } = c;
const db = d.ctx.db; const db = d.ctx.db;
const skip = (reason: string) => c.out.skipped.push({ sourceReleaseId: rel.id, reason }); 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<string, SubscriptionRow>( const activeSubs = new Map<string, SubscriptionRow>(
subs.filter((s) => (TRACKING_SUB_STATUSES as readonly string[]).includes(s.status)).map((s) => [s.id, s]), 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 tracked = activeSubs.size > 0 || slots.some((s) => s.subscription_id === null);
const releaseSeason = parsed.season ?? (parsed.episodes ? 1 : undefined); const releaseSeason = parsed.season ?? (parsed.episodes ? 1 : undefined);
let planned: string | null = null; let planned: string | null = null;
for (const slot of slots) { for (const slot of slots) {
if (onlySlotId && slot.id !== onlySlotId) continue; 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; const sub = slot.subscription_id ? activeSubs.get(slot.subscription_id) : undefined;
if (slot.subscription_id && !sub) { if (slot.subscription_id && !sub) {
// 订阅已暂停/完成/取消:不评估,也不留延后行 // 订阅已暂停/完成/取消:不评估,也不留延后行
@@ -523,15 +664,8 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon
.executeTakeFirst(); .executeTakeFirst();
if (!seasonOk) continue; if (!seasonOk) continue;
} }
// 仅本季:发布季号与订阅季不一致则跳过 // 订阅季范围:仅本季 = 同季,接续后续季 = 不早于订阅季(与覆盖矩阵同口径)
if (sub?.season_scope === "current_season" && sub.season_unit_id && releaseSeason !== undefined) { if (sub && releaseSeason !== undefined && !seasonInScope(sub, await subSeasonOf(db, sub), releaseSeason)) continue;
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;
}
c.out.evaluated++; c.out.evaluated++;
const outcome = await evaluateForSlot(c, { const outcome = await evaluateForSlot(c, {
rel, rel,
@@ -583,33 +717,6 @@ async function clearEpisodeUnresolvedNotes(c: Ctx, sourceReleaseId: string, enti
and content->'reasons' ? 'episode_unresolved'`.execute(c.d.ctx.db); 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 { interface SlotInput {
rel: SourceReleaseRow; rel: SourceReleaseRow;
entity: MediaEntityRow; 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"] }); await note(c, { ...base, decision: "skipped", mode: null, rejected: [], reasons: ["site_out_of_scope"] });
return done; return done;
} }
const spec = slotSpecFromRow(toApi<AcquisitionSlot>(slot)); // 规则组:收录版本 ruleGroupIds → 订阅 ruleGroupIds → 订阅默认链;开升级叠加洗版默认链;私有组包含 / 排除并入硬条件
const ctx = contextOf(rel, now);
// 规则组:收录版本 ruleGroupIds → 订阅 ruleGroupIds → 订阅默认链;开升级叠加洗版默认链
c.rules ??= loadRuleCatalog(db); c.rules ??= loadRuleCatalog(db);
const chain = resolveRuleChain(await c.rules, { const spec = slotSpecWithRules(await c.rules, slot, sub, p.entity, rel.site_id);
kind: "acquisition", const ctx = contextOf(rel, now);
slotRuleGroupIds: (slot.rule_group_ids ?? []) as string[], if (spec.rules) {
subRuleGroupIds: params.ruleGroupIds, const lang = await entityLang(c, p.entity.id);
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);
if (lang) ctx.originalLanguage = lang; 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 inflightKeys = inflight.flatMap((x) => x.unitKeys);
const deferWithRetry = async (slotId: string, releaseId: string, reason: string, unitKeys: string[]) => { const deferWithRetry = async (slotId: string, releaseId: string, reason: string, unitKeys: string[]) => {
const attempts = await recordDeferral(db, { slotId, sourceReleaseId: releaseId, reason, unitKeys }); const attempts = await recordDeferral(db, { slotId, sourceReleaseId: releaseId, reason, unitKeys });
await scheduleSlotRetry(db, slotId, new Date(now.getTime() + gateRetryDelayMs(attempts))); await scheduleSlotRetry(db, slotId, new Date(now.getTime() + gateRetryDelayMs(attempts)));
}; };
const defer = async (reason: string, unitKeys: string[], mode: AcquisitionMode | null = null) => { 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, { await note(c, {
...base, ...base,
decision: "deferred", decision: "deferred",
apps/server/src/modules/acquisition-organize/services/release-facts.ts
+70
View File
@@ -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<SourceReleaseRow, "episode_resolution">): 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<string, unknown>, 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<ParsedRelease, "season" | "episodes" | "airDate">): 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) }
: {}),
};
}
apps/server/src/modules/acquisition-organize/services/subscriptions.ts
+316 -17
View File
@@ -1,7 +1,9 @@
import type { AcquisitionSlot, SubscriptionParams, SubscriptionProgress, UnitProgressState } from "@umt/contracts"; 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 { sql } from "@umt/db";
import { import {
currentPri,
effectiveLadder,
episodeKey, episodeKey,
movieCompletion, movieCompletion,
type ParsedRelease, type ParsedRelease,
@@ -10,13 +12,18 @@ import {
priorityLadder, priorityLadder,
type QualityFacts, type QualityFacts,
resolveEpisodeTotal, resolveEpisodeTotal,
ruleChainOf,
type SeasonProgress, type SeasonProgress,
type SlotSpec,
seasonProgress, seasonProgress,
seasonWildcard,
type UnitRange, type UnitRange,
} from "@umt/domain"; } from "@umt/domain";
import { slotSpecFromRow } from "@umt/mapping"; import { slotSpecFromRow } from "@umt/mapping";
import { errMessage, jsonb, toApi } from "../util.ts"; import { errMessage, jsonb, toApi } from "../util.ts";
import { type AoDeps, type Exec, upsertDecision } from "./common.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 重评调度、分集质量账本、主动搜索调度。 // W5-C 订阅业务:季总集数账本、缺集/在途、完成判定与历史、Slot 重评调度、分集质量账本、主动搜索调度。
// 对照 docs/porting/moviepilot-subscribe.md §3/§7/§8/§9。 // 对照 docs/porting/moviepilot-subscribe.md §3/§7/§8/§9。
@@ -39,6 +46,30 @@ export function rangeOf(sub: Pick<SubscriptionRow, "params"> | null | undefined)
return { ...(r.start !== undefined ? { start: r.start } : {}), ...(r.end !== undefined ? { end: r.end } : {}) }; return { ...(r.start !== undefined ? { start: r.start } : {}), ...(r.end !== undefined ? { end: r.end } : {}) };
} }
/** 订阅季号(订阅没有季单元或季单元没有季号时为 null) */
export async function subSeasonOf(exec: Exec, sub: Pick<SubscriptionRow, "season_unit_id">): Promise<number | null> {
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<SubscriptionRow, "season_scope">,
subSeason: number | null,
releaseSeason: number,
): boolean {
if (subSeason === null) return true;
return sub.season_scope === "following_seasons" ? releaseSeason >= subSeason : releaseSeason === subSeason;
}
// ── 季总集数账本 ── // ── 季总集数账本 ──
export interface SeasonLedger { export interface SeasonLedger {
@@ -268,15 +299,259 @@ export async function slotInflightPlans(exec: Exec, slotId: string, excludePlanI
return [...byId.values()]; return [...byId.values()];
} }
/**
* 同作品的手动下载计划(slot_id 为空)里仍在途的(M6):订阅评估把它们并入在途单元,手动下载中的集不再被订阅重复下。
* 手动计划没记单元,按资源解析补:单集 / 多集 → 集键;季包 → 整季通配;电影 → 空(电影收录版本有在途即忙)。
*/
export async function manualInflightPlans(exec: Exec, entityId: string): Promise<InflightPlan[]> {
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<string>`p.evaluation->>'mediaEntityId'`, "=", entityId)]),
)
.execute();
const byId = new Map<string, InflightPlan>();
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 { export interface SlotCompletion {
/** 覆盖齐(Slot 静止态按它落 satisfied / empty) */
complete: boolean; complete: boolean;
/**
* 有升级目标(规则组 / 阶梯按 pri 洗版)时,范围内每个已覆盖单元(电影 = 片子)都已到目标 pri;
* 没有升级目标恒为 true。订阅完成要求 complete && atTarget(H2)。
*/
atTarget: boolean;
seasons: SeasonProgress[]; seasons: SeasonProgress[];
/** 继续追踪时的疑似缺集(总集数未知,已见最高集以下的空洞) */ /** 继续追踪时的疑似缺集(总集数未知,已见最高集以下的空洞) */
suspectedMissing: { season: number; episodes: number[] }[]; 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<SubscriptionRow, "id" | "params"> | null | undefined,
entity: { id: string; kind: string },
siteId?: string,
): SlotSpec {
const spec = slotSpecFromRow(toApi<AcquisitionSlot>(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<SubscriptionRow, "mode" | "season_scope">): boolean =>
sub.mode === "once" || sub.season_scope === "current_season";
/** 收录版本是否算完成(订阅完成用):覆盖齐且到洗版目标 */
export const slotDone = (c: Pick<SlotCompletion, "complete" | "atTarget">): boolean => c.complete && c.atTarget;
interface PriVersion {
/** 覆盖的单元:季 → 集号集合;null = 整季(季包通配) */
units: Map<number, Set<number> | 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<number, readonly number[]>;
},
): Promise<boolean> {
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<string>(),
),
];
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<number, Set<number> | 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<number>();
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<number, Set<number> | 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<number>()).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 限定季单元 > 订阅季单元 > 作品已知的所有正片季 */ /** 剧集 Slot 覆盖的季:Slot 限定季单元 > 订阅季单元 > 作品已知的所有正片季 */
async function scopeSeasons( async function scopeSeasons(
exec: Exec, exec: Exec,
@@ -314,10 +589,7 @@ async function scopeSeasons(
*/ */
export async function slotComplete( export async function slotComplete(
exec: Exec, exec: Exec,
slot: Pick< slot: AcquisitionSlotRow,
AcquisitionSlotRow,
"id" | "media_entity_id" | "content_unit_id" | "completion" | "subscription_id" | "current_asset_id"
>,
mediaType: "movie" | "tv", mediaType: "movie" | "tv",
newAssets: readonly string[], newAssets: readonly string[],
opts: { includeSubmitted?: boolean } = {}, opts: { includeSubmitted?: boolean } = {},
@@ -333,6 +605,32 @@ export async function slotComplete(
.where("p.status_reason", "!=", "accepted_unconfirmed") .where("p.status_reason", "!=", "accepted_unconfirmed")
.execute() .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<number, readonly number[]>, 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") { if (mediaType === "movie") {
const assets = newAssets.length const assets = newAssets.length
? await exec ? await exec
@@ -355,19 +653,16 @@ export async function slotComplete(
: {}), : {}),
})), })),
]).complete; ]).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 const slotCount = await exec
.selectFrom("acquisition_slot") .selectFrom("acquisition_slot")
.select((eb) => eb.fn.countAll<number>().as("n")) .select((eb) => eb.fn.countAll<number>().as("n"))
.where("media_entity_id", "=", slot.media_entity_id) .where("media_entity_id", "=", slot.media_entity_id)
.executeTakeFirstOrThrow(); .executeTakeFirstOrThrow();
// 单 Slot 作品:作品下所有资产都算该 Slot 的覆盖(手动整理/历史导入);多 Slot 只算本 Slot 的 // 单 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 inflight = await slotInflightPlans(exec, slot.id);
const inflightEps = new Map<number, number[]>(); const inflightEps = new Map<number, number[]>();
for (const p of inflight) 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 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 { return {
complete, complete,
atTarget: complete && (await reached(inScope, entityWide)),
seasons: progress, seasons: progress,
suspectedMissing: progress suspectedMissing: progress
.filter((p) => p.suspectedMissing.length > 0) .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 const sub = await trx
.selectFrom("subscription") .selectFrom("subscription")
.selectAll() .selectAll()
.where("id", "=", slot.subscription_id) .where("id", "=", slot.subscription_id)
.executeTakeFirst(); .executeTakeFirst();
const seasonDone = sub?.media_kind !== "movie" && sub?.season_scope === "current_season"; if (sub && sub.status === "active" && completesOnCoverage(sub)) {
if (sub && sub.status === "active" && (sub.mode === "once" || seasonDone)) {
const siblings = await trx const siblings = await trx
.selectFrom("acquisition_slot") .selectFrom("acquisition_slot")
.selectAll() .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 }); 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(); .executeTakeFirst();
if (sub?.status !== "active") return; if (sub?.status !== "active") return;
await touchProgress(trx, { subscriptionId: sub.id }, d.ports.now()); 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(); const slots = await trx.selectFrom("acquisition_slot").selectAll().where("subscription_id", "=", sub.id).execute();
if (!slots.length) return; if (!slots.length) return;
const progress = await Promise.all( 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", { await completeSubscription(d, trx, sub, "download_accepted", {
slots: progress, slots: progress,
seasons: progress.flatMap((p) => p.seasons), seasons: progress.flatMap((p) => p.seasons),
apps/server/test/acquisition-organize/upgrade-completion.test.ts
+155
View File
@@ -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<string, Uint8Array>();
let base: Awaited<ReturnType<typeof seedBase>>;
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);
});
apps/server/test/subscriptions/subscription-scope-reevaluate.test.ts
+222
View File
@@ -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<ReturnType<typeof seedBase>>;
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<string, unknown>) {
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);
});
});