From 410d7fd636ab7f9b9efc02f9f2c913c6896980a3 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 28 Sep 2026 15:54:29 +0800 Subject: [PATCH] feat(subscriptions): overview groups, coverage matrix, multi-version auto-fill, form field normalization (W9-E) - 0035: subscription.last_progress_at, acquisition_slot.position (primary version) - GET /subscriptions: server-side groups + counts, card fields, sort/filter, offset cursor - GET /subscriptions/{id}/coverage: season x episode (movie: edition) x slot matrix, 30s cache - auto-fill for non-primary slots: inherit/on/off, per-subscription daily budget, upgrade priority - params normalization with MP/legacy aliases; duplicate 409; pause/resume/resubscribe/batch/fill/slot-order Co-Authored-By: Claude Opus 5.5 --- .../modules/acquisition-organize/handlers.ts | 2 + .../acquisition-organize/routes/rules.ts | 252 +------ .../routes/subscriptions.ts | 682 ++++++++++++++++++ .../services/acquisition.ts | 53 +- .../acquisition-organize/services/coverage.ts | 678 +++++++++++++++++ .../services/evaluate-autofill.ts | 123 ++++ .../services/subscriptions-params.ts | 260 +++++++ .../services/subscriptions.ts | 26 + packages/contracts/src/api.ts | 83 +++ packages/contracts/src/policy.ts | 201 +++++- .../db/migrations/0035_subscriptions_v2.ts | 48 ++ packages/db/src/types.ts | 2 + 12 files changed, 2158 insertions(+), 252 deletions(-) create mode 100644 apps/server/src/modules/acquisition-organize/routes/subscriptions.ts create mode 100644 apps/server/src/modules/acquisition-organize/services/coverage.ts create mode 100644 apps/server/src/modules/acquisition-organize/services/evaluate-autofill.ts create mode 100644 apps/server/src/modules/acquisition-organize/services/subscriptions-params.ts create mode 100644 packages/db/migrations/0035_subscriptions_v2.ts diff --git a/apps/server/src/modules/acquisition-organize/handlers.ts b/apps/server/src/modules/acquisition-organize/handlers.ts index 0d82d13a..e6344625 100644 --- a/apps/server/src/modules/acquisition-organize/handlers.ts +++ b/apps/server/src/modules/acquisition-organize/handlers.ts @@ -81,6 +81,8 @@ export function registerAoTasks(registry: TaskRegistry, d: AoDeps) { acquisitionSlotId: id.optional(), sourceReleaseId: id.optional(), contentUnitIds: z.array(id).optional(), + /** 手动补全某版本(POST /subscriptions/{id}/slots/{slotId}/fill) */ + forceFill: z.object({ unitKeys: z.array(z.string().min(1)).optional() }).optional(), }), retryPolicy: DB_RETRY, handler: async (t) => diff --git a/apps/server/src/modules/acquisition-organize/routes/rules.ts b/apps/server/src/modules/acquisition-organize/routes/rules.ts index 96a63f3e..a14384c2 100644 --- a/apps/server/src/modules/acquisition-organize/routes/rules.ts +++ b/apps/server/src/modules/acquisition-organize/routes/rules.ts @@ -5,8 +5,6 @@ import { type PolicyRevision, type PreferenceSuppression, type PreferenceTag, - type Subscription, - type SubscriptionHistory, } from "@umt/contracts"; import { type DB, type PreferenceSuppressionRow, type RuleGroupRow, sql, type Transaction } from "@umt/db"; import { evaluateCandidate, parseRelease, type RuleExpr, type RuleLayer, serializeLayers } from "@umt/domain"; @@ -15,10 +13,8 @@ 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 type { AoDeps, Exec } from "../services/common.ts"; import { normalizeGroupInput } from "../services/rules.ts"; -import { applyEpisodeTotal, enqueueSubscriptionSearch, subscriptionProgress } from "../services/subscriptions.ts"; import { actorOf, conflict, errMessage, hashJson, jsonb, notFound, toApi, toApiList } from "../util.ts"; import { mount, paginate } from "./helpers.ts"; import { @@ -28,8 +24,9 @@ import { registerRuleGroupRoutes, suppressRuleGroup, } from "./rule-groups.ts"; +import { registerSubscriptionRoutes } from "./subscriptions.ts"; -// 规则 API:订阅、收录 Slot、策略版本(草案→发布→回滚,发布即持续授权)、偏好标签(规则组别名)与抑制记录; +// 规则 API:收录 Slot、策略版本(草案→发布→回滚,发布即持续授权)、偏好标签(规则组别名)与抑制记录; // 规则组本体见 ./rule-groups.ts。 const TAG = "rules"; @@ -92,249 +89,8 @@ async function auditPref( export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { const db = () => d.ctx.db; - // ── 订阅 ── - mount( - app, - Api.subscriptions.list, - async (_c, { query }) => { - let q = db() - .selectFrom("subscription as s") - .leftJoin("media_entity as e", "e.id", "s.media_entity_id") - .selectAll("s") - .select("e.title as media_entity_title"); - if (query.status) q = q.where("s.status", "=", query.status); - if (query.mediaEntityId) q = q.where("s.media_entity_id", "=", query.mediaEntityId); - if (query.mediaKind) q = q.where("s.media_kind", "=", query.mediaKind); - if (query.q) { - const like = `%${String(query.q).replace(/[\\%_]/g, (m) => `\\${m}`)}%`; - q = q.where((eb) => - eb.or([ - eb("e.title", "ilike", like), - eb("e.original_title", "ilike", like), - eb("s.title_query", "ilike", like), - eb.exists( - eb - .selectFrom("title_alias as t") - .select("t.id") - .whereRef("t.media_entity_id", "=", "s.media_entity_id") - .where("t.alias", "ilike", like), - ), - ]), - ); - } - const p = await paginate(q, query, "s."); - return { items: toApiList(p.rows), nextCursor: p.nextCursor }; - }, - { tag: TAG }, - ); - mount( - app, - Api.subscriptions.get, - async (_c, { params }) => { - const r = await db().selectFrom("subscription").selectAll().where("id", "=", params.id).executeTakeFirst(); - return toApi(r ?? notFound("订阅")); - }, - { tag: TAG }, - ); - mount( - app, - Api.subscriptions.create, - async (c, { body }) => { - if (body.mediaEntityId) { - const e = await db() - .selectFrom("media_entity") - .select("id") - .where("id", "=", body.mediaEntityId) - .executeTakeFirst(); - if (!e) throw new ApiException("bad_request", "mediaEntityId 不存在"); - } - // pending:TMDB 未就绪(未建档,或剧集没有已接受的 TMDB 身份 → 总集数无来源):照常搜索/评估,不能完成 - const tmdbReady = - !!body.mediaEntityId && - (body.mediaKind === "movie" || - !!(await db() - .selectFrom("external_identity") - .select("id") - .where("media_entity_id", "=", body.mediaEntityId) - .where("provider", "=", "tmdb") - .where("status", "=", "accepted") - .executeTakeFirst())); - const row = await db() - .transaction() - .execute(async (trx) => { - const r = await trx - .insertInto("subscription") - .values({ - status: tmdbReady ? "active" : "pending", - status_reason: tmdbReady ? null : "metadata_pending", - media_entity_id: body.mediaEntityId ?? null, - season_unit_id: body.seasonUnitId ?? null, - title_query: body.titleQuery ?? null, - media_kind: body.mediaKind, - mode: body.mode ?? "continuous", - // 用户主动添加的剧集默认持续追踪并接续后续季;电影无季概念 - season_scope: body.seasonScope ?? (body.mediaKind === "movie" ? "current_season" : "following_seasons"), - origin: body.origin ?? "user", - priority: body.priority ?? "tracking", - policy_revision_id: body.policyRevisionId ?? null, - params: jsonb(body.params), - created_by: actorOf(c), - }) - .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); - if (pol?.rules.defaultSlots?.length) { - await createSlotsFromTemplates(trx, { - entityId: r.media_entity_id, - subscriptionId: r.id, - policyRevisionId: pol.row.id, - templates: pol.rules.defaultSlots, - }); - } - } - } - return r; - }); - if (row.media_entity_id) { - await kickEvaluate(d, { subscriptionId: row.id }); - // 新订阅:立即主动搜索一次(之后按周期);剧集先刷新元数据拿季总集数 - await enqueueSubscriptionSearch(d, row.id, "created").catch((err) => - d.ctx.logger.debug({ err: { message: errMessage(err) } }, "订阅搜索入队失败"), - ); - if (row.media_kind !== "movie" && tmdbReady) - await d.ctx.engine - .enqueue( - "refreshMetadata", - { mediaEntityId: row.media_entity_id }, - { idempotencyKey: `refreshMetadata:subscription:${row.id}`, traceId: row.id }, - ) - .catch((err) => d.ctx.logger.debug({ err: { message: errMessage(err) } }, "元数据刷新入队失败")); - } - return toApi(row); - }, - { tag: TAG }, - ); - mount( - app, - Api.subscriptions.update, - async (_c, { params, body }) => { - const r = await db() - .updateTable("subscription") - .set({ - ...(body.mode !== undefined ? { mode: body.mode } : {}), - ...(body.seasonScope !== undefined ? { season_scope: body.seasonScope } : {}), - ...(body.priority !== undefined ? { priority: body.priority } : {}), - ...(body.status !== undefined ? { status: body.status } : {}), - ...(body.policyRevisionId !== undefined ? { policy_revision_id: body.policyRevisionId } : {}), - ...(body.params !== undefined ? { params: jsonb(body.params) } : {}), - ...(body.statusReason !== undefined ? { status_reason: body.statusReason } : {}), - updated_at: new Date(), - }) - .where("id", "=", params.id) - .returningAll() - .executeTakeFirst(); - return toApi(r ?? notFound("订阅")); - }, - { tag: TAG }, - ); - - mount( - app, - Api.subscriptions.search, - async (_c, { params }) => { - const sub = await db() - .selectFrom("subscription") - .select(["id", "media_entity_id", "status"]) - .where("id", "=", params.id) - .executeTakeFirst(); - if (!sub) notFound("订阅"); - if (!sub!.media_entity_id) throw new ApiException("bad_request", "订阅未关联作品,无法搜索"); - try { - return await enqueueSubscriptionSearch(d, sub!.id, "manual"); - } catch (err) { - throw new ApiException("upstream_unavailable", `搜索任务入队失败:${errMessage(err)}`); - } - }, - { tag: TAG }, - ); - mount( - app, - Api.subscriptions.progress, - async (_c, { params }) => (await subscriptionProgress(db(), params.id)) ?? notFound("订阅"), - { tag: TAG }, - ); - mount( - app, - Api.subscriptions.setEpisodeTotal, - async (_c, { params, body }) => { - const sub = await db() - .selectFrom("subscription") - .select(["id", "media_entity_id", "media_kind"]) - .where("id", "=", params.id) - .executeTakeFirst(); - if (!sub) notFound("订阅"); - if (!sub!.media_entity_id || sub!.media_kind === "movie") - throw new ApiException("bad_request", "只有已关联作品的剧集/综艺订阅有季总集数"); - await db() - .transaction() - .execute(async (trx) => { - const now = d.ports.now(); - if (body.locked) { - await applyEpisodeTotal(trx, { - entityId: sub!.media_entity_id!, - season: body.seasonNumber, - manual: { total: body.total, locked: true }, - now, - }); - } else { - // 未锁定:按注入口径只升 - await applyEpisodeTotal(trx, { - entityId: sub!.media_entity_id!, - season: body.seasonNumber, - injected: body.total, - now, - }); - } - }); - return (await subscriptionProgress(db(), params.id))!; - }, - { tag: TAG }, - ); - mount( - app, - Api.subscriptionHistory.list, - async (_c, { query }) => { - let q = db().selectFrom("subscription_history").selectAll(); - if (query.mediaEntityId) q = q.where("media_entity_id", "=", query.mediaEntityId); - if (query.subscriptionId) q = q.where("subscription_id", "=", query.subscriptionId); - const p = await paginate(q, query); - return { items: toApiList(p.rows), nextCursor: p.nextCursor }; - }, - { tag: TAG }, - ); - mount( - app, - Api.subscriptionHistory.get, - async (_c, { params }) => { - const r = await db() - .selectFrom("subscription_history") - .selectAll() - .where("id", "=", params.id) - .executeTakeFirst(); - return toApi(r ?? notFound("订阅历史")); - }, - { tag: TAG }, - ); + // ── 订阅(W9-E 起在 ./subscriptions.ts) ── + registerSubscriptionRoutes(app, d); // ── 收录 Slot ── mount( diff --git a/apps/server/src/modules/acquisition-organize/routes/subscriptions.ts b/apps/server/src/modules/acquisition-organize/routes/subscriptions.ts new file mode 100644 index 00000000..f94b8fb3 --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/routes/subscriptions.ts @@ -0,0 +1,682 @@ +import { + type AcquisitionSlot, + Api, + type Subscription, + type SubscriptionGroup, + type SubscriptionHistory, + type SubscriptionOverviewItem, + type SubscriptionOverviewPage, + type SubscriptionSort, +} from "@umt/contracts"; +import type { SubscriptionRow } from "@umt/db"; +import type { Context } from "hono"; +import type { AuthEnv } from "../../../http/auth.ts"; +import { ApiException } from "../../../http/errors.ts"; +import type { ModuleApp } from "../../types.ts"; +import { createSlotsFromTemplates, publishedAcquisitionPolicies } from "../services/acquisition.ts"; +import type { AoDeps, Exec } from "../services/common.ts"; +import { + buildSubscriptionModels, + invalidateCoverage, + lastUpdatedOf, + overviewGroup, + type SubscriptionModel, + slotOrder, + slotSummaries, + slotsOfSub, + subscriptionCoverage, +} from "../services/coverage.ts"; +import { loadRuleCatalog } from "../services/rules.ts"; +import { + applyEpisodeTotal, + enqueueSubscriptionSearch, + subParams, + subscriptionProgress, + TRACKING_SUB_STATUSES, +} from "../services/subscriptions.ts"; +import { + applySlotUpgrade, + type NormalizedParams, + normalizeSubscriptionParams, +} from "../services/subscriptions-params.ts"; +import { actorOf, conflict, errMessage, jsonb, notFound, toApi, toApiList } from "../util.ts"; +import { mount, paginate } from "./helpers.ts"; + +// W9-E 订阅 API(docs/design/subscriptions.md §1–§3、§5):总览(分组/排序/筛选)、覆盖矩阵、表单字段规范化、 +// 暂停/恢复/重新订阅、批量、手动补全某版本、版本目标排序;主动搜索、进度、总集数、完成历史沿用 W5-C。 + +const TAG = "subscriptions"; +const OVERVIEW_MAX = 2000; +const GROUPS: readonly SubscriptionGroup[] = ["following", "waiting", "completed", "paused", "pending"]; + +/** 规则变化后按需重评(引擎未就绪/入队失败不影响写入) */ +async function kickEvaluate(d: AoDeps, input: { subscriptionId?: string; acquisitionSlotId?: string }) { + try { + const key = input.acquisitionSlotId ?? input.subscriptionId ?? "x"; + await d.ctx.engine.enqueue("evaluateAcquisition", input, { + idempotencyKey: `reeval:${key}:${Date.now()}`, + traceId: key, + }); + } catch (err) { + d.ctx.logger.debug({ err: { message: errMessage(err) } }, "重评入队失败"); + } +} + +const encodeOffset = (n: number) => Buffer.from(`o:${n}`, "utf8").toString("base64url"); +function decodeOffset(cursor: string | undefined): number { + if (!cursor) return 0; + const m = /^o:(\d+)$/.exec(Buffer.from(cursor, "base64url").toString("utf8")); + if (!m) throw new ApiException("bad_request", "cursor 无效"); + return Number(m[1]); +} + +/** 剧集是否有已接受的 TMDB 身份(pending 判定) */ +async function tmdbReady(exec: Exec, entityId: string | null, mediaKind: string): Promise { + if (!entityId) return false; + if (mediaKind === "movie") return true; + return !!(await exec + .selectFrom("external_identity") + .select("id") + .where("media_entity_id", "=", entityId) + .where("provider", "=", "tmdb") + .where("status", "=", "accepted") + .executeTakeFirst()); +} + +/** 同作品同季已有未结束的订阅 → 409(差距 #18) */ +async function assertNoDuplicate( + exec: Exec, + input: { entityId: string | null; seasonUnitId: string | null; excludeId?: string }, +) { + if (!input.entityId) return; + let q = exec + .selectFrom("subscription") + .select("id") + .where("media_entity_id", "=", input.entityId) + .where("status", "in", ["active", "pending", "paused"]); + q = input.seasonUnitId ? q.where("season_unit_id", "=", input.seasonUnitId) : q.where("season_unit_id", "is", null); + if (input.excludeId) q = q.where("id", "!=", input.excludeId); + const hit = await q.executeTakeFirst(); + if (hit) conflict(`该作品同一季已有订阅:${hit.id}`); +} + +/** 保存后应用表单的派生写入:MP total_episode → 季总集数;best_version → 版本目标升级策略 */ +async function applyDerived(exec: Exec, sub: SubscriptionRow, n: NormalizedParams, now: Date) { + if (!sub.media_entity_id) return; + if (n.episodeTotal) { + const season = sub.season_unit_id + ? (( + await exec + .selectFrom("content_unit") + .select("season_number") + .where("id", "=", sub.season_unit_id) + .executeTakeFirst() + )?.season_number ?? 1) + : 1; + await applyEpisodeTotal(exec, { + entityId: sub.media_entity_id, + season, + manual: { total: n.episodeTotal.total, locked: n.episodeTotal.locked }, + now, + }); + } + if (n.upgrade) + await applySlotUpgrade(exec, { subscriptionId: sub.id, entityId: sub.media_entity_id, strategy: n.upgrade }); +} + +async function audit( + exec: Exec, + c: Context, + action: string, + subId: string, + payload: Record = {}, +) { + const actor = c.get("actor"); + await exec + .insertInto("audit_event") + .values({ + actor_type: actor?.type === "api_key" ? "api_key" : "user", + actor_id: actor?.id ?? null, + action: `subscription.${action}`, + target_type: "subscription", + target_id: subId, + payload: jsonb(payload), + }) + .execute(); +} + +function overviewItem(row: SubscriptionRow & { media_entity_title?: string | null }, m: SubscriptionModel) { + const slots = slotSummaries(m); + const primary = slots.find((s) => s.primary); + const g = overviewGroup(m); + const p = subParams(row); + return { + ...toApi(row), + group: g.group, + year: m.entity?.year ?? null, + seasonNumber: m.seasonNumber, + progress: primary + ? { have: primary.have, total: primary.total, inflight: primary.inflight } + : { have: 0, total: null, inflight: 0 }, + missing: primary?.missing ?? 0, + slots, + lastUpdatedAt: lastUpdatedOf(m).toISOString(), + searchEnabled: p.searchIntervalHours !== 0, + sites: [] as { id: string; name: string }[], + ruleGroups: [] as { id: string; name: string }[], + defaultRuleGroups: m.defaultRuleGroups, + autoFill: m.autoFill, + upgrading: slots.some((s) => s.upgrading), + awaitingNewSeason: g.awaitingNewSeason, + _ruleGroupIds: m.ruleGroupIds, + _slotRuleGroupIds: slots.flatMap((s) => s.ruleGroupIds), + }; +} + +const SORT_DEFAULT: Record = { + lastUpdated: "desc", + nextSearch: "asc", + missing: "desc", + created: "desc", + name: "asc", +}; + +type Item = ReturnType; + +function sorter(sort: SubscriptionSort, order: "asc" | "desc") { + const dir = order === "asc" ? 1 : -1; + const name = (x: Item) => x.mediaEntityTitle ?? x.titleQuery ?? ""; + return (a: Item, b: Item): number => { + let v = 0; + switch (sort) { + case "lastUpdated": + v = a.lastUpdatedAt.localeCompare(b.lastUpdatedAt); + break; + case "created": + v = a.createdAt.localeCompare(b.createdAt); + break; + case "missing": + v = a.missing - b.missing; + break; + case "name": + v = name(a).localeCompare(name(b), "zh"); + break; + case "nextSearch": { + // 没排期(已关/已完成)的永远排最后 + if (!a.nextSearchAt || !b.nextSearchAt) + return a.nextSearchAt ? -1 : b.nextSearchAt ? 1 : a.id.localeCompare(b.id); + v = a.nextSearchAt.localeCompare(b.nextSearchAt); + break; + } + } + return v * dir || a.id.localeCompare(b.id); + }; +} + +export function registerSubscriptionRoutes(app: ModuleApp, d: AoDeps) { + const db = () => d.ctx.db; + + const load = async (id: string) => + (await db().selectFrom("subscription").selectAll().where("id", "=", id).executeTakeFirst()) ?? notFound("订阅"); + + // ── 总览 ── + mount( + app, + Api.subscriptions.list, + async (_c, { query }): Promise => { + const now = d.ports.now(); + let q = db() + .selectFrom("subscription as s") + .leftJoin("media_entity as e", "e.id", "s.media_entity_id") + .selectAll("s") + .select("e.title as media_entity_title"); + if (query.status) q = q.where("s.status", "=", query.status); + else q = q.where("s.status", "!=", "cancelled"); + if (query.mediaEntityId) q = q.where("s.media_entity_id", "=", query.mediaEntityId); + if (query.mediaKind) q = q.where("s.media_kind", "=", query.mediaKind); + if (query.origin) q = q.where("s.origin", "=", query.origin); + if (query.q) { + const like = `%${String(query.q).replace(/[\\%_]/g, (m) => `\\${m}`)}%`; + q = q.where((eb) => + eb.or([ + eb("e.title", "ilike", like), + eb("e.original_title", "ilike", like), + eb("s.title_query", "ilike", like), + eb.exists( + eb + .selectFrom("title_alias as t") + .select("t.id") + .whereRef("t.media_entity_id", "=", "s.media_entity_id") + .where("t.alias", "ilike", like), + ), + ]), + ); + } + const rows = await q.orderBy("s.created_at", "desc").limit(OVERVIEW_MAX).execute(); + const cat = await loadRuleCatalog(db()); + const models = await buildSubscriptionModels(db(), rows, now, cat); + let items = rows.map((r) => overviewItem(r, models.get(r.id)!)); + if (query.siteId) + items = items.filter((x) => { + const ids = (subParams(x as never).siteIds ?? []) as string[]; + return ids.length === 0 || ids.includes(query.siteId); + }); + if (query.ruleGroupId) + items = items.filter( + (x) => x._ruleGroupIds.includes(query.ruleGroupId) || x._slotRuleGroupIds.includes(query.ruleGroupId), + ); + if (query.upgrading) items = items.filter((x) => x.upgrading === (query.upgrading === "true")); + const counts = Object.fromEntries(GROUPS.map((g) => [g, 0])) as Record; + for (const x of items) if (x.group) counts[x.group]++; + if (query.group) items = items.filter((x) => x.group === query.group); + const sort: SubscriptionSort = query.sort ?? "lastUpdated"; + items.sort(sorter(sort, query.order ?? SORT_DEFAULT[sort])); + const total = items.length; + const offset = decodeOffset(query.cursor); + const page = items.slice(offset, offset + query.limit); + // 站点 / 规则组名只查这一页用到的 + const siteIds = [...new Set(page.flatMap((x) => (subParams(x as never).siteIds ?? []) as string[]))]; + const sites = siteIds.length + ? await db().selectFrom("site").select(["id", "name"]).where("id", "in", siteIds).execute() + : []; + const out: SubscriptionOverviewItem[] = page.map(({ _ruleGroupIds, _slotRuleGroupIds, ...x }) => ({ + ...x, + sites: ((subParams(x as never).siteIds ?? []) as string[]).map((id) => ({ + id, + name: sites.find((s) => s.id === id)?.name ?? id, + })), + ruleGroups: _ruleGroupIds.map((id) => ({ id, name: cat.groups.get(id)?.name ?? id })), + })); + return { + items: out, + nextCursor: offset + query.limit < total ? encodeOffset(offset + query.limit) : null, + counts, + total, + }; + }, + { tag: TAG }, + ); + mount(app, Api.subscriptions.get, async (_c, { params }) => toApi(await load(params.id)), { + tag: TAG, + }); + + // ── 新建 / 修改(表单字段规范化,兼容 MP 字段与旧键) ── + mount( + app, + Api.subscriptions.create, + async (c, { body }) => { + if (body.mediaEntityId) { + const e = await db() + .selectFrom("media_entity") + .select("id") + .where("id", "=", body.mediaEntityId) + .executeTakeFirst(); + if (!e) throw new ApiException("bad_request", "mediaEntityId 不存在"); + } + const norm = await normalizeSubscriptionParams(db(), body.params, { mediaKind: body.mediaKind }); + await assertNoDuplicate(db(), { entityId: body.mediaEntityId ?? null, seasonUnitId: body.seasonUnitId ?? null }); + // pending:TMDB 未就绪(未建档,或剧集没有已接受的 TMDB 身份 → 总集数无来源):照常搜索/评估,不能完成 + const ready = await tmdbReady(db(), body.mediaEntityId ?? null, body.mediaKind); + const now = d.ports.now(); + const row = await db() + .transaction() + .execute(async (trx) => { + const r = await trx + .insertInto("subscription") + .values({ + status: ready ? "active" : "pending", + status_reason: ready ? null : "metadata_pending", + media_entity_id: body.mediaEntityId ?? null, + season_unit_id: body.seasonUnitId ?? null, + title_query: body.titleQuery ?? null, + media_kind: body.mediaKind, + mode: body.mode ?? "continuous", + // 用户主动添加的剧集默认持续追踪并接续后续季;电影无季概念 + season_scope: body.seasonScope ?? (body.mediaKind === "movie" ? "current_season" : "following_seasons"), + origin: body.origin ?? "user", + priority: body.priority ?? "tracking", + policy_revision_id: body.policyRevisionId ?? null, + params: jsonb(norm.params), + created_by: actorOf(c), + }) + .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); + if (pol?.rules.defaultSlots?.length) { + await createSlotsFromTemplates(trx, { + entityId: r.media_entity_id, + subscriptionId: r.id, + policyRevisionId: pol.row.id, + templates: pol.rules.defaultSlots, + }); + } + } + } + await applyDerived(trx, r, norm, now); + return r; + }); + if (row.media_entity_id) { + await kickEvaluate(d, { subscriptionId: row.id }); + // 新订阅:立即主动搜索一次(之后按周期);剧集先刷新元数据拿季总集数 + await enqueueSubscriptionSearch(d, row.id, "created").catch((err) => + d.ctx.logger.debug({ err: { message: errMessage(err) } }, "订阅搜索入队失败"), + ); + if (row.media_kind !== "movie" && ready) + await d.ctx.engine + .enqueue( + "refreshMetadata", + { mediaEntityId: row.media_entity_id }, + { idempotencyKey: `refreshMetadata:subscription:${row.id}`, traceId: row.id }, + ) + .catch((err) => d.ctx.logger.debug({ err: { message: errMessage(err) } }, "元数据刷新入队失败")); + } + return toApi(row); + }, + { tag: TAG }, + ); + mount( + app, + Api.subscriptions.update, + async (_c, { params, body }) => { + const cur = await load(params.id); + const norm = + body.params !== undefined + ? await normalizeSubscriptionParams(db(), body.params, { mediaKind: cur.media_kind }) + : undefined; + if (body.status !== undefined && ["active", "pending", "paused"].includes(body.status)) + await assertNoDuplicate(db(), { + entityId: cur.media_entity_id, + seasonUnitId: cur.season_unit_id, + excludeId: cur.id, + }); + const r = await db() + .transaction() + .execute(async (trx) => { + const r = await trx + .updateTable("subscription") + .set({ + ...(body.mode !== undefined ? { mode: body.mode } : {}), + ...(body.seasonScope !== undefined ? { season_scope: body.seasonScope } : {}), + ...(body.priority !== undefined ? { priority: body.priority } : {}), + ...(body.status !== undefined ? { status: body.status } : {}), + ...(body.policyRevisionId !== undefined ? { policy_revision_id: body.policyRevisionId } : {}), + ...(norm ? { params: jsonb(norm.params) } : {}), + ...(body.statusReason !== undefined ? { status_reason: body.statusReason } : {}), + updated_at: new Date(), + }) + .where("id", "=", params.id) + .returningAll() + .executeTakeFirstOrThrow(); + if (norm) await applyDerived(trx, r, norm, d.ports.now()); + return r; + }); + invalidateCoverage(r.id); + if (norm && r.media_entity_id && (TRACKING_SUB_STATUSES as readonly string[]).includes(r.status)) + await kickEvaluate(d, { subscriptionId: r.id }); + return toApi(r); + }, + { tag: TAG }, + ); + + // ── 状态动作 ── + const setStatus = async ( + sub: SubscriptionRow, + to: "active" | "pending" | "paused" | "cancelled", + reason: string, + extra: { nextSearchNow?: boolean } = {}, + ) => { + const now = d.ports.now(); + const searchOn = subParams(sub).searchIntervalHours !== 0; + const r = await db() + .updateTable("subscription") + .set({ + status: to, + status_reason: reason, + ...(to === "paused" || to === "cancelled" ? { next_search_at: null } : {}), + ...(extra.nextSearchNow && searchOn ? { next_search_at: now } : {}), + }) + .where("id", "=", sub.id) + .returningAll() + .executeTakeFirstOrThrow(); + invalidateCoverage(sub.id); + return r; + }; + const pause = async (sub: SubscriptionRow) => { + if (sub.status === "paused") return sub; + if (!(TRACKING_SUB_STATUSES as readonly string[]).includes(sub.status)) + conflict(`订阅状态为 ${sub.status},不能暂停`); + return setStatus(sub, "paused", "user_paused"); + }; + const resume = async (sub: SubscriptionRow) => { + if ((TRACKING_SUB_STATUSES as readonly string[]).includes(sub.status)) return sub; + if (sub.status !== "paused") conflict(`订阅状态为 ${sub.status},已完成/已取消的订阅用"重新订阅"`); + const ready = await tmdbReady(db(), sub.media_entity_id, sub.media_kind); + const r = await setStatus(sub, ready ? "active" : "pending", ready ? "resumed" : "metadata_pending", { + nextSearchNow: true, + }); + if (r.media_entity_id) await kickEvaluate(d, { subscriptionId: r.id }); + return r; + }; + const search = async (sub: SubscriptionRow) => { + if (!sub.media_entity_id) throw new ApiException("bad_request", "订阅未关联作品,无法搜索"); + try { + return await enqueueSubscriptionSearch(d, sub.id, "manual"); + } catch (err) { + throw new ApiException("upstream_unavailable", `搜索任务入队失败:${errMessage(err)}`); + } + }; + + mount( + app, + Api.subscriptions.pause, + async (_c, { params }) => toApi(await pause(await load(params.id))), + { + tag: TAG, + }, + ); + mount( + app, + Api.subscriptions.resume, + async (_c, { params }) => toApi(await resume(await load(params.id))), + { tag: TAG }, + ); + mount( + app, + Api.subscriptions.resubscribe, + async (c, { params }) => { + const sub = await load(params.id); + if (sub.status !== "completed" && sub.status !== "cancelled") + conflict(sub.status === "paused" ? "暂停的订阅用恢复" : "订阅仍在追踪"); + await assertNoDuplicate(db(), { + entityId: sub.media_entity_id, + seasonUnitId: sub.season_unit_id, + excludeId: sub.id, + }); + const ready = await tmdbReady(db(), sub.media_entity_id, sub.media_kind); + const r = await setStatus(sub, ready ? "active" : "pending", ready ? "resubscribed" : "metadata_pending"); + await audit(db(), c, "resubscribed", r.id, { from: sub.status }); + if (r.media_entity_id) { + await kickEvaluate(d, { subscriptionId: r.id }); + await search(r).catch((err) => d.ctx.logger.debug({ err: { message: errMessage(err) } }, "订阅搜索入队失败")); + } + return toApi(r); + }, + { tag: TAG }, + ); + mount( + app, + Api.subscriptions.batch, + async (c, { body }) => { + const results: { id: string; ok: boolean; error: string | null }[] = []; + for (const id of body.ids as string[]) { + try { + const sub = await load(id); + if (body.action === "pause") await pause(sub); + else if (body.action === "resume") await resume(sub); + else if (body.action === "search") await search(sub); + else if (sub.status !== "cancelled") { + await setStatus(sub, "cancelled", "user_deleted"); + await audit(db(), c, "deleted", id, { from: sub.status, batch: true }); + } + results.push({ id, ok: true, error: null }); + } catch (err) { + results.push({ id, ok: false, error: errMessage(err) }); + } + } + return { results }; + }, + { tag: TAG, status: 200 }, + ); + + // ── 搜索 / 进度 / 覆盖矩阵 ── + mount(app, Api.subscriptions.search, async (_c, { params }) => search(await load(params.id)), { tag: TAG }); + mount( + app, + Api.subscriptions.progress, + async (_c, { params }) => (await subscriptionProgress(db(), params.id)) ?? notFound("订阅"), + { tag: TAG }, + ); + mount( + app, + Api.subscriptions.coverage, + async (_c, { params, query }) => + (await subscriptionCoverage(db(), params.id, d.ports.now(), { refresh: query.refresh === "true" })) ?? + notFound("订阅"), + { tag: TAG }, + ); + + // ── 版本目标:手动补全 / 排序 ── + mount( + app, + Api.subscriptions.fillSlot, + async (_c, { params, body }) => { + const sub = await load(params.id); + if (!(TRACKING_SUB_STATUSES as readonly string[]).includes(sub.status)) + conflict("订阅不在追踪中(暂停/已完成/已取消),先恢复或重新订阅"); + const slots = sub.media_entity_id + ? await db() + .selectFrom("acquisition_slot") + .selectAll() + .where("media_entity_id", "=", sub.media_entity_id) + .execute() + : []; + const slot = slotsOfSub(slots, sub).find((s) => s.id === params.slotId); + if (!slot) notFound("版本目标"); + invalidateCoverage(sub.id); + try { + return await d.ctx.engine.enqueue( + "evaluateAcquisition", + { acquisitionSlotId: params.slotId, forceFill: { ...(body.unitKeys ? { unitKeys: body.unitKeys } : {}) } }, + { idempotencyKey: `fill:${params.slotId}:${d.ports.now().getTime()}`, traceId: params.slotId }, + ); + } catch (err) { + throw new ApiException("upstream_unavailable", `补全任务入队失败:${errMessage(err)}`); + } + }, + { tag: TAG, status: 202 }, + ); + mount( + app, + Api.subscriptions.slotOrder, + async (_c, { params, body }) => { + const sub = await load(params.id); + if (!sub.media_entity_id) throw new ApiException("bad_request", "订阅未关联作品"); + const out = await db() + .transaction() + .execute(async (trx) => { + const all = await trx + .selectFrom("acquisition_slot") + .selectAll() + .where("media_entity_id", "=", sub.media_entity_id!) + .forUpdate() + .execute(); + const mine = slotsOfSub(all, sub); + const want = body.slotIds as string[]; + if ( + want.length !== mine.length || + new Set(want).size !== want.length || + !want.every((id) => mine.some((s) => s.id === id)) + ) + throw new ApiException("bad_request", "slotIds 必须恰好是该订阅的全部版本目标"); + for (const [i, id] of want.entries()) + await trx.updateTable("acquisition_slot").set({ position: i }).where("id", "=", id).execute(); + return (await trx.selectFrom("acquisition_slot").selectAll().where("id", "in", want).execute()).sort( + slotOrder, + ); + }); + invalidateCoverage(sub.id); + return toApiList(out); + }, + { tag: TAG }, + ); + + // ── 总集数 / 完成历史 ── + mount( + app, + Api.subscriptions.setEpisodeTotal, + async (_c, { params, body }) => { + const sub = await load(params.id); + if (!sub.media_entity_id || sub.media_kind === "movie") + throw new ApiException("bad_request", "只有已关联作品的剧集/综艺订阅有季总集数"); + await db() + .transaction() + .execute(async (trx) => { + const now = d.ports.now(); + if (body.locked) { + await applyEpisodeTotal(trx, { + entityId: sub.media_entity_id!, + season: body.seasonNumber, + manual: { total: body.total, locked: true }, + now, + }); + } else { + // 未锁定:按注入口径只升 + await applyEpisodeTotal(trx, { + entityId: sub.media_entity_id!, + season: body.seasonNumber, + injected: body.total, + now, + }); + } + }); + invalidateCoverage(sub.id); + return (await subscriptionProgress(db(), params.id))!; + }, + { tag: TAG }, + ); + mount( + app, + Api.subscriptionHistory.list, + async (_c, { query }) => { + let q = db().selectFrom("subscription_history").selectAll(); + if (query.mediaEntityId) q = q.where("media_entity_id", "=", query.mediaEntityId); + if (query.subscriptionId) q = q.where("subscription_id", "=", query.subscriptionId); + const p = await paginate(q, query); + return { items: toApiList(p.rows), nextCursor: p.nextCursor }; + }, + { tag: TAG }, + ); + mount( + app, + Api.subscriptionHistory.get, + async (_c, { params }) => { + const r = await db() + .selectFrom("subscription_history") + .selectAll() + .where("id", "=", params.id) + .executeTakeFirst(); + return toApi(r ?? notFound("订阅历史")); + }, + { tag: TAG }, + ); +} diff --git a/apps/server/src/modules/acquisition-organize/services/acquisition.ts b/apps/server/src/modules/acquisition-organize/services/acquisition.ts index 8f30d0df..e6bf0e64 100644 --- a/apps/server/src/modules/acquisition-organize/services/acquisition.ts +++ b/apps/server/src/modules/acquisition-organize/services/acquisition.ts @@ -32,6 +32,7 @@ import { import { fromParsedColumns, slotSpecFromRow, toContractSlotEvaluation } from "@umt/mapping"; import { hashJson, jsonb, startOfDay, toApi } from "../util.ts"; import { type AoDeps, automationBlock, type Exec, enqueueTracked, loadInstances, recordCandidate } from "./common.ts"; +import { autoFillCheck, type ForceFill } from "./evaluate-autofill.ts"; import { entityOriginalLanguage, loadRuleCatalog, type RuleCatalog, resolveRuleChain } from "./rules.ts"; import { clearDeferral, @@ -44,6 +45,7 @@ import { subParams, TRACKING_SUB_STATUSES, } from "./subscriptions.ts"; +import { withSubscriptionFilters } from "./subscriptions-params.ts"; import { decideVolume, loadVolumes } from "./volumes.ts"; // evaluateAcquisition:release.resolved → 匹配订阅/Slot → 模式 → 确定性评估 → 预算/缓存卷 → 冻结计划并预留。 @@ -60,6 +62,8 @@ export interface EvaluateInput { * 由 download.submit 在授权通过后转为 reserved 再提交。 */ draft?: boolean | undefined; + /** 手动补全某版本(W9-E):不受自动补全开关与补全预算限制,可限定单元 */ + forceFill?: ForceFill | undefined; } export interface EvaluateOutcome { @@ -183,10 +187,12 @@ export async function createSlotsFromTemplates( templates: readonly SlotTemplate[]; }, ): Promise { - for (const t of input.templates) { + for (const [i, t] of input.templates.entries()) { await exec .insertInto("acquisition_slot") .values({ + // 同一事务建的版本目标 created_at 相同:按模板顺序给 position,第一个为主版本 + position: i, media_entity_id: input.entityId, subscription_id: input.subscriptionId, policy_revision_id: input.policyRevisionId, @@ -233,6 +239,7 @@ interface Ctx { langs?: Map>; /** 现有资产在规则组下的 pri 缓存 */ assetPri?: Map; + forceFill?: ForceFill | undefined; } async function note(c: Ctx, base: Omit) { @@ -245,6 +252,7 @@ export async function evaluateAcquisition(d: AoDeps, input: EvaluateInput): Prom now: d.ports.now(), out: { evaluated: 0, plans: [], skipped: [], deferred: [] }, draft: input.draft === true, + forceFill: input.forceFill, }; const db = d.ctx.db; if (input.sourceReleaseId) { @@ -330,10 +338,12 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon const isMovie = entity.kind === "movie"; let subs = await db.selectFrom("subscription").selectAll().where("media_entity_id", "=", entity.id).execute(); + // 主版本(position 最小)先评:同一发布只建一个计划时主版本优先 let slots = await db .selectFrom("acquisition_slot") .selectAll() .where("media_entity_id", "=", entity.id) + .orderBy("position") .orderBy("created_at") .execute(); let forcedMode: AcquisitionMode | null = null; @@ -397,6 +407,7 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon .selectFrom("acquisition_slot") .selectAll() .where("media_entity_id", "=", entity.id) + .orderBy("position") .orderBy("created_at") .execute(); forcedMode = "new_title"; @@ -534,7 +545,7 @@ 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 spec = withSubscriptionFilters(slotSpecFromRow(toApi(slot)), params); const ctx = contextOf(rel, now); // 规则组:版本目标 ruleGroupIds → 订阅 ruleGroupIds → 订阅默认链;开升级叠加洗版默认链 c.rules ??= loadRuleCatalog(db); @@ -702,6 +713,37 @@ async function evaluateForSlot(c: Ctx, p: SlotInput): Promise<{ planId: string | }); return done; } + // 多版本自动补全(W9-E):非主版本按订阅开关/补全预算决定是否建计划、用什么优先级 + c.rules ??= loadRuleCatalog(db); + const fill = await autoFillCheck(db, { + slot, + sub, + isMovie, + planUnits, + releaseBytes: rel.size_bytes, + catalog: await c.rules, + now, + force: c.forceFill, + }); + if (fill.action === "skip") { + await note(c, { ...base, decision: "skipped", mode, rejected: [], reasons: [fill.reason] }); + return done; + } + if (fill.action === "defer") { + if (fill.record) return defer(fill.reason, planUnits, mode); + await note(c, { + ...base, + decision: "deferred", + mode, + rejected: [], + reasons: [`deferred:${fill.reason}`, ...fill.detail], + ...evalExtra, + unitKeys: planUnits, + }); + c.out.deferred.push({ sourceReleaseId: rel.id, slotId: slot.id, reason: fill.reason }); + return done; + } + planUnits = fill.planUnits; const gate = await gates(c, rel, mode, sub); if (!gate.ok) { await note(c, { ...base, decision: "skipped", mode, rejected: gate.rejected, reasons: gate.reasons, ...evalExtra }); @@ -719,6 +761,7 @@ async function evaluateForSlot(c: Ctx, p: SlotInput): Promise<{ planId: string | savePath: gate.savePath, unitKeys: planUnits, multiInflight: !isMovie, + ...(fill.action === "fill" ? { priority: fill.priority, evaluationExtra: fill.evaluation } : {}), }); if (r.conflict) return defer("units_inflight", r.conflict, mode); const planId = r.planId; @@ -900,11 +943,14 @@ async function reserve( unitKeys: string[]; /** 剧集 Slot:允许与其它在途计划并存(单元不相交) */ multiInflight: boolean; + /** 自动补全计划:优先级降为 upgrade,评估里记 autoFill(补全预算按它计) */ + priority?: AcquisitionPriority; + evaluationExtra?: Record; }, ): Promise<{ planId: string | null; conflict?: string[] }> { const { d, now } = c; const immediate = p.sub?.priority === "immediate"; - const priority = PRIORITY[priorityOf(p.mode, immediate)]; + const priority = immediate ? PRIORITY.immediate : (p.priority ?? PRIORITY[priorityOf(p.mode, immediate)]); const payload = { slotId: p.slot.id, sourceReleaseId: p.rel.id, @@ -919,6 +965,7 @@ async function reserve( instanceId: p.instanceId, volumeKey: p.volumeKey, savePath: p.savePath, + ...(p.evaluationExtra ?? {}), }); const purpose = p.mode === "upgrade" ? ("upgrade" as const) : ("fill" as const); if (c.draft) { diff --git a/apps/server/src/modules/acquisition-organize/services/coverage.ts b/apps/server/src/modules/acquisition-organize/services/coverage.ts new file mode 100644 index 00000000..20368fb0 --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/services/coverage.ts @@ -0,0 +1,678 @@ +import type { + AcquisitionSlot, + CoverageColumn, + SubscriptionCoverage, + SubscriptionGroup, + SubscriptionOverviewItem, + SubscriptionSlotSummary, +} from "@umt/contracts"; +import type { AcquisitionSlotRow, SubscriptionRow } from "@umt/db"; +import { sql } from "@umt/db"; +import { + type CoverageColumnInput, + type CoveredFact, + coverageRow, + episodeKey, + parseUnitKey, + resolveAutoFill, + subscriptionGroup, + type UnitRange, +} from "@umt/domain"; +import { slotSpecFromRow } from "@umt/mapping"; +import { toApi } from "../util.ts"; +import type { Exec } from "./common.ts"; +import { defaultChainIds, loadRuleCatalog, type RuleCatalog, resolveRuleChain } from "./rules.ts"; +import { INFLIGHT_PLAN_STATUSES, rangeOf, subParams } from "./subscriptions.ts"; + +// W9-E 覆盖矩阵与订阅总览(docs/design/subscriptions.md §1、§2.1、§5): +// 一批订阅一次取齐 版本目标 / 资产 / 覆盖 / 分集账本 / 在途计划 / 内容单元 / 延后(固定 8 条查询,与订阅数无关), +// 在内存里按版本目标算每格状态(domain coverageRow)。覆盖口径与 slotCoveredUnits 相同: +// 版本目标覆盖 = 该目标计划产出的资产 + 当前资产 + 分集账本资产(作品只有一个版本目标时,作品下全部资产都算)。 + +const MAX_EPISODES = 2000; +const COVERAGE_TTL_MS = 30_000; + +type Quality = { resolution: string | null; source: string | null; label: string | null }; + +interface SlotModel { + row: AcquisitionSlotRow; + primary: boolean; + autoFill: boolean; + ruleGroupIds: string[]; + upgrade: { enabled: boolean; targetPri: number }; + requiredEditions: string[]; +} + +interface RowModel { + slotId: string | null; + name: string; + primary: boolean; + autoFill: boolean; + status: string; + summary: { have: number; inflight: number; missing: number; skipped: number }; + cells: SubscriptionCoverage["rows"][number]["cells"]; + minPri: number | null; + upgrade: { enabled: boolean; targetPri: number }; + ruleGroupIds: string[]; +} + +export interface SubscriptionModel { + sub: SubscriptionRow; + entity: { id: string; title: string; year: number | null; kind: string } | null; + seasonNumber: number | null; + seasons: SubscriptionCoverage["seasons"]; + rows: RowModel[]; + totalsKnown: boolean; + lastPlanAt: Date | null; + autoFill: SubscriptionOverviewItem["autoFill"]; + ruleGroupIds: string[]; + defaultRuleGroups: boolean; +} + +const qualityOf = (q: unknown): Quality | null => { + if (!q || typeof q !== "object") return null; + const o = q as { resolution?: string; source?: string }; + const resolution = o.resolution ?? null; + const source = o.source ?? null; + if (!resolution && !source) return null; + return { resolution, source, label: [resolution, source].filter(Boolean).join(" ") || null }; +}; + +const dateKey = (yyyymmdd: string) => `d${yyyymmdd}`; +/** 综艺分段(上/中/下);整集或未知不标 */ +const segOf = (u: { segment: string } | undefined) => + u && u.segment !== "full" && u.segment !== "unknown" ? u.segment : null; +const editionKey = (e: string) => `edition:${e}`; +const ymd = (d: Date) => d.toISOString().slice(0, 10); + +/** 订阅有效自动补全:params.autoFill → 规则链上后者优先的 auto_fill → 系统缺省开 */ +export function subscriptionAutoFill( + cat: RuleCatalog, + sub: Pick, +): SubscriptionOverviewItem["autoFill"] & { chainIds: string[] } { + const p = subParams(sub); + const chain = resolveRuleChain(cat, { + kind: "acquisition", + subRuleGroupIds: p.ruleGroupIds, + subscriptionId: sub.id, + mediaKind: sub.media_kind, + entityId: sub.media_entity_id ?? undefined, + }); + const r = resolveAutoFill( + p.autoFill, + chain.ids.map((id) => ({ id, autoFill: cat.groups.get(id)?.auto_fill ?? null })), + ); + return { + enabled: r.enabled, + mode: r.mode, + source: r.source, + budget: p.fillBudget ?? {}, + chainIds: chain.ids, + }; +} + +/** 版本目标排序:position → created_at → id;第一个为主版本 */ +export const slotOrder = (a: AcquisitionSlotRow, b: AcquisitionSlotRow) => + a.position - b.position || a.created_at.getTime() - b.created_at.getTime() || a.id.localeCompare(b.id); + +/** 订阅关联的版本目标:同作品、属于该订阅或不属于任何订阅 */ +export const slotsOfSub = (all: readonly AcquisitionSlotRow[], sub: Pick) => + all + .filter( + (s) => s.media_entity_id === sub.media_entity_id && (s.subscription_id === sub.id || s.subscription_id === null), + ) + .sort(slotOrder); + +/** 批量构建订阅模型(总览与矩阵共用) */ +export async function buildSubscriptionModels( + exec: Exec, + subs: readonly SubscriptionRow[], + now: Date, + catalog?: RuleCatalog, +): Promise> { + const out = new Map(); + if (!subs.length) return out; + const cat = catalog ?? (await loadRuleCatalog(exec)); + const entityIds = [...new Set(subs.map((s) => s.media_entity_id).filter((x): x is string => !!x))]; + const none = ["00000000-0000-0000-0000-000000000000"]; + const E = entityIds.length ? entityIds : none; + + const [entities, slots, units] = await Promise.all([ + exec.selectFrom("media_entity").select(["id", "title", "year", "kind"]).where("id", "in", E).execute(), + exec.selectFrom("acquisition_slot").selectAll().where("media_entity_id", "in", E).execute(), + exec + .selectFrom("content_unit") + .select([ + "id", + "media_entity_id", + "kind", + "season_number", + "episode_number", + sql`to_char(air_date, 'YYYY-MM-DD')`.as("air"), + "episode_total", + "episode_total_source", + "episode_total_locked", + "content_type", + "segment", + "status", + "created_at", + ]) + .where("media_entity_id", "in", E) + .where("status", "!=", "retired") + .where("kind", "in", ["season", "episode"]) + .orderBy("created_at") + .execute(), + ]); + const slotIds = slots.length ? slots.map((s) => s.id) : none; + const currentAssets = slots.map((s) => s.current_asset_id).filter((x): x is string => !!x); + + const [assets, ledgerRows, inflightRows, lastPlans, deferrals] = await Promise.all([ + sql<{ + id: string; + media_entity_id: string | null; + content_edition: string | null; + quality: unknown; + unit_keys: unknown; + plan_slot_id: string | null; + }>` + select a.id, a.media_entity_id, a.content_edition, a.media_info->'_umt'->'quality' as quality, + a.media_info->'_umt'->'unitKeys' as unit_keys, p.slot_id as plan_slot_id + from media_asset a + left join download_task dt on dt.id = a.download_task_id + left join acquisition_plan p on p.id = dt.acquisition_plan_id + where a.status = 'active' and a.parent_asset_id is null + and (a.media_entity_id in (${sql.join(E)}) + ${currentAssets.length ? sql`or a.id in (${sql.join(currentAssets)})` : sql``})`.execute(exec), + exec + .selectFrom("slot_unit as su") + .innerJoin("content_unit as u", "u.id", "su.content_unit_id") + .select([ + "su.slot_id", + "su.media_asset_id", + "su.quality", + "su.pri", + "u.id as unit_id", + "u.kind", + "u.season_number", + "u.episode_number", + sql`to_char(u.air_date, 'YYYYMMDD')`.as("air"), + ]) + .where("su.slot_id", "in", slotIds) + .where("su.media_asset_id", "is not", null) + .execute(), + exec + .selectFrom("acquisition_plan as p") + .leftJoin("download_task as dt", (j) => + j.onRef("dt.acquisition_plan_id", "=", "p.id").on("dt.removed_at", "is", null), + ) + .select(["p.id", "p.slot_id", "p.status", "p.unit_keys", "dt.status as dt_status"]) + .where("p.slot_id", "in", slotIds) + .where("p.status", "in", [...INFLIGHT_PLAN_STATUSES]) + .execute(), + exec + .selectFrom("acquisition_plan") + .select(["slot_id", (eb) => eb.fn.max("created_at").as("at")]) + .where("slot_id", "in", slotIds) + .where("status", "!=", "draft") + .groupBy("slot_id") + .execute(), + exec.selectFrom("slot_deferral").select(["slot_id", "unit_keys"]).where("slot_id", "in", slotIds).execute(), + ]); + const assetIds = assets.rows.map((a) => a.id); + const coverage = assetIds.length + ? await sql<{ + asset_id: string; + unit_id: string; + kind: string; + season_number: number | null; + episode_number: number | null; + air: string | null; + }>` + select c.media_asset_id as asset_id, u.id as unit_id, u.kind, u.season_number, u.episode_number, + to_char(u.air_date, 'YYYYMMDD') as air + from asset_unit_coverage c join content_unit u on u.id = c.content_unit_id + where c.media_asset_id in (${sql.join(assetIds)})`.execute(exec) + : { rows: [] }; + + // ── 索引 ── + const assetById = new Map(assets.rows.map((a) => [a.id, a])); + const covByAsset = new Map(); + for (const c of coverage.rows) covByAsset.set(c.asset_id, [...(covByAsset.get(c.asset_id) ?? []), c]); + const ledgerBySlot = new Map(); + for (const r of ledgerRows) ledgerBySlot.set(r.slot_id, [...(ledgerBySlot.get(r.slot_id) ?? []), r]); + const slotsByEntity = new Map(); + for (const s of slots) slotsByEntity.set(s.media_entity_id, [...(slotsByEntity.get(s.media_entity_id) ?? []), s]); + const inflightBySlot = new Map(); + for (const r of inflightRows) { + const stage = + r.status === "reserved" + ? "reserved" + : r.dt_status === "completed" || r.dt_status === "importing" + ? "verifying" + : "downloading"; + const list = inflightBySlot.get(r.slot_id) ?? []; + const prev = list.find((x) => x.id === r.id); + if (!prev) list.push({ id: r.id, stage, unitKeys: r.unit_keys }); + else if (stage === "verifying") prev.stage = stage; + inflightBySlot.set(r.slot_id, list); + } + const lastPlanBySlot = new Map(lastPlans.map((r) => [r.slot_id, r.at ? new Date(r.at as Date) : null])); + const deferBySlot = new Map(); + for (const r of deferrals) deferBySlot.set(r.slot_id, [...(deferBySlot.get(r.slot_id) ?? []), ...r.unit_keys]); + const unitsByEntity = new Map(); + for (const u of units) unitsByEntity.set(u.media_entity_id, [...(unitsByEntity.get(u.media_entity_id) ?? []), u]); + const unitById = new Map(units.map((u) => [u.id, u])); + const today = ymd(now); + + /** 资产集合 → 覆盖事实(单元键 → 资产/质量) */ + const coveredOf = (assetSet: Iterable, ledger: typeof ledgerRows): Map => { + const m = new Map(); + const put = (k: string, f: CoveredFact, override = false) => { + if (override || !m.has(k)) m.set(k, f); + }; + for (const id of assetSet) { + const a = assetById.get(id); + if (!a) continue; + const q = qualityOf(a.quality); + const pri = + typeof (a.quality as { pri?: unknown } | null)?.pri === "number" ? (a.quality as { pri: number }).pri : null; + for (const c of covByAsset.get(id) ?? []) { + const f = { assetId: id, quality: q, pri }; + if (c.kind === "episode" && c.episode_number !== null) + put(episodeKey(c.season_number ?? 1, c.episode_number), f); + else if (c.air) put(dateKey(c.air), f); + } + for (const k of Array.isArray(a.unit_keys) ? (a.unit_keys as string[]) : []) { + const p = parseUnitKey(k); + if (p?.episode !== undefined || /^d\d{8}$/.test(k)) put(k, { assetId: id, quality: q, pri }); + } + } + // 分集账本优先(质量与 pri 按版本目标记) + for (const r of ledger) { + const f = { assetId: r.media_asset_id, quality: qualityOf(r.quality), pri: r.pri }; + if (r.kind === "episode" && r.episode_number !== null) + put(episodeKey(r.season_number ?? 1, r.episode_number), f, true); + else if (r.air) put(dateKey(r.air), f, true); + } + return m; + }; + + for (const sub of subs) { + const entity = entities.find((e) => e.id === sub.media_entity_id) ?? null; + const af = subscriptionAutoFill(cat, sub); + const p = subParams(sub); + const explicitGroups = p.ruleGroupIds ?? []; + const subSlots = entity ? slotsOfSub(slotsByEntity.get(entity.id) ?? [], sub) : []; + const entitySlotCount = entity ? (slotsByEntity.get(entity.id) ?? []).length : 0; + const slotRuleIds = [...new Set(subSlots.flatMap((s) => (s.rule_group_ids ?? []) as string[]))]; + const ruleGroupIds = explicitGroups.length + ? explicitGroups + : slotRuleIds.length + ? slotRuleIds + : defaultChainIds(cat, "acquisition"); + const models: SlotModel[] = subSlots.map((s, i) => { + const spec = slotSpecFromRow(toApi(s)); + const cuts = (spec.completion && spec.completion.kind === "movie" ? spec.completion.cuts : undefined) ?? []; + return { + row: s, + primary: i === 0, + autoFill: i === 0 || af.enabled, + ruleGroupIds: (s.rule_group_ids ?? []) as string[], + upgrade: { enabled: !!spec.upgrade?.enabled, targetPri: spec.upgrade?.targetPri ?? 100 }, + requiredEditions: cuts.length ? cuts.map(String) : ["default"], + }; + }); + const assetSetOf = (s: AcquisitionSlotRow): Set => { + const set = new Set(); + for (const a of assets.rows) { + if (a.plan_slot_id === s.id || a.id === s.current_asset_id) set.add(a.id); + else if (entitySlotCount <= 1 && a.media_entity_id === s.media_entity_id) set.add(a.id); + } + for (const r of ledgerBySlot.get(s.id) ?? []) if (r.media_asset_id) set.add(r.media_asset_id); + return set; + }; + const entityAssets = entity ? assets.rows.filter((a) => a.media_entity_id === entity.id).map((a) => a.id) : []; + const allLedger = subSlots.flatMap((s) => ledgerBySlot.get(s.id) ?? []); + const allInflight = subSlots.flatMap((s) => inflightBySlot.get(s.id) ?? []); + const lastPlanAt = subSlots.reduce((acc, s) => { + const t = lastPlanBySlot.get(s.id) ?? null; + return t && (!acc || t > acc) ? t : acc; + }, null); + const range = rangeOf(sub); + const isMovie = sub.media_kind === "movie"; + + const seasons: SubscriptionCoverage["seasons"] = []; + let colInputs: CoverageColumnInput[] = []; + const anyCovered = coveredOf(entityAssets, allLedger); + const slotCovered = new Map( + models.map((m) => [m.row.id, coveredOf(assetSetOf(m.row), ledgerBySlot.get(m.row.id) ?? [])]), + ); + let totalsKnown = true; + let seasonNumber: number | null = null; + + if (entity && isMovie) { + const editions = new Set(models.flatMap((m) => m.requiredEditions)); + if (!editions.size) editions.add("default"); + for (const a of entityAssets) editions.add(assetById.get(a)?.content_edition ?? "default"); + const columns: CoverageColumn[] = [...editions].map((e) => ({ + unitKey: editionKey(e), + episode: null, + airDate: null, + contentUnitId: null, + temporary: false, + segment: null, + edition: e, + })); + seasons.push({ + seasonNumber: null, + seasonUnitId: null, + total: columns.length, + totalKnown: true, + source: null, + locked: false, + columns, + }); + colInputs = columns.map((c) => ({ unitKey: c.unitKey })); + } else if (entity) { + const eUnits = unitsByEntity.get(entity.id) ?? []; + // 季总集数账本:每个季号取最早建的季单元 + const ledgers = new Map(); + for (const u of eUnits) + if (u.kind === "season" && u.season_number !== null && u.season_number >= 1 && !ledgers.has(u.season_number)) + ledgers.set(u.season_number, u); + const epUnits = eUnits.filter((u) => u.kind === "episode"); + const seen = new Set([...ledgers.keys()]); + for (const u of epUnits) + if (u.season_number !== null && u.season_number >= 1 && u.content_type === "main") seen.add(u.season_number); + const epSeasonsOf = (keys: Iterable) => { + for (const k of keys) { + const pk = parseUnitKey(k); + if (pk && pk.season >= 1) seen.add(pk.season); + } + }; + epSeasonsOf(anyCovered.keys()); + for (const pl of allInflight) epSeasonsOf(pl.unitKeys); + let scope = [...seen].sort((a, b) => a - b); + const subSeason = sub.season_unit_id ? (unitById.get(sub.season_unit_id)?.season_number ?? null) : null; + if (subSeason !== null) { + seasonNumber = subSeason; + scope = + sub.season_scope === "following_seasons" + ? [...new Set([subSeason, ...scope.filter((n) => n >= subSeason)])].sort((a, b) => a - b) + : [subSeason]; + } else { + const slotSeasons = models + .map((m) => (m.row.content_unit_id ? (unitById.get(m.row.content_unit_id)?.season_number ?? null) : null)) + .filter((x): x is number => x !== null); + if (models.length && slotSeasons.length === models.length) + scope = [...new Set(slotSeasons)].sort((a, b) => a - b); + } + const inflightEps = (season: number) => + allInflight.flatMap((pl) => + pl.unitKeys + .map(parseUnitKey) + .filter((x) => x?.season === season && x.episode !== undefined) + .map((x) => x!.episode!), + ); + const coveredEps = (season: number) => + [...anyCovered.keys(), ...[...slotCovered.values()].flatMap((m) => [...m.keys()])] + .map(parseUnitKey) + .filter((x) => x?.season === season && x.episode !== undefined) + .map((x) => x!.episode!); + for (const n of scope) { + const l = ledgers.get(n); + const known = !!l && l.episode_total !== null && l.episode_total > 0; + if (!known) totalsKnown = false; + const sUnits = epUnits.filter((u) => (u.season_number ?? 1) === n && u.episode_number !== null); + const maxEp = Math.min( + MAX_EPISODES, + Math.max( + known ? l!.episode_total! : 0, + 0, + ...coveredEps(n), + ...inflightEps(n), + ...(known ? [] : sUnits.map((u) => u.episode_number!)), + ), + ); + const columns: CoverageColumn[] = []; + for (let e = 1; e <= maxEp; e++) { + const cu = sUnits.find((u) => u.episode_number === e); + columns.push({ + unitKey: episodeKey(n, e), + episode: e, + airDate: cu?.air ?? null, + contentUnitId: cu?.id ?? null, + temporary: !cu || cu.status === "provisional", + segment: segOf(cu), + edition: null, + }); + colInputs.push({ unitKey: episodeKey(n, e), episode: e, airDate: cu?.air ?? null, totalKnown: known }); + } + seasons.push({ + seasonNumber: n, + seasonUnitId: l?.id ?? null, + total: known ? l!.episode_total : null, + totalKnown: known, + source: l?.episode_total_source ?? null, + locked: !!l?.episode_total_locked, + columns, + }); + } + // 综艺:只有播出日期的单元(d{YYYYMMDD})单列一段 + const dateKeys = new Set(); + for (const u of epUnits) if (u.episode_number === null && u.air) dateKeys.add(dateKey(u.air.replaceAll("-", ""))); + for (const k of anyCovered.keys()) if (/^d\d{8}$/.test(k)) dateKeys.add(k); + for (const pl of allInflight) for (const k of pl.unitKeys) if (/^d\d{8}$/.test(k)) dateKeys.add(k); + if (dateKeys.size) { + const columns: CoverageColumn[] = [...dateKeys].sort().map((k) => { + const iso = `${k.slice(1, 5)}-${k.slice(5, 7)}-${k.slice(7, 9)}`; + const cu = epUnits.find((u) => u.episode_number === null && u.air === iso); + return { + unitKey: k, + episode: null, + airDate: iso, + contentUnitId: cu?.id ?? null, + temporary: !cu || cu.status === "provisional", + segment: segOf(cu), + edition: null, + }; + }); + seasons.push({ + seasonNumber: null, + seasonUnitId: null, + total: null, + totalKnown: false, + source: null, + locked: false, + columns, + }); + for (const c of columns) colInputs.push({ unitKey: c.unitKey, airDate: c.airDate }); + } + if (sub.media_kind === "series" && scope.length === 0) totalsKnown = false; + } + + const movieInflight = (plans: { id: string; stage: string; unitKeys: string[] }[], required: string[]) => + plans.map((pl) => ({ ...pl, unitKeys: pl.unitKeys.length ? pl.unitKeys : required.map(editionKey) })); + const movieCovered = (assetSet: Iterable): Map => { + const m = new Map(); + for (const id of assetSet) { + const a = assetById.get(id); + if (!a) continue; + const k = editionKey(a.content_edition ?? "default"); + const q = a.quality as { pri?: unknown } | null; + if (!m.has(k)) + m.set(k, { assetId: id, quality: qualityOf(a.quality), pri: typeof q?.pri === "number" ? q.pri : null }); + } + return m; + }; + + const rows: RowModel[] = []; + if (entity) { + const allRequired = [...new Set(models.flatMap((m) => m.requiredEditions))]; + const any = coverageRow({ + columns: colInputs, + covered: isMovie ? movieCovered(entityAssets) : anyCovered, + inflight: isMovie ? movieInflight(allInflight, allRequired.length ? allRequired : ["default"]) : allInflight, + deferredKeys: subSlots.flatMap((s) => deferBySlot.get(s.id) ?? []), + ...(range ? { range: range as UnitRange } : {}), + today, + }); + rows.push({ + slotId: null, + name: "库内任一版本", + primary: false, + autoFill: false, + status: "", + summary: any.summary, + cells: any.cells, + minPri: any.minPri, + upgrade: { enabled: false, targetPri: 100 }, + ruleGroupIds: [], + }); + for (const m of models) { + const r = coverageRow({ + columns: colInputs, + covered: isMovie ? movieCovered(assetSetOf(m.row)) : slotCovered.get(m.row.id)!, + inflight: isMovie + ? movieInflight(inflightBySlot.get(m.row.id) ?? [], m.requiredEditions) + : (inflightBySlot.get(m.row.id) ?? []), + deferredKeys: deferBySlot.get(m.row.id) ?? [], + ...(range ? { range: range as UnitRange } : {}), + today, + upgrade: m.upgrade, + }); + if (isMovie) { + // 电影:该版本目标不要求的内容版本不算缺 + for (const c of r.cells) + if (c.state === "missing" && !m.requiredEditions.includes(c.unitKey.slice("edition:".length))) { + c.state = "skipped"; + c.reason = "not_required"; + r.summary.missing--; + r.summary.skipped++; + } + } + rows.push({ + slotId: m.row.id, + name: m.row.name, + primary: m.primary, + autoFill: m.autoFill, + status: m.row.status, + summary: r.summary, + cells: r.cells, + minPri: r.minPri, + upgrade: m.upgrade, + ruleGroupIds: m.ruleGroupIds, + }); + } + } + out.set(sub.id, { + sub, + entity, + seasonNumber, + seasons, + rows, + totalsKnown, + lastPlanAt, + autoFill: { enabled: af.enabled, mode: af.mode, source: af.source, budget: af.budget }, + ruleGroupIds, + defaultRuleGroups: !explicitGroups.length && !slotRuleIds.length, + }); + } + return out; +} + +// ── 覆盖矩阵 ── + +export function coverageOf(m: SubscriptionModel, now: Date): SubscriptionCoverage { + return { + subscriptionId: m.sub.id, + mediaKind: m.sub.media_kind, + computedAt: now.toISOString(), + seasons: m.seasons, + rows: m.rows.map((r) => ({ + slotId: r.slotId, + name: r.name, + primary: r.primary, + autoFill: r.autoFill, + summary: r.summary, + cells: r.cells, + })), + }; +} + +const coverageCache = new Map(); + +/** 订阅写入(暂停/恢复/改 params/排序/补全)后清掉该订阅的矩阵缓存 */ +export function invalidateCoverage(subId?: string): void { + if (subId) coverageCache.delete(subId); + else coverageCache.clear(); +} + +/** 覆盖矩阵(缓存 30s;refresh 强制重算) */ +export async function subscriptionCoverage( + exec: Exec, + subId: string, + now: Date, + opts: { refresh?: boolean } = {}, +): Promise { + const hit = coverageCache.get(subId); + if (!opts.refresh && hit && Date.now() - hit.at < COVERAGE_TTL_MS) return hit.value; + const sub = await exec.selectFrom("subscription").selectAll().where("id", "=", subId).executeTakeFirst(); + if (!sub) return null; + const m = (await buildSubscriptionModels(exec, [sub], now)).get(sub.id)!; + const value = coverageOf(m, now); + coverageCache.set(subId, { at: Date.now(), value }); + if (coverageCache.size > 500) { + const oldest = [...coverageCache.entries()].sort((a, b) => a[1].at - b[1].at)[0]; + if (oldest) coverageCache.delete(oldest[0]); + } + return value; +} + +// ── 总览 ── + +export function slotSummaries(m: SubscriptionModel): SubscriptionSlotSummary[] { + return m.rows + .filter((r) => r.slotId !== null) + .map((r) => { + const upgrading = r.upgrade.enabled && r.cells.some((c) => c.upgradable); + return { + slotId: r.slotId!, + name: r.name, + primary: r.primary, + status: r.status, + have: r.summary.have, + inflight: r.summary.inflight, + missing: r.summary.missing, + total: m.totalsKnown ? r.summary.have + r.summary.inflight + r.summary.missing : null, + upgrading, + minPri: r.upgrade.enabled ? r.minPri : null, + targetPri: r.upgrade.enabled ? r.upgrade.targetPri : null, + autoFill: r.autoFill, + ruleGroupIds: r.ruleGroupIds, + }; + }); +} + +export function overviewGroup(m: SubscriptionModel): { group: SubscriptionGroup | null; awaitingNewSeason: boolean } { + const slotRows = m.rows.filter((r) => r.slotId !== null); + const any = m.rows.find((r) => r.slotId === null); + const g = subscriptionGroup({ + status: m.sub.status, + mediaKind: m.sub.media_kind, + mode: m.sub.mode, + totalsKnown: m.totalsKnown, + anyProgress: !!any && any.summary.have + any.summary.inflight > 0, + allSlotsComplete: + m.totalsKnown && + slotRows.length > 0 && + slotRows.every((r) => r.summary.missing === 0 && r.summary.inflight === 0 && r.summary.have > 0), + }); + return { group: g.group, awaitingNewSeason: g.awaitingNewSeason }; +} + +export function lastUpdatedOf(m: SubscriptionModel): Date { + let t = m.sub.last_progress_at ?? m.sub.created_at; + if (m.lastPlanAt && m.lastPlanAt > t) t = m.lastPlanAt; + return t; +} diff --git a/apps/server/src/modules/acquisition-organize/services/evaluate-autofill.ts b/apps/server/src/modules/acquisition-organize/services/evaluate-autofill.ts new file mode 100644 index 00000000..194294ef --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/services/evaluate-autofill.ts @@ -0,0 +1,123 @@ +import type { AcquisitionPriority } from "@umt/contracts"; +import type { AcquisitionSlotRow, SubscriptionRow } from "@umt/db"; +import { sql } from "@umt/db"; +import { checkFillBudget, parseUnitKey } from "@umt/domain"; +import { startOfDay } from "../util.ts"; +import type { Exec } from "./common.ts"; +import { slotOrder, subscriptionAutoFill } from "./coverage.ts"; +import type { RuleCatalog } from "./rules.ts"; +import { slotCoveredUnits } from "./subscriptions.ts"; + +// W9-E 多版本自动补全(docs/design/subscriptions.md §2.2),在 evaluateForSlot 建计划前调用: +// - 主版本(同订阅 position 最小)照旧; +// - 非主版本:自动补全关 → 不建计划,候选记"延后(未开自动补全)"; +// 开且计划单元都已被主版本满足 → 补全计划:优先级 upgrade(低于追更与新作品),受订阅补全预算约束,超出记延后; +// 主版本也缺的单元照常追更(优先级不变)。 +// - 手动补全(POST /subscriptions/{id}/slots/{slotId}/fill)不受开关与补全预算限制,可限定单元。 + +export type AutoFillCheck = + | { action: "normal"; planUnits: string[] } + | { + action: "fill"; + planUnits: string[]; + priority: AcquisitionPriority; + evaluation: Record; + } + | { action: "defer"; reason: "auto_fill_off" | "fill_budget"; record: boolean; detail: string[] } + | { action: "skip"; reason: string }; + +/** 手动补全参数(经 evaluateAcquisition 输入传入) */ +export interface ForceFill { + unitKeys?: string[] | undefined; +} + +/** 今天该订阅已建的自动补全计划用量(单元数 + 预留字节;草案/取消/失败/过期不计) */ +export async function fillUsageToday(exec: Exec, subId: string, now: Date) { + const r = await sql<{ units: number; bytes: string | number }>` + select coalesce(sum(greatest(cardinality(p.unit_keys), 1)), 0)::int as units, + coalesce(sum(p.reserved_bytes), 0)::bigint as bytes + from acquisition_plan p join acquisition_slot s on s.id = p.slot_id + where s.subscription_id = ${subId} and p.created_at >= ${startOfDay(now)} + and p.status not in ('draft', 'cancelled', 'failed', 'expired') + and p.evaluation->>'autoFill' = 'true'`.execute(exec); + const row = r.rows[0]; + return { unitsToday: row?.units ?? 0, bytesToday: Number(row?.bytes ?? 0) }; +} + +const sameUnit = (a: string, b: string) => { + if (a === b) return true; + const pa = parseUnitKey(a); + const pb = parseUnitKey(b); + return !!pa && !!pb && pa.season === pb.season && pa.episode !== undefined && pa.episode === pb.episode; +}; + +export async function autoFillCheck( + exec: Exec, + input: { + slot: AcquisitionSlotRow; + sub: SubscriptionRow | undefined; + isMovie: boolean; + planUnits: string[]; + releaseBytes: number | null; + catalog: RuleCatalog; + now: Date; + force?: ForceFill | undefined; + }, +): Promise { + const { slot, sub } = input; + let planUnits = input.planUnits; + if (input.force?.unitKeys?.length && !input.isMovie) { + const want = input.force.unitKeys; + planUnits = planUnits.filter((k) => want.some((w) => sameUnit(k, w))); + if (!planUnits.length) return { action: "skip", reason: "fill_units_not_requested" }; + } + if (!sub) return { action: "normal", planUnits }; + const slots = ( + await exec + .selectFrom("acquisition_slot") + .selectAll() + .where("media_entity_id", "=", slot.media_entity_id) + .where((eb) => eb.or([eb("subscription_id", "=", sub.id), eb("subscription_id", "is", null)])) + .execute() + ).sort(slotOrder); + const primary = slots[0]; + if (!primary || slots.length <= 1 || primary.id === slot.id) return { action: "normal", planUnits }; + if (input.force) return { action: "normal", planUnits }; + + const af = subscriptionAutoFill(input.catalog, sub); + if (!af.enabled) + return { + action: "defer", + reason: "auto_fill_off", + record: false, + detail: [`auto_fill:${af.source}`], + }; + let primaryCovers: boolean; + if (input.isMovie) { + primaryCovers = + primary.current_asset_id !== null || primary.status === "satisfied" || primary.status === "upgrade_pending"; + } else { + const cov = await slotCoveredUnits(exec, primary); + primaryCovers = planUnits.length > 0 && planUnits.every((k) => cov.keys.has(k)); + } + if (!primaryCovers) return { action: "normal", planUnits }; + + const usage = await fillUsageToday(exec, sub.id, input.now); + const b = checkFillBudget(af.budget, usage, { + bytes: input.releaseBytes, + units: input.isMovie ? 1 : planUnits.length, + }); + if (!b.allow) + return { + action: "defer", + reason: "fill_budget", + record: true, + detail: b.violations.map((v) => `fill_budget.${v.kind}:${v.current}+${v.requested}>${v.limit}`), + }; + return { + action: "fill", + planUnits, + priority: "upgrade", + evaluation: { autoFill: true, primarySlotId: primary.id }, + }; +} diff --git a/apps/server/src/modules/acquisition-organize/services/subscriptions-params.ts b/apps/server/src/modules/acquisition-organize/services/subscriptions-params.ts new file mode 100644 index 00000000..dbf34759 --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/services/subscriptions-params.ts @@ -0,0 +1,260 @@ +import { type SubscriptionMediaKind, SubscriptionParams } from "@umt/contracts"; +import { type SlotSpec, validateKeywordPattern } from "@umt/domain"; +import { ApiException } from "../../../http/errors.ts"; +import { jsonb } from "../util.ts"; +import type { Exec } from "./common.ts"; + +// W9-E 订阅编辑表单字段(docs/design/subscriptions.md §3):params 规范化与校验。 +// 写入时兼容 MP 订阅字段与旧键,规范化后只存 SubscriptionParams 列出的键(其余未知键原样保留)。 +// sites → siteIds(uuid 或站点 key) downloader → downloaderInstanceId(uuid 或实例名) +// save_path / savePath → downloadSavePath start_episode / end_episode → episodeRange +// total_episode (+ manual_total_episode) → 保存后写季总集数(manual 锁定) +// include / exclude(字符串或数组)→ include[] / exclude[](正则白名单校验) +// keyword / keywords(字符串)→ keywords[] filter_groups(组名或 id)→ ruleGroupIds +// best_version / best_version_full → 订阅版本目标的升级策略(按集 / 整季;电影 = 单片) +// autoFillVersions{enabled, budget{perDay, bytes}} / autoFill:boolean → autoFill + fillBudget{unitsPerDay, bytesPerDay} + +const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; + +export type UpgradeStrategy = "off" | "per_unit" | "full_pack" | "movie"; + +export interface NormalizedParams { + params: SubscriptionParams; + /** MP total_episode:保存后写入订阅季的总集数 */ + episodeTotal?: { total: number; locked: boolean }; + /** MP best_version:写到订阅版本目标的升级策略 */ + upgrade?: UpgradeStrategy; + /** 被改写的兼容字段(调试/审计用) */ + mapped: string[]; +} + +const bad = (msg: string): never => { + throw new ApiException("bad_request", msg); +}; + +const truthy = (v: unknown) => v === true || v === 1 || v === "1" || v === "true"; +const list = (v: unknown): unknown[] => (Array.isArray(v) ? v : v === undefined || v === null || v === "" ? [] : [v]); +const strList = (v: unknown, key: string): string[] => + list(v).map((x) => { + if (typeof x !== "string" && typeof x !== "number") bad(`${key} 只能是字符串或字符串数组`); + return String(x).trim(); + }); + +/** 订阅级 include/exclude 并入版本目标硬条件(作用于全部版本目标) */ +export function withSubscriptionFilters(spec: SlotSpec, params: SubscriptionParams): SlotSpec { + const inc = params.include ?? []; + const exc = params.exclude ?? []; + if (!inc.length && !exc.length) return spec; + return { + ...spec, + hard: { + ...spec.hard, + ...(inc.length ? { include: [...(spec.hard.include ?? []), ...inc] } : {}), + ...(exc.length ? { exclude: [...(spec.hard.exclude ?? []), ...exc] } : {}), + }, + }; +} + +async function idsByKey( + exec: Exec, + table: "site" | "downloader_instance" | "rule_group", + values: string[], + label: string, +): Promise { + if (!values.length) return []; + const uuids = values.filter((v) => UUID.test(v)); + const names = values.filter((v) => !UUID.test(v)); + const none = ["00000000-0000-0000-0000-000000000000"]; + const ids = uuids.length ? uuids : none; + const keys = names.length ? names : [""]; + const rows: { id: string; k: string }[] = + table === "site" + ? await exec + .selectFrom("site") + .select(["id", "key as k"]) + .where((eb) => eb.or([eb("id", "in", ids), eb("key", "in", keys)])) + .execute() + : table === "downloader_instance" + ? await exec + .selectFrom("downloader_instance") + .select(["id", "name as k"]) + .where((eb) => eb.or([eb("id", "in", ids), eb("name", "in", keys)])) + .execute() + : await exec + .selectFrom("rule_group") + .select(["id", "name as k"]) + .where("deleted_at", "is", null) + .where((eb) => eb.or([eb("id", "in", ids), eb("name", "in", keys)])) + .execute(); + return values.map((v) => { + const hit = rows.find((r) => r.id === v || r.k === v); + if (!hit) bad(`${label}不存在:${v}`); + return hit!.id; + }); +} + +/** + * 规范化 + 校验订阅 params(POST / PATCH 共用)。未知键原样保留;兼容字段改写后删除原键。 + */ +export async function normalizeSubscriptionParams( + exec: Exec, + raw: Record | undefined, + opts: { mediaKind: SubscriptionMediaKind }, +): Promise { + const src: Record = { ...(raw ?? {}) }; + const mapped: string[] = []; + const take = (k: string) => { + const v = src[k]; + delete src[k]; + if (v !== undefined) mapped.push(k); + return v; + }; + const out: NormalizedParams = { params: {} as SubscriptionParams, mapped }; + + // ── MP / 旧字段 → 规范键 ── + const sites = take("sites"); + if (sites !== undefined && src.siteIds === undefined) src.siteIds = strList(sites, "sites"); + const downloader = take("downloader"); + if (downloader !== undefined && downloader !== null && downloader !== "" && src.downloaderInstanceId === undefined) + src.downloaderInstanceId = String(downloader); + for (const k of ["save_path", "savePath"]) { + const v = take(k); + if (typeof v === "string" && v && src.downloadSavePath === undefined) src.downloadSavePath = v; + } + const startEp = take("start_episode"); + const endEp = take("end_episode"); + if (startEp !== undefined || endEp !== undefined) { + const r = { ...((src.episodeRange as object | undefined) ?? {}) } as { start?: number; end?: number }; + if (startEp !== undefined && startEp !== null && Number(startEp) > 0) r.start = Number(startEp); + if (endEp !== undefined && endEp !== null && Number(endEp) > 0) r.end = Number(endEp); + src.episodeRange = r; + } + if (src.episodeRange && typeof src.episodeRange === "object") { + // 旧键 locked 等只作记录,范围只保留 start/end + const r = src.episodeRange as { start?: unknown; end?: unknown }; + src.episodeRange = { + ...(r.start !== undefined && r.start !== null ? { start: Number(r.start) } : {}), + ...(r.end !== undefined && r.end !== null ? { end: Number(r.end) } : {}), + }; + } + const total = take("total_episode"); + const manualTotal = take("manual_total_episode"); + if (total !== undefined && total !== null && Number(total) > 0) { + if (opts.mediaKind === "movie") bad("电影订阅没有总集数"); + out.episodeTotal = { total: Math.trunc(Number(total)), locked: truthy(manualTotal) }; + } + for (const k of ["include", "exclude"] as const) { + const v = src[k]; + if (typeof v === "string") src[k] = v.trim() ? [v.trim()] : []; + // MP 正则常带 (?i):统一不区分大小写,去掉 + if (Array.isArray(src[k])) + src[k] = (src[k] as unknown[]).map((x) => (typeof x === "string" ? x.replace(/^\(\?i\)/, "") : x)); + } + const keyword = take("keyword"); + if (keyword !== undefined && src.keywords === undefined) src.keywords = strList(keyword, "keyword").filter(Boolean); + if (typeof src.keywords === "string") src.keywords = [src.keywords]; + const filterGroups = take("filter_groups"); + if (filterGroups !== undefined && src.ruleGroupIds === undefined) + src.ruleGroupIds = strList(filterGroups, "filter_groups"); + const best = take("best_version"); + const bestFull = take("best_version_full"); + if (best !== undefined || bestFull !== undefined) { + out.upgrade = !truthy(best) + ? "off" + : opts.mediaKind === "movie" + ? "movie" + : truthy(bestFull) + ? "full_pack" + : "per_unit"; + } + const legacyFill = take("autoFillVersions") as + | { enabled?: unknown; budget?: { perDay?: unknown; bytes?: unknown } } + | undefined; + if (legacyFill && typeof legacyFill === "object") { + if (src.autoFill === undefined && legacyFill.enabled !== undefined) + src.autoFill = truthy(legacyFill.enabled) ? "on" : "off"; + if (src.fillBudget === undefined && legacyFill.budget) { + const b = legacyFill.budget; + src.fillBudget = { + ...(b.perDay !== undefined && b.perDay !== null ? { unitsPerDay: Number(b.perDay) } : {}), + ...(b.bytes !== undefined && b.bytes !== null ? { bytesPerDay: Number(b.bytes) } : {}), + }; + } + } + if (typeof src.autoFill === "boolean") src.autoFill = src.autoFill ? "on" : "off"; + if (src.fillBudget && typeof src.fillBudget === "object") { + const b = src.fillBudget as Record; + const nb: Record = { ...b }; + if (b.perDay !== undefined) { + nb.unitsPerDay ??= b.perDay; + delete nb.perDay; + } + if (b.bytes !== undefined) { + nb.bytesPerDay ??= b.bytes; + delete nb.bytes; + } + src.fillBudget = nb; + } + // 空值 = 不设 + for (const [k, v] of Object.entries(src)) if (v === null || v === "") delete src[k]; + + // 引用存在性(站点可给 key、下载器/规则组可给名称,统一换成 id) + if (Array.isArray(src.siteIds)) src.siteIds = await idsByKey(exec, "site", strList(src.siteIds, "siteIds"), "站点"); + if (typeof src.downloaderInstanceId === "string") + src.downloaderInstanceId = (await idsByKey(exec, "downloader_instance", [src.downloaderInstanceId], "下载器"))[0]; + if (Array.isArray(src.ruleGroupIds)) + src.ruleGroupIds = await idsByKey(exec, "rule_group", strList(src.ruleGroupIds, "ruleGroupIds"), "规则组"); + + const parsed = SubscriptionParams.safeParse(src); + if (!parsed.success) { + const i = parsed.error.issues[0]; + bad(`params.${i?.path.join(".") ?? ""} 无效:${i?.message ?? ""}`); + } + const p = parsed.data as SubscriptionParams; + + // ── 库根存在性与正则白名单 ── + if (p.libraryRootId) { + const r = await exec.selectFrom("library_root").select("id").where("id", "=", p.libraryRootId).executeTakeFirst(); + if (!r) bad(`库根不存在:${p.libraryRootId}`); + } + for (const k of ["include", "exclude"] as const) + for (const pat of p[k] ?? []) { + const err = validateKeywordPattern(pat); + if (err) bad(`${k} 正则不可用(${err}):${pat}`); + try { + new RegExp(pat, "i"); + } catch { + bad(`${k} 正则语法错误:${pat}`); + } + } + if (p.stopAfter && Number.isNaN(Date.parse(p.stopAfter))) bad("stopAfter 不是有效时间"); + const r = p.episodeRange; + if (r?.start !== undefined && r.end !== undefined && r.end < r.start) bad("结束集不能小于开始集"); + if (opts.mediaKind === "movie" && r) delete p.episodeRange; + out.params = p; + return out; +} + +/** best_version 等 → 订阅版本目标的升级策略(保留其余升级参数;revision+1) */ +export async function applySlotUpgrade( + exec: Exec, + input: { subscriptionId: string; entityId: string; strategy: UpgradeStrategy }, +): Promise { + const slots = await exec + .selectFrom("acquisition_slot") + .select(["id", "upgrade_threshold", "revision"]) + .where("media_entity_id", "=", input.entityId) + .where("subscription_id", "=", input.subscriptionId) + .execute(); + for (const s of slots) { + const cur = (s.upgrade_threshold ?? {}) as Record; + const next = + input.strategy === "off" ? { ...cur, enabled: false } : { ...cur, enabled: true, mode: input.strategy }; + await exec + .updateTable("acquisition_slot") + .set({ upgrade_threshold: jsonb(next), revision: s.revision + 1 }) + .where("id", "=", s.id) + .execute(); + } + return slots.length; +} diff --git a/apps/server/src/modules/acquisition-organize/services/subscriptions.ts b/apps/server/src/modules/acquisition-organize/services/subscriptions.ts index 34fbb703..695e0f0a 100644 --- a/apps/server/src/modules/acquisition-organize/services/subscriptions.ts +++ b/apps/server/src/modules/acquisition-organize/services/subscriptions.ts @@ -131,6 +131,7 @@ export async function applyEpisodeTotal( .forUpdate() .executeTakeFirstOrThrow(); if (input.manual) { + if (input.manual.total !== cur.episode_total) await touchProgress(exec, { entityId: input.entityId }, input.now); await exec .updateTable("content_unit") .set({ @@ -161,6 +162,7 @@ export async function applyEpisodeTotal( .set({ episode_total: r.total, episode_total_source: r.source, episode_total_at: input.now }) .where("id", "=", unitId) .execute(); + await touchProgress(exec, { entityId: input.entityId }, input.now); } return { seasonUnitId: unitId, total: r.total, known: r.known, changed: r.changed }; } @@ -457,6 +459,13 @@ export async function afterSlotVerified( } } + if (input.verifiedAssets.length) + await touchProgress( + trx, + slot.subscription_id ? { subscriptionId: slot.subscription_id } : { entityId: slot.media_entity_id }, + now, + ); + if (input.completion.suspectedMissing.length) { const desc = input.completion.suspectedMissing .map((s) => `S${String(s.season).padStart(2, "0")}:${s.episodes.map((e) => `E${e}`).join(",")}`) @@ -488,6 +497,23 @@ export async function afterSlotVerified( await markSlotDue(trx, slot.id, now); } +/** + * 订阅"上次进展"(总览上次更新):单元入库、总集数变化时写;计划预留时间由总览查询合并。 + * entityId 形式:作品下所有未取消订阅。 + */ +export async function touchProgress( + exec: Exec, + target: { subscriptionId: string } | { entityId: string }, + now: Date, +): Promise { + let q = exec.updateTable("subscription").set({ last_progress_at: now }); + q = + "subscriptionId" in target + ? q.where("id", "=", target.subscriptionId) + : q.where("media_entity_id", "=", target.entityId).where("status", "!=", "cancelled"); + await q.execute(); +} + /** Slot 标记立即到期重评(事务内写,由编排循环 sweepDueSlots 消费) */ export async function markSlotDue(exec: Exec, slotId: string, now: Date): Promise { await exec.updateTable("acquisition_slot").set({ next_evaluate_at: now }).where("id", "=", slotId).execute(); diff --git a/packages/contracts/src/api.ts b/packages/contracts/src/api.ts index 29f4c27c..23b351fe 100644 --- a/packages/contracts/src/api.ts +++ b/packages/contracts/src/api.ts @@ -629,6 +629,89 @@ export const Api = { }, R, ), + /** 总览:在 crud 列表上加分组/排序/筛选,条目带卡片派生字段,另给各分组计数(docs/design/subscriptions.md §1) */ + list: ep({ + method: "get", + path: "/subscriptions", + scope: "read", + summary: + "订阅总览(分组、进度、各版本摘要、上次更新、下次搜索、站点、规则组;cancelled 只在显式 status=cancelled 时返回)", + query: listQuery({ + status: policy.SubscriptionStatus.optional(), + mediaEntityId: Id.optional(), + mediaKind: policy.SubscriptionMediaKind.optional(), + /** 作品标题 / 原名 / 别名 / 未建档检索词,模糊匹配 */ + q: z.string().max(200).optional(), + group: policy.SubscriptionGroup.optional(), + sort: policy.SubscriptionSort.optional(), + order: z.enum(["asc", "desc"]).optional(), + siteId: Id.optional(), + ruleGroupId: Id.optional(), + origin: policy.SubscriptionOrigin.optional(), + upgrading: z.enum(["true", "false"]).optional(), + }), + response: policy.SubscriptionOverviewPage, + }), + coverage: ep({ + method: "get", + path: "/subscriptions/{id}/coverage", + scope: "read", + summary: + "覆盖矩阵:季 → 集 × 版本目标(电影:内容版本 × 版本目标),每格 have/inflight/missing/skipped;缓存 30s,refresh=true 强制重算", + params: IdParam, + query: z.object({ refresh: z.enum(["true", "false"]).optional() }), + response: policy.SubscriptionCoverage, + }), + pause: ep({ + method: "post", + path: "/subscriptions/{id}/pause", + scope: "admin", + summary: "暂停订阅(不评估、不搜索)", + params: IdParam, + response: policy.Subscription, + }), + resume: ep({ + method: "post", + path: "/subscriptions/{id}/resume", + scope: "admin", + summary: "恢复已暂停的订阅(TMDB 未就绪回到 pending),立即重评并排上搜索", + params: IdParam, + response: policy.Subscription, + }), + resubscribe: ep({ + method: "post", + path: "/subscriptions/{id}/resubscribe", + scope: "admin", + summary: "重新订阅已完成/已取消的订阅(历史保留),立即重评并搜索", + params: IdParam, + response: policy.Subscription, + }), + batch: ep({ + method: "post", + path: "/subscriptions/batch", + scope: "admin", + summary: "批量:立即搜索 / 暂停 / 恢复 / 删除(删除 = 取消,写审计)", + body: policy.SubscriptionBatch, + response: policy.SubscriptionBatchResult, + }), + fillSlot: ep({ + method: "post", + path: "/subscriptions/{id}/slots/{slotId}/fill", + scope: "admin", + summary: "手动补全某版本目标(不受自动补全开关与补全预算限制)", + params: z.object({ id: Id, slotId: Id }), + body: policy.SlotFillRequest, + response: z.object({ runId: z.string(), deduplicated: z.boolean() }), + }), + slotOrder: ep({ + method: "put", + path: "/subscriptions/{id}/slot-order", + scope: "admin", + summary: "版本目标排序(第一个为主版本)", + params: IdParam, + body: policy.SlotOrderSet, + response: z.array(policy.AcquisitionSlot), + }), search: ep({ method: "post", path: "/subscriptions/{id}/search", diff --git a/packages/contracts/src/policy.ts b/packages/contracts/src/policy.ts index 6934cb02..abe03987 100644 --- a/packages/contracts/src/policy.ts +++ b/packages/contracts/src/policy.ts @@ -79,14 +79,32 @@ export const Subscription = EntityBase.extend({ /** 上次主动搜索时间 / 下次到期(按 params.searchIntervalHours,缺省 24h) */ lastSearchedAt: Timestamp.nullable(), nextSearchAt: Timestamp.nullable(), + /** 上次进展:任一版本目标的单元转为已入库、或总集数变化时写入;总览"上次更新"另合并计划预留时间 */ + lastProgressAt: Timestamp.nullable(), /** 派生字段(列表返回):作品标题,未建档时为 null */ mediaEntityTitle: z.string().nullable().optional(), }); export type Subscription = z.infer; +/** 自动补全其他版本:inherit = 取引用规则组(后者优先)的 autoFill,都没设为开 */ +export const AutoFillMode = z.enum(["inherit", "on", "off"]); +export type AutoFillMode = z.infer; +/** 补全预算:只约束非主版本的自动补全计划(按订阅、按自然日);同时仍计入全局每日预算 */ +export const FillBudget = z + .object({ + bytesPerDay: z.number().int().min(0).optional(), + unitsPerDay: z.number().int().min(0).optional(), + }) + .strict(); +export type FillBudget = z.infer; + /** * subscription.params 的已知键(JSON 列,其余键原样保留): * siteIds 站点范围;episodeRange 集范围(对照 MP start/total_episode);searchIntervalHours 主动搜索周期(0 = 关); - * stopAfter 截止时间;downloaderInstanceId 指定下载器。 + * stopAfter 截止时间;downloaderInstanceId 指定下载器;ruleGroupIds 规则组;keywords 追加检索词; + * include/exclude 订阅级关键词正则(作用于全部版本目标);libraryRootId 库根;downloadSavePath 下载保存路径; + * autoFill/fillBudget 多版本自动补全。 + * 写入时兼容 MP 字段与旧键(sites/downloader/save_path/start_episode/filter_groups/keyword/autoFillVersions…), + * 服务端规范化后只存本表列出的键,见 docs/runbook/subscriptions.md。 */ export const SubscriptionParams = z .object({ @@ -103,6 +121,15 @@ export const SubscriptionParams = z downloaderInstanceId: Id.optional(), /** 引用的规则组(有序,串联取交集,pri 取最后一个带层的组);空 = 订阅默认组 */ ruleGroupIds: z.array(Id).optional(), + /** 追加到标题/别名之后的检索词(MP keyword) */ + keywords: z.array(z.string().min(1).max(100)).max(10).optional(), + /** 订阅级包含/排除正则(MP include/exclude;不区分大小写,匹配 标题+副标题+标签) */ + include: z.array(z.string().min(1).max(200)).max(20).optional(), + exclude: z.array(z.string().min(1).max(200)).max(20).optional(), + libraryRootId: Id.optional(), + downloadSavePath: z.string().min(1).max(1000).optional(), + autoFill: AutoFillMode.optional(), + fillBudget: FillBudget.optional(), }) .passthrough(); export type SubscriptionParams = z.infer; @@ -233,6 +260,176 @@ export const EpisodeTotalSet = z.object({ }); export type EpisodeTotalSet = z.infer; +// ── 订阅总览 / 覆盖矩阵(W9-E,docs/design/subscriptions.md §1–§2) ── + +/** + * 总览分组(派生,按顺序判定):paused 暂停 → pending 待定(未建档/无 TMDB 身份,或范围内有季总集数未知) + * → completed 已完成(含 continuous 订阅范围内全覆盖且总集数已知 = 等待新季)→ following 追更中(有已入库或在途单元) + * → waiting 待搜索(没有任何已入库/在途单元)。cancelled 不进总览分组。 + */ +export const SubscriptionGroup = z.enum(["following", "waiting", "completed", "paused", "pending"]); +export type SubscriptionGroup = z.infer; +export const SubscriptionSort = z.enum(["lastUpdated", "nextSearch", "missing", "created", "name"]); +export type SubscriptionSort = z.infer; + +/** 版本目标在订阅范围内的计数摘要(卡片徽标 / 矩阵行尾) */ +export const SubscriptionSlotSummary = z.object({ + slotId: Id, + name: z.string(), + /** 同订阅内顺序最小的版本目标 */ + primary: z.boolean(), + status: z.string(), + have: z.number().int(), + inflight: z.number().int(), + missing: z.number().int(), + /** 范围内单元总数;任一季总集数未知为 null */ + total: z.number().int().nullable(), + /** 开了升级且有已入库单元低于目标 pri */ + upgrading: z.boolean(), + /** 已入库单元的最低 pri 与目标 pri(未开升级 / 无阶梯为 null) */ + minPri: z.number().int().nullable(), + targetPri: z.number().int().nullable(), + /** 非主版本是否自动补全(主版本恒为 true) */ + autoFill: z.boolean(), + ruleGroupIds: z.array(Id), +}); +export type SubscriptionSlotSummary = z.infer; + +/** 总览卡片:订阅行 + 派生字段 */ +export const SubscriptionOverviewItem = Subscription.extend({ + group: SubscriptionGroup.nullable(), + /** 作品年份 */ + year: z.number().int().nullable(), + /** 剧集季:季单元季号;following_seasons 显示"S{n} 起";null = 全部季 / 电影 */ + seasonNumber: z.number().int().nullable(), + /** 主版本进度:已入库/总(总未知为 null)与在途 */ + progress: z.object({ have: z.number().int(), total: z.number().int().nullable(), inflight: z.number().int() }), + /** 主版本范围内缺失单元数(排序"缺集数"用) */ + missing: z.number().int(), + slots: z.array(SubscriptionSlotSummary), + /** 上次更新 = max(last_progress_at ?? created_at, 最近计划预留时间) */ + lastUpdatedAt: Timestamp, + /** 周期搜索关闭(searchIntervalHours = 0)时为 false */ + searchEnabled: z.boolean(), + /** 站点范围;空 = 全部启用站点 */ + sites: z.array(z.object({ id: Id, name: z.string() })), + /** 生效的规则组;defaultRuleGroups = true 表示取的是订阅默认链 */ + ruleGroups: z.array(z.object({ id: Id, name: z.string() })), + defaultRuleGroups: z.boolean(), + /** 订阅级自动补全(有效值与来源) */ + autoFill: z.object({ + enabled: z.boolean(), + mode: AutoFillMode, + source: z.enum(["subscription", "rule_group", "default"]), + budget: FillBudget, + }), + /** 任一版本目标开了升级且未达顶档 */ + upgrading: z.boolean(), + /** 已完成分组里 continuous 订阅在等下一季 */ + awaitingNewSeason: z.boolean(), +}); +export type SubscriptionOverviewItem = z.infer; +export const SubscriptionOverviewPage = z.object({ + items: z.array(SubscriptionOverviewItem), + nextCursor: z.string().nullable(), + /** 当前筛选(不含 group)下各分组数量 */ + counts: z.record(SubscriptionGroup, z.number().int()), + total: z.number().int(), +}); +export type SubscriptionOverviewPage = z.infer; + +/** 覆盖矩阵格状态:have 已入库;inflight 在途;missing 缺失;skipped 不需要(范围外 / 未播出) */ +export const CoverageCellState = z.enum(["have", "inflight", "missing", "skipped"]); +export type CoverageCellState = z.infer; +export const CoverageQuality = z.object({ + resolution: z.string().nullable(), + source: z.string().nullable(), + /** 如 "2160p REMUX" */ + label: z.string().nullable(), +}); +export const CoverageCell = z.object({ + unitKey: z.string(), + state: CoverageCellState, + assetId: Id.nullable(), + quality: CoverageQuality.nullable(), + pri: z.number().int().nullable(), + planId: Id.nullable(), + /** 在途阶段 reserved / downloading / verifying */ + planStage: z.string().nullable(), + /** skipped 的原因:out_of_range / not_aired;missing 且总集数未知为 suspected */ + reason: z.string().nullable(), + /** 已入库但低于目标 pri(开了升级) */ + upgradable: z.boolean(), + /** 该单元有延后的资源(slot_deferral) */ + deferred: z.boolean(), +}); +export type CoverageCell = z.infer; +/** 矩阵列:剧集 = 集,综艺 = 集或播出日期单元,电影 = 内容版本 */ +export const CoverageColumn = z.object({ + unitKey: z.string(), + episode: z.number().int().nullable(), + airDate: z.string().nullable(), + contentUnitId: Id.nullable(), + /** 临时单元:没有内容单元行(合成键 / 资产上记的旧单元键),或内容单元仍是 provisional */ + temporary: z.boolean(), + /** 综艺分段 upper / middle / lower(整集为 null;分段单元的独立列待身份层按分段建单元后启用) */ + segment: z.string().nullable(), + /** 电影内容版本名 */ + edition: z.string().nullable(), +}); +export type CoverageColumn = z.infer; +export const SubscriptionCoverage = z.object({ + subscriptionId: Id, + mediaKind: SubscriptionMediaKind, + computedAt: Timestamp, + /** 列分段:剧集/综艺按季;电影一段(seasonNumber = null,列 = 内容版本) */ + seasons: z.array( + z.object({ + seasonNumber: z.number().int().nullable(), + seasonUnitId: Id.nullable(), + total: z.number().int().nullable(), + totalKnown: z.boolean(), + source: z.string().nullable(), + locked: z.boolean(), + columns: z.array(CoverageColumn), + }), + ), + /** 首行 slotId = null 为"库内任一版本";其后每个版本目标一行,按 position */ + rows: z.array( + z.object({ + slotId: Id.nullable(), + name: z.string(), + primary: z.boolean(), + autoFill: z.boolean(), + summary: z.object({ + have: z.number().int(), + inflight: z.number().int(), + missing: z.number().int(), + skipped: z.number().int(), + }), + /** 与 seasons[].columns 展平后的顺序一致 */ + cells: z.array(CoverageCell), + }), + ), +}); +export type SubscriptionCoverage = z.infer; + +export const SubscriptionBatch = z.object({ + ids: z.array(Id).min(1).max(200), + action: z.enum(["search", "pause", "resume", "delete"]), +}); +export type SubscriptionBatch = z.infer; +export const SubscriptionBatchResult = z.object({ + results: z.array(z.object({ id: Id, ok: z.boolean(), error: z.string().nullable() })), +}); +export type SubscriptionBatchResult = z.infer; + +/** 手动补全某版本:不受自动补全开关与补全预算限制(仍受全局预算/缓存卷阈值) */ +export const SlotFillRequest = z.object({ unitKeys: z.array(z.string().min(1)).max(500).optional() }); +export type SlotFillRequest = z.infer; +export const SlotOrderSet = z.object({ slotIds: z.array(Id).min(1).max(50) }); +export type SlotOrderSet = z.infer; + /** Slot 状态(REQ §6.1) */ export const SlotStatus = z.enum([ "empty", @@ -268,6 +465,8 @@ export const AcquisitionSlot = EntityBase.extend({ lastEvaluatedAt: Timestamp.nullable(), nextEvaluateAt: Timestamp.nullable(), revision: z.number().int().min(1), + /** 同一订阅内的顺序;最小者为主版本(自动补全只约束非主版本),改顺序用 PUT /subscriptions/{id}/slot-order */ + position: z.number().int().min(0), /** 引用的规则组(有序);空 = 订阅的 ruleGroupIds,再空 = 订阅默认组(开升级时叠加洗版默认组) */ ruleGroupIds: z.array(Id), /** 派生字段(列表返回):当前资产的交付特征摘要,无资产为 null */ diff --git a/packages/db/migrations/0035_subscriptions_v2.ts b/packages/db/migrations/0035_subscriptions_v2.ts new file mode 100644 index 00000000..f6af8434 --- /dev/null +++ b/packages/db/migrations/0035_subscriptions_v2.ts @@ -0,0 +1,48 @@ +import type { Kysely } from "kysely"; +import { execDdl } from "../src/migration-helpers.ts"; + +// W9-E 订阅重做(docs/design/subscriptions.md §1、§2.2、§5): +// - subscription.last_progress_at:总览"上次更新"。任一关联版本目标的单元转为已入库、或总集数变化时写入; +// 计划预留时间由查询侧取 max(acquisition_plan.created_at) 合并,不在这里冗余;空 = 取 created_at。 +// 存量订阅按分集账本与计划时间回填一次。 +// - acquisition_slot.position:同一订阅内版本目标的顺序,最小者为主版本(自动补全只约束非主版本); +// 存量按 created_at 编号,与原"第一个版本目标"口径一致。 + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl( + db, + ` +alter table subscription add column last_progress_at timestamptz; +create index subscription_last_progress_idx on subscription (coalesce(last_progress_at, created_at) desc); + +alter table acquisition_slot add column position integer not null default 0 check (position >= 0); +update acquisition_slot s set position = x.n + from (select id, (row_number() over (partition by media_entity_id, subscription_id order by created_at, id) - 1)::int as n + from acquisition_slot) x + where x.id = s.id; + +update subscription sub set last_progress_at = x.at + from (select s.subscription_id as id, + greatest(max(su.updated_at), max(p.created_at)) as at + from acquisition_slot s + left join slot_unit su on su.slot_id = s.id and su.media_asset_id is not null + left join acquisition_plan p on p.slot_id = s.id and p.status <> 'draft' + where s.subscription_id is not null + group by s.subscription_id) x + where x.id = sub.id and x.at is not null; +`, + ); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl( + db, + ` +alter table acquisition_slot drop column if exists position; +drop index if exists subscription_last_progress_idx; +alter table subscription drop column if exists last_progress_at; +`, + ); +} diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index d8ba1df4..42820c48 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -293,6 +293,7 @@ export interface AcquisitionSlotTable { updated_at: GeneratedTimestamptz; reserved_plan_id: string | null; rule_group_ids: Generated; + position: Generated; } export interface AdminSessionTable { @@ -1442,6 +1443,7 @@ export interface SubscriptionTable { updated_at: GeneratedTimestamptz; last_searched_at: Timestamptz | null; next_search_at: Timestamptz | null; + last_progress_at: Timestamptz | null; } export interface SubscriptionHistoryTable {