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 <[email protected]>
This commit is contained in:
rootandClaude Opus 5.5 committed 2026-09-28 15:54:29 +08:00
1 parent e77fdc6549
commit 410d7fd636
12 files changed
+2158 -252

No files matched your search

@@ -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) =>
@@ -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<Subscription>(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<Subscription>(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<Subscription>(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<Subscription>(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<SubscriptionHistory>(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<SubscriptionHistory>(r ?? notFound("订阅历史"));
},
{ tag: TAG },
);
// ── 订阅(W9-E 起在 ./subscriptions.ts) ──
registerSubscriptionRoutes(app, d);
// ── 收录 Slot ──
mount(
@@ -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<boolean> {
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<AuthEnv>,
action: string,
subId: string,
payload: Record<string, unknown> = {},
) {
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<Subscription>(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<SubscriptionSort, "asc" | "desc"> = {
lastUpdated: "desc",
nextSearch: "asc",
missing: "desc",
created: "desc",
name: "asc",
};
type Item = ReturnType<typeof overviewItem>;
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<SubscriptionOverviewPage> => {
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<SubscriptionGroup, number>;
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<Subscription>(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<Subscription>(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<Subscription>(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<Subscription>(await pause(await load(params.id))),
{
tag: TAG,
},
);
mount(
app,
Api.subscriptions.resume,
async (_c, { params }) => toApi<Subscription>(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<Subscription>(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<AcquisitionSlot>(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<SubscriptionHistory>(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<SubscriptionHistory>(r ?? notFound("订阅历史"));
},
{ tag: TAG },
);
}
@@ -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<void> {
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<string, Promise<string | undefined>>;
/** 现有资产在规则组下的 pri 缓存 */
assetPri?: Map<string, number | undefined>;
forceFill?: ForceFill | undefined;
}
async function note(c: Ctx, base: Omit<CandidateEvaluation, "evaluatedAt">) {
@@ -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<AcquisitionSlot>(slot));
const spec = withSubscriptionFilters(slotSpecFromRow(toApi<AcquisitionSlot>(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<string, unknown>;
},
): 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) {
@@ -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<SubscriptionRow, "id" | "params" | "media_kind" | "media_entity_id">,
): 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<SubscriptionRow, "id" | "media_entity_id">) =>
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<Map<string, SubscriptionModel>> {
const out = new Map<string, SubscriptionModel>();
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<string | null>`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<string | null>`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<string, typeof coverage.rows>();
for (const c of coverage.rows) covByAsset.set(c.asset_id, [...(covByAsset.get(c.asset_id) ?? []), c]);
const ledgerBySlot = new Map<string, typeof ledgerRows>();
for (const r of ledgerRows) ledgerBySlot.set(r.slot_id, [...(ledgerBySlot.get(r.slot_id) ?? []), r]);
const slotsByEntity = new Map<string, AcquisitionSlotRow[]>();
for (const s of slots) slotsByEntity.set(s.media_entity_id, [...(slotsByEntity.get(s.media_entity_id) ?? []), s]);
const inflightBySlot = new Map<string, { id: string; stage: string; unitKeys: string[] }[]>();
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<string, string[]>();
for (const r of deferrals) deferBySlot.set(r.slot_id, [...(deferBySlot.get(r.slot_id) ?? []), ...r.unit_keys]);
const unitsByEntity = new Map<string, typeof units>();
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<string>, ledger: typeof ledgerRows): Map<string, CoveredFact> => {
const m = new Map<string, CoveredFact>();
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<AcquisitionSlot>(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<string> => {
const set = new Set<string>();
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<Date | null>((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<string>(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<number, (typeof eUnits)[number]>();
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<number>([...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<string>) => {
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<string>();
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<string>): Map<string, CoveredFact> => {
const m = new Map<string, CoveredFact>();
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<string, { at: number; value: SubscriptionCoverage }>();
/** 订阅写入(暂停/恢复/改 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<SubscriptionCoverage | null> {
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;
}
@@ -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<string, unknown>;
}
| { 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<AutoFillCheck> {
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 },
};
}
@@ -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<string[]> {
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<string, unknown> | undefined,
opts: { mediaKind: SubscriptionMediaKind },
): Promise<NormalizedParams> {
const src: Record<string, unknown> = { ...(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<string, unknown>;
const nb: Record<string, unknown> = { ...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<number> {
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<string, unknown>;
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;
}
@@ -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<void> {
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<void> {
await exec.updateTable("acquisition_slot").set({ next_evaluate_at: now }).where("id", "=", slotId).execute();
+83
View File
@@ -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",
+200 -1
View File
@@ -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<typeof Subscription>;
/** 自动补全其他版本:inherit = 取引用规则组(后者优先)的 autoFill,都没设为开 */
export const AutoFillMode = z.enum(["inherit", "on", "off"]);
export type AutoFillMode = z.infer<typeof AutoFillMode>;
/** 补全预算:只约束非主版本的自动补全计划(按订阅、按自然日);同时仍计入全局每日预算 */
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<typeof FillBudget>;
/**
* 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<typeof SubscriptionParams>;
@@ -233,6 +260,176 @@ export const EpisodeTotalSet = z.object({
});
export type EpisodeTotalSet = z.infer<typeof EpisodeTotalSet>;
// ── 订阅总览 / 覆盖矩阵(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<typeof SubscriptionGroup>;
export const SubscriptionSort = z.enum(["lastUpdated", "nextSearch", "missing", "created", "name"]);
export type SubscriptionSort = z.infer<typeof SubscriptionSort>;
/** 版本目标在订阅范围内的计数摘要(卡片徽标 / 矩阵行尾) */
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<typeof SubscriptionSlotSummary>;
/** 总览卡片:订阅行 + 派生字段 */
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<typeof SubscriptionOverviewItem>;
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<typeof SubscriptionOverviewPage>;
/** 覆盖矩阵格状态:have 已入库;inflight 在途;missing 缺失;skipped 不需要(范围外 / 未播出) */
export const CoverageCellState = z.enum(["have", "inflight", "missing", "skipped"]);
export type CoverageCellState = z.infer<typeof CoverageCellState>;
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<typeof CoverageCell>;
/** 矩阵列:剧集 = 集,综艺 = 集或播出日期单元,电影 = 内容版本 */
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<typeof CoverageColumn>;
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<typeof SubscriptionCoverage>;
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<typeof SubscriptionBatch>;
export const SubscriptionBatchResult = z.object({
results: z.array(z.object({ id: Id, ok: z.boolean(), error: z.string().nullable() })),
});
export type SubscriptionBatchResult = z.infer<typeof SubscriptionBatchResult>;
/** 手动补全某版本:不受自动补全开关与补全预算限制(仍受全局预算/缓存卷阈值) */
export const SlotFillRequest = z.object({ unitKeys: z.array(z.string().min(1)).max(500).optional() });
export type SlotFillRequest = z.infer<typeof SlotFillRequest>;
export const SlotOrderSet = z.object({ slotIds: z.array(Id).min(1).max(50) });
export type SlotOrderSet = z.infer<typeof SlotOrderSet>;
/** 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 */
@@ -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<any>): Promise<void> {
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<any>): Promise<void> {
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;
`,
);
}
+2
View File
@@ -293,6 +293,7 @@ export interface AcquisitionSlotTable {
updated_at: GeneratedTimestamptz;
reserved_plan_id: string | null;
rule_group_ids: Generated<Json>;
position: Generated<number>;
}
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 {