From 0105e5c6acb8f203eace0f9669c19f4bb73f6c32 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 28 Sep 2026 15:08:00 +0800 Subject: [PATCH] feat(rules): rule_group/rule_atom (0033), /rules endpoints, evaluation via referenced rule groups - migration 0033: rule_group (layers, hard excludes, scope, source, revision, fingerprint, default_order) and rule_atom; acquisition_slot.rule_group_ids; preference tags migrated into rule groups (same id/fingerprint), suppressions linked; seeded default groups for acquisition / upgrade / search - contracts: RuleGroup/RuleAtom/RuleExpr/RuleDefaults, parse/serialize/preview schemas; Api.rules, Api.ruleAtoms, Api.rulePolicies, Api.ruleSuppressions - server: GET/POST/PATCH/DELETE /rules, /rules/{id}/revisions, /rules/{id}/preview, /rules/parse, /rules/serialize, /rules/keywords, /rules/defaults, /rules/atoms; disable/delete writes suppression, agent re-add -> 409 - /preference-tags now aliases rule groups; /policy-revisions and /preference-suppressions kept as aliases of /rules/policies, /rules/suppressions - evaluateAcquisition resolves slot -> subscription -> default chain (+ upgrade defaults) and records pri / rule_group@revision / per-layer trace Co-Authored-By: Claude Opus 5.5 --- .../routes/rule-groups.ts | 617 ++++++++++++++ .../acquisition-organize/routes/rules.ts | 762 ++++++++++-------- .../services/acquisition.ts | 34 +- .../acquisition-organize/services/rules.ts | 347 ++++++++ apps/server/test/rules/rule-groups.test.ts | 282 +++++++ packages/contracts/src/api.ts | 160 ++++ packages/contracts/src/policy.ts | 283 +++++++ packages/contracts/test/db-sync.test.ts | 5 + packages/db/migrations/0033_rule_groups.ts | 90 +++ packages/db/src/types.ts | 53 ++ packages/db/test/schema.test.ts | 3 + 11 files changed, 2296 insertions(+), 340 deletions(-) create mode 100644 apps/server/src/modules/acquisition-organize/routes/rule-groups.ts create mode 100644 apps/server/src/modules/acquisition-organize/services/rules.ts create mode 100644 apps/server/test/rules/rule-groups.test.ts create mode 100644 packages/db/migrations/0033_rule_groups.ts diff --git a/apps/server/src/modules/acquisition-organize/routes/rule-groups.ts b/apps/server/src/modules/acquisition-organize/routes/rule-groups.ts new file mode 100644 index 00000000..b6ae044f --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/routes/rule-groups.ts @@ -0,0 +1,617 @@ +import { Api, type RuleGroupCreate, type RuleGroupUpdate, type RuleKeyword } from "@umt/contracts"; +import { type DB, type RuleGroupRow, sql, type Transaction } from "@umt/db"; +import { + atomsOf, + BUILTIN_ATOM_MAP, + BUILTIN_ATOMS, + parseRuleString, + type RuleExpr, + type RuleLayer, + serializeLayers, +} from "@umt/domain"; +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 { contextOf, parsedOf } from "../services/acquisition.ts"; +import type { AoDeps } from "../services/common.ts"; +import { + checkAtomPatterns, + loadRuleCatalog, + normalizeGroupInput, + previewRuleGroup, + ruleAtomToApi, + ruleFingerprint, + ruleGroupToApi, + ruleStringOf, +} from "../services/rules.ts"; +import { actorOf, conflict, jsonb, notFound } from "../util.ts"; +import { mount, paginate } from "./helpers.ts"; + +// 规则页后端(W9-D):规则组 CRUD / 修订 / 试算、规则串解析与序列化、关键字表、自定义条件、三套默认链。 +// 停用或删除规则组写抑制记录(同偏好标签):source=agent 的新建命中有效抑制 → 409;手动/对话新建则解除抑制。 +// 注意注册顺序:静态路径(/rules/keywords 等)必须先于 /rules/{id}。 + +const TAG = "rules"; + +type Trx = Transaction; + +/** 写审计(修订历史端点据此回放) */ +export async function auditRuleGroup( + trx: Trx, + c: Context, + action: "created" | "updated" | "deleted", + row: RuleGroupRow, + changedFields: string[], +) { + const actor = c.get("actor"); + await trx + .insertInto("audit_event") + .values({ + actor_type: actor?.type === "api_key" ? "api_key" : "user", + actor_id: actor?.id ?? null, + action: `rule_group.${action}`, + target_type: "rule_group", + target_id: row.id, + payload: jsonb({ revision: row.revision, changedFields, snapshot: ruleGroupToApi(row) }), + }) + .execute(); +} + +/** 写抑制记录;preference_tag_id 只在该 id 仍是旧偏好标签行时填(外键) */ +export async function suppressRuleGroup(trx: Trx, groupId: string, fingerprint: string, by: string, reason: string) { + await trx + .insertInto("preference_suppression") + .values({ + preference_tag_id: sql`(select id from preference_tag where id = ${groupId})`, + rule_group_id: groupId, + fingerprint, + reason, + evidence: jsonb({ ruleGroupId: groupId }), + suppressed_by: by, + }) + .execute(); +} + +export async function liftSuppressions(trx: Trx, fingerprint: string, by: string) { + await trx + .updateTable("preference_suppression") + .set({ active: false, lifted_at: new Date(), lifted_by: by }) + .where("fingerprint", "=", fingerprint) + .where("active", "=", true) + .execute(); +} + +/** 新建前的抑制检查:agent 命中有效抑制 → 409;其余来源解除抑制 */ +export async function checkSuppression(trx: Trx, fingerprint: string, source: string, by: string) { + const sup = await trx + .selectFrom("preference_suppression") + .select("id") + .where("fingerprint", "=", fingerprint) + .where("active", "=", true) + .executeTakeFirst(); + if (!sup) return; + if (source === "agent") conflict("用户已停用或删除过同一语义的规则,Agent 不能自动加回"); + await liftSuppressions(trx, fingerprint, by); +} + +/** 被版本目标 / 订阅显式引用的次数 */ +async function referenceCount(trx: Trx, id: string): Promise { + const r = await sql<{ n: number }>` + select (select count(*) from acquisition_slot where rule_group_ids @> ${JSON.stringify([id])}::jsonb) + + (select count(*) from subscription where params -> 'ruleGroupIds' @> ${JSON.stringify([id])}::jsonb) + as n`.execute(trx); + return Number(r.rows[0]?.n ?? 0); +} + +async function checkIdsExist(exec: Trx | AoDeps["ctx"]["db"], ids: readonly string[]) { + if (!ids.length) return; + const rows = await exec + .selectFrom("rule_group") + .select("id") + .where("id", "in", [...ids]) + .where("deleted_at", "is", null) + .execute(); + const have = new Set(rows.map((r) => r.id)); + const missing = ids.filter((x) => !have.has(x)); + if (missing.length) throw new ApiException("bad_request", `规则组不存在:${missing.join(", ")}`); +} + +/** 版本目标 / 订阅引用的规则组 id 校验(供 slots 路由复用) */ +export const assertRuleGroupIds = checkIdsExist; + +const TRACKED_FIELDS = [ + "name", + "description", + "kind", + "scope", + "layers", + "hardExcludes", + "ladderRequired", + "enabled", + "evidence", + "autoFill", +] as const; + +export function registerRuleGroupRoutes(app: ModuleApp, d: AoDeps) { + const db = () => d.ctx.db; + + // ── 静态路径 ── + mount( + app, + Api.rules.keywords, + async () => { + const custom = await db().selectFrom("rule_atom").selectAll().orderBy("key").execute(); + const items: RuleKeyword[] = [ + ...BUILTIN_ATOMS.map((a) => ({ + id: a.id, + name: a.name, + description: a.description, + builtin: true, + supported: a.supported, + basis: a.basis, + mp: { ...a.mp } as Record, + })), + ...custom.map((a) => ({ + id: a.key, + name: a.name ?? a.key, + description: a.description ?? "自定义条件", + builtin: false, + supported: true, + basis: "custom" as const, + mp: null, + })), + ]; + return { items }; + }, + { tag: TAG }, + ); + mount( + app, + Api.rules.parse, + async (_c, { body }) => { + const atoms = await db().selectFrom("rule_atom").select("key").execute(); + const p = parseRuleString(body.ruleString, { customAtoms: atoms.map((a) => a.key) }); + return { layers: p.layers, ruleString: p.errors.length ? null : serializeLayers(p.layers), errors: p.errors }; + }, + { tag: TAG, status: 200 }, + ); + mount( + app, + Api.rules.serialize, + async (_c, { body }) => ({ ruleString: serializeLayers(body.layers as RuleLayer[]) }), + { tag: TAG, status: 200 }, + ); + mount( + app, + Api.rules.defaults, + async () => { + const cat = await loadRuleCatalog(db()); + const pick = (k: string) => + [...cat.groups.values()] + .filter((g) => g.kind === k && g.default_order !== null) + .sort((a, b) => a.default_order! - b.default_order!) + .map((g) => g.id); + return { acquisition: pick("acquisition"), upgrade: pick("upgrade"), search: pick("search") }; + }, + { tag: TAG }, + ); + mount( + app, + Api.rules.setDefaults, + async (_c, { body }) => { + await db() + .transaction() + .execute(async (trx) => { + const all = [...body.acquisition, ...body.upgrade, ...body.search]; + await checkIdsExist(trx, all); + const rows = await trx + .selectFrom("rule_group") + .select(["id", "kind"]) + .where("id", "in", all.length ? all : ["00000000-0000-0000-0000-000000000000"]) + .execute(); + for (const k of ["acquisition", "upgrade", "search"] as const) { + const wrong = rows.filter((r) => body[k].includes(r.id) && r.kind !== k); + if (wrong.length) + throw new ApiException("bad_request", `规则组类型不符(${k}):${wrong.map((w) => w.id).join(", ")}`); + } + await trx + .updateTable("rule_group") + .set({ default_order: null }) + .where("default_order", "is not", null) + .execute(); + for (const k of ["acquisition", "upgrade", "search"] as const) + for (const [i, id] of body[k].entries()) + await trx.updateTable("rule_group").set({ default_order: i }).where("id", "=", id).execute(); + }); + return body; + }, + { tag: TAG }, + ); + + // ── 自定义条件 ── + mount( + app, + Api.ruleAtoms.list, + async (_c, { query }) => { + const p = await paginate(db().selectFrom("rule_atom").selectAll(), query); + return { items: p.rows.map(ruleAtomToApi), nextCursor: p.nextCursor }; + }, + { tag: TAG }, + ); + mount( + app, + Api.ruleAtoms.get, + async (_c, { params }) => { + const r = await db().selectFrom("rule_atom").selectAll().where("id", "=", params.id).executeTakeFirst(); + return ruleAtomToApi(r ?? notFound("自定义条件")); + }, + { tag: TAG }, + ); + mount( + app, + Api.ruleAtoms.create, + async (_c, { body }) => { + const key = String(body.key).toUpperCase(); + if (BUILTIN_ATOM_MAP.has(key)) conflict(`${key} 是内置关键字,不能覆盖`); + checkAtomPatterns(body.include); + checkAtomPatterns(body.exclude); + const r = await db() + .insertInto("rule_atom") + .values({ + key, + name: body.name ?? null, + description: body.description ?? null, + include: JSON.stringify(body.include ?? []), + exclude: JSON.stringify(body.exclude ?? []), + match_fields: JSON.stringify(body.matchFields ?? []), + unit_size_mb: body.unitSizeMb ? jsonb(body.unitSizeMb) : null, + min_seeders: body.minSeeders ?? null, + publish_delay_minutes: body.publishDelayMinutes ? jsonb(body.publishDelayMinutes) : null, + free: body.free ?? false, + }) + .onConflict((oc) => oc.column("key").doNothing()) + .returningAll() + .executeTakeFirst(); + if (!r) conflict(`自定义条件 ${key} 已存在`); + return ruleAtomToApi(r!); + }, + { tag: TAG }, + ); + mount( + app, + Api.ruleAtoms.update, + async (_c, { params, body }) => { + checkAtomPatterns(body.include); + checkAtomPatterns(body.exclude); + const r = await db() + .updateTable("rule_atom") + .set({ + ...(body.name !== undefined ? { name: body.name } : {}), + ...(body.description !== undefined ? { description: body.description } : {}), + ...(body.include !== undefined ? { include: JSON.stringify(body.include) } : {}), + ...(body.exclude !== undefined ? { exclude: JSON.stringify(body.exclude) } : {}), + ...(body.matchFields !== undefined ? { match_fields: JSON.stringify(body.matchFields) } : {}), + ...(body.unitSizeMb !== undefined ? { unit_size_mb: body.unitSizeMb ? jsonb(body.unitSizeMb) : null } : {}), + ...(body.minSeeders !== undefined ? { min_seeders: body.minSeeders } : {}), + ...(body.publishDelayMinutes !== undefined + ? { publish_delay_minutes: body.publishDelayMinutes ? jsonb(body.publishDelayMinutes) : null } + : {}), + ...(body.free !== undefined ? { free: body.free } : {}), + }) + .where("id", "=", params.id) + .returningAll() + .executeTakeFirst(); + return ruleAtomToApi(r ?? notFound("自定义条件")); + }, + { tag: TAG }, + ); + mount( + app, + Api.ruleAtoms.remove, + async (_c, { params }) => { + const a = await db().selectFrom("rule_atom").select(["id", "key"]).where("id", "=", params.id).executeTakeFirst(); + if (!a) notFound("自定义条件"); + const cat = await loadRuleCatalog(db()); + const users = [...cat.groups.values()].filter((g) => + [...((g.layers ?? []) as RuleLayer[]).map((l) => l.expr), ...((g.hard_excludes ?? []) as RuleExpr[])].some( + (e) => atomsOf(e).has(a!.key), + ), + ); + if (users.length) conflict(`自定义条件 ${a!.key} 被规则组引用:${users.map((u) => u.name).join("、")}`); + await db().deleteFrom("rule_atom").where("id", "=", params.id).execute(); + return { ok: true as const }; + }, + { tag: TAG, status: 200 }, + ); + + // ── 规则组 ── + mount( + app, + Api.rules.list, + async (_c, { query }) => { + let q = db().selectFrom("rule_group").selectAll(); + if (query.includeDeleted !== "true") q = q.where("deleted_at", "is", null); + if (query.kind) q = q.where("kind", "=", query.kind); + if (query.source) q = q.where("source", "=", query.source); + if (query.enabled) q = q.where("enabled", "=", query.enabled === "true"); + if (query.q) q = q.where("name", "ilike", `%${String(query.q).replace(/[\\%_]/g, (m) => `\\${m}`)}%`); + const p = await paginate(q, query); + return { items: p.rows.map(ruleGroupToApi), nextCursor: p.nextCursor }; + }, + { tag: TAG }, + ); + mount( + app, + Api.rules.get, + async (_c, { params }) => { + const r = await db().selectFrom("rule_group").selectAll().where("id", "=", params.id).executeTakeFirst(); + return ruleGroupToApi(r ?? notFound("规则组")); + }, + { tag: TAG }, + ); + mount( + app, + Api.rules.create, + async (c, { body }) => { + const b = body as RuleGroupCreate; + const by = actorOf(c); + const r = await db() + .transaction() + .execute(async (trx) => { + const atoms = await trx.selectFrom("rule_atom").select("key").execute(); + const n = normalizeGroupInput( + b, + atoms.map((a) => a.key), + ); + const layers = n.layers ?? []; + const hardExcludes = n.hardExcludes ?? []; + const kind = b.kind ?? "acquisition"; + const scope = b.scope ?? {}; + const ladderRequired = b.ladderRequired ?? true; + const source = b.source ?? "manual"; + const fp = ruleFingerprint({ kind, scope, layers, hardExcludes, ladderRequired }); + await sql`select pg_advisory_xact_lock(hashtext(${`rule_group:${fp}`}))`.execute(trx); + const dupName = await trx + .selectFrom("rule_group") + .select("id") + .where("name", "=", b.name) + .where("deleted_at", "is", null) + .executeTakeFirst(); + if (dupName) conflict(`已有同名规则组「${b.name}」`); + if (source !== "manual") { + const dup = await trx + .selectFrom("rule_group") + .select("id") + .where("fingerprint", "=", fp) + .where("deleted_at", "is", null) + .executeTakeFirst(); + if (dup) conflict("已存在同一语义的规则组"); + } + await checkSuppression(trx, fp, source, by); + const row = await trx + .insertInto("rule_group") + .values({ + name: b.name, + description: b.description ?? null, + kind, + scope: jsonb(scope), + layers: JSON.stringify(layers), + hard_excludes: JSON.stringify(hardExcludes), + ladder_required: ladderRequired, + rule_string: ruleStringOf(layers), + source, + enabled: b.enabled ?? true, + fingerprint: fp, + evidence: jsonb(b.evidence), + auto_fill: b.autoFill ?? null, + }) + .returningAll() + .executeTakeFirstOrThrow(); + await auditRuleGroup(trx, c, "created", row, []); + return row; + }); + return ruleGroupToApi(r); + }, + { tag: TAG }, + ); + mount( + app, + Api.rules.update, + async (c, { params, body }) => { + const b = body as RuleGroupUpdate; + const by = actorOf(c); + const r = await db() + .transaction() + .execute(async (trx) => { + const cur = await trx + .selectFrom("rule_group") + .selectAll() + .where("id", "=", params.id) + .where("deleted_at", "is", null) + .forUpdate() + .executeTakeFirst(); + if (!cur) notFound("规则组"); + if (b.expectedRevision !== undefined && b.expectedRevision !== cur!.revision) + conflict(`revision 不匹配(当前 ${cur!.revision})`); + if (b.name !== undefined && b.name !== cur!.name) { + const dupName = await trx + .selectFrom("rule_group") + .select("id") + .where("name", "=", b.name) + .where("deleted_at", "is", null) + .where("id", "!=", cur!.id) + .executeTakeFirst(); + if (dupName) conflict(`已有同名规则组「${b.name}」`); + } + const atoms = await trx.selectFrom("rule_atom").select("key").execute(); + const n = normalizeGroupInput( + b, + atoms.map((a) => a.key), + ); + const next = { + kind: b.kind ?? cur!.kind, + scope: b.scope ?? cur!.scope, + layers: n.layers ?? (cur!.layers as RuleLayer[]), + hardExcludes: n.hardExcludes ?? (cur!.hard_excludes as RuleExpr[]), + ladderRequired: b.ladderRequired ?? cur!.ladder_required, + }; + if (b.kind !== undefined && b.kind !== cur!.kind && cur!.default_order !== null) + conflict("该规则组在默认链里,先从默认值移出再改类型"); + // 偏好标签迁来的组沿用旧指纹口径,规则没变就不重算 + const semanticChanged = + b.kind !== undefined || + b.scope !== undefined || + n.layers !== undefined || + n.hardExcludes !== undefined || + b.ladderRequired !== undefined; + const fp = semanticChanged ? ruleFingerprint(next) : cur!.fingerprint; + const enabled = b.enabled ?? cur!.enabled; + let suppressedAt = cur!.suppressed_at; + if (!enabled && cur!.enabled) { + await suppressRuleGroup(trx, cur!.id, cur!.fingerprint, by, "disabled"); + suppressedAt = new Date(); + } + if (enabled && !cur!.enabled) { + await liftSuppressions(trx, fp, by); + suppressedAt = null; + } + // 对话 / Agent 来源的规则被用户改掉:旧语义留抑制记录,Agent 不能据旧证据改回 + if (fp !== cur!.fingerprint && cur!.enabled && enabled && cur!.source !== "manual") + await suppressRuleGroup(trx, cur!.id, cur!.fingerprint, by, "edited"); + const before = ruleGroupToApi(cur!) as unknown as Record; + const row = await trx + .updateTable("rule_group") + .set({ + ...(b.name !== undefined ? { name: b.name } : {}), + ...(b.description !== undefined ? { description: b.description } : {}), + kind: next.kind, + scope: jsonb(next.scope), + layers: JSON.stringify(next.layers), + hard_excludes: JSON.stringify(next.hardExcludes), + ladder_required: next.ladderRequired, + rule_string: ruleStringOf(next.layers), + enabled, + fingerprint: fp, + suppressed_at: suppressedAt, + ...(b.evidence !== undefined ? { evidence: jsonb(b.evidence) } : {}), + ...(b.autoFill !== undefined ? { auto_fill: b.autoFill } : {}), + revision: cur!.revision + 1, + }) + .where("id", "=", cur!.id) + .returningAll() + .executeTakeFirstOrThrow(); + const after = ruleGroupToApi(row) as unknown as Record; + const changed = TRACKED_FIELDS.filter((k) => JSON.stringify(before[k]) !== JSON.stringify(after[k])); + await auditRuleGroup(trx, c, "updated", row, changed); + return row; + }); + await kickReevaluate(d, r.id); + return ruleGroupToApi(r); + }, + { tag: TAG }, + ); + mount( + app, + Api.rules.remove, + async (c, { params }) => { + await db() + .transaction() + .execute(async (trx) => { + const cur = await trx + .selectFrom("rule_group") + .selectAll() + .where("id", "=", params.id) + .where("deleted_at", "is", null) + .forUpdate() + .executeTakeFirst(); + if (!cur) notFound("规则组"); + const refs = await referenceCount(trx, cur!.id); + if (refs > 0) conflict(`规则组被 ${refs} 个版本目标或订阅引用,先解除引用`); + const row = await trx + .updateTable("rule_group") + .set({ + deleted_at: new Date(), + enabled: false, + default_order: null, + suppressed_at: cur!.suppressed_at ?? new Date(), + revision: cur!.revision + 1, + }) + .where("id", "=", cur!.id) + .returningAll() + .executeTakeFirstOrThrow(); + await auditRuleGroup(trx, c, "deleted", row, []); + if (cur!.enabled) await suppressRuleGroup(trx, cur!.id, cur!.fingerprint, actorOf(c), "deleted"); + }); + return { ok: true as const }; + }, + { tag: TAG, status: 200 }, + ); + mount( + app, + Api.rules.revisions, + async (_c, { params }) => { + const g = await db().selectFrom("rule_group").select("id").where("id", "=", params.id).executeTakeFirst(); + if (!g) notFound("规则组"); + const rows = await db() + .selectFrom("audit_event") + .selectAll() + .where("target_id", "=", params.id) + .where((eb) => eb.or([eb("target_type", "=", "rule_group"), eb("target_type", "=", "preference_tag")])) + .orderBy("at", "desc") + .orderBy("id", "desc") + .limit(200) + .execute(); + return { + items: rows.map((r) => { + const p = (r.payload ?? {}) as { + revision?: number; + changedFields?: string[]; + snapshot?: Record; + }; + return { + id: r.id, + at: r.at.toISOString(), + action: r.action, + actorType: r.actor_type, + actorId: r.actor_id, + revision: p.revision ?? null, + changedFields: p.changedFields ?? [], + snapshot: p.snapshot ?? {}, + }; + }), + }; + }, + { tag: TAG }, + ); + mount( + app, + Api.rules.preview, + async (_c, { params, body }) => { + const row = await db().selectFrom("rule_group").selectAll().where("id", "=", params.id).executeTakeFirst(); + if (!row) notFound("规则组"); + return previewRuleGroup(d, row!, body ?? {}, { parsedOf, contextOf }); + }, + { tag: TAG, status: 200 }, + ); +} + +/** 规则组改动后重评引用它的版本目标(显式引用;默认链改动由周期重评覆盖) */ +async function kickReevaluate(d: AoDeps, groupId: string) { + try { + const slots = await d.ctx.db + .selectFrom("acquisition_slot") + .select("id") + .where(sql`rule_group_ids @> ${JSON.stringify([groupId])}::jsonb`) + .limit(50) + .execute(); + for (const s of slots) + await d.ctx.engine.enqueue( + "evaluateAcquisition", + { acquisitionSlotId: s.id }, + { idempotencyKey: `reeval:${s.id}:${Date.now()}`, traceId: s.id }, + ); + } catch (err) { + d.ctx.logger.debug({ err: { message: (err as Error).message } }, "规则组重评入队失败"); + } +} diff --git a/apps/server/src/modules/acquisition-organize/routes/rules.ts b/apps/server/src/modules/acquisition-organize/routes/rules.ts index 0c74f347..96a63f3e 100644 --- a/apps/server/src/modules/acquisition-organize/routes/rules.ts +++ b/apps/server/src/modules/acquisition-organize/routes/rules.ts @@ -8,20 +8,29 @@ import { type Subscription, type SubscriptionHistory, } from "@umt/contracts"; -import { type DB, sql, type Transaction } from "@umt/db"; -import { evaluateCandidate, parseRelease } from "@umt/domain"; +import { type DB, type PreferenceSuppressionRow, type RuleGroupRow, sql, type Transaction } from "@umt/db"; +import { evaluateCandidate, parseRelease, type RuleExpr, type RuleLayer, serializeLayers } from "@umt/domain"; import { slotSpecFromRow } from "@umt/mapping"; 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 } from "../services/common.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 { + assertRuleGroupIds, + checkSuppression, + liftSuppressions, + registerRuleGroupRoutes, + suppressRuleGroup, +} from "./rule-groups.ts"; -// 规则 API:订阅、收录 Slot、策略版本(草案→发布→回滚,发布即持续授权)、偏好标签与抑制记录。 +// 规则 API:订阅、收录 Slot、策略版本(草案→发布→回滚,发布即持续授权)、偏好标签(规则组别名)与抑制记录; +// 规则组本体见 ./rule-groups.ts。 const TAG = "rules"; @@ -385,6 +394,7 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { .where("id", "=", body.mediaEntityId) .executeTakeFirst(); if (!e) throw new ApiException("bad_request", "mediaEntityId 不存在"); + await assertRuleGroupIds(db(), body.ruleGroupIds ?? []); const r = await db() .insertInto("acquisition_slot") .values({ @@ -400,6 +410,7 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { completion: jsonb(body.completion), distinct_group: body.distinctGroup ?? null, require_distinct_asset: body.requireDistinctAsset ?? false, + rule_group_ids: JSON.stringify(body.ruleGroupIds ?? []), }) .onConflict((oc) => oc.doNothing()) .returningAll() @@ -415,6 +426,7 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { Api.acquisitionSlots.update, async (_c, { params, body }) => { const { expectedRevision, ...b } = body; + if (b.ruleGroupIds !== undefined) await assertRuleGroupIds(db(), b.ruleGroupIds); const r = await db() .updateTable("acquisition_slot") .set({ @@ -427,6 +439,7 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { ...(b.completion !== undefined ? { completion: jsonb(b.completion) } : {}), ...(b.distinctGroup !== undefined ? { distinct_group: b.distinctGroup } : {}), ...(b.requireDistinctAsset !== undefined ? { require_distinct_asset: b.requireDistinctAsset } : {}), + ...(b.ruleGroupIds !== undefined ? { rule_group_ids: JSON.stringify(b.ruleGroupIds) } : {}), revision: sql`revision + 1`, }) .where("id", "=", params.id) @@ -448,199 +461,257 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { { tag: TAG }, ); - // ── 策略版本 ── - mount( - app, - Api.policyRevisions.list, - async (_c, { query }) => { - let q = db().selectFrom("policy_revision").selectAll(); - if (query.policyKey) q = q.where("policy_key", "=", query.policyKey); - if (query.status) q = q.where("status", "=", query.status); - const p = await paginate(q, query); - return { items: toApiList(p.rows), nextCursor: p.nextCursor }; - }, - { tag: TAG }, - ); - mount( - app, - Api.policyRevisions.get, - async (_c, { params }) => { - const r = await db().selectFrom("policy_revision").selectAll().where("id", "=", params.id).executeTakeFirst(); - return toApi(r ?? notFound("策略版本")); - }, - { tag: TAG }, - ); - const checkRules = (kind: string, rules: unknown) => { - if (kind !== "acquisition") return; - const r = AcquisitionPolicyRules.safeParse(rules); - if (!r.success) - throw new ApiException( - "validation_failed", - `rules 不符合收录策略结构: ${r.error.issues.map((i) => `${i.path.join(".")} ${i.message}`).join("; ")}`, - ); - }; - mount( - app, - Api.policyRevisions.create, - async (c, { body }) => { - checkRules(body.kind, body.rules); - const r = await db() - .transaction() - .execute(async (trx) => { - await sql`select pg_advisory_xact_lock(hashtext(${`policy:${body.policyKey}`}))`.execute(trx); - const prev = await trx - .selectFrom("policy_revision") - .select(["kind"]) - .select(sql`max(revision)`.as("max")) - .where("policy_key", "=", body.policyKey) - .groupBy("kind") - .executeTakeFirst(); - if (prev && prev.kind !== body.kind) conflict(`policyKey ${body.policyKey} 已属于 ${prev.kind}`); - return trx - .insertInto("policy_revision") - .values({ - policy_key: body.policyKey, - revision: (prev?.max ?? 0) + 1, - kind: body.kind, - status: "draft", - title: body.title, - summary: body.summary ?? null, - scope: jsonb(body.scope), - rules: jsonb(body.rules), - based_on_revision_id: body.basedOnRevisionId ?? null, - created_by: actorOf(c), - }) - .returningAll() - .executeTakeFirstOrThrow(); - }); - return toApi(r); - }, - { tag: TAG }, - ); - mount( - app, - Api.policyRevisions.update, - async (_c, { params, body }) => { - const cur = await db().selectFrom("policy_revision").selectAll().where("id", "=", params.id).executeTakeFirst(); - if (!cur) notFound("策略版本"); - if (cur!.status !== "draft") conflict("只能修改草案。要改已发布版本,新建草案"); - if (body.rules !== undefined) checkRules(cur!.kind, body.rules); - const r = await db() - .updateTable("policy_revision") - .set({ - ...(body.title !== undefined ? { title: body.title } : {}), - ...(body.summary !== undefined ? { summary: body.summary } : {}), - ...(body.scope !== undefined ? { scope: jsonb(body.scope) } : {}), - ...(body.rules !== undefined ? { rules: jsonb(body.rules) } : {}), - ...(body.basedOnRevisionId !== undefined ? { based_on_revision_id: body.basedOnRevisionId } : {}), - preview: null, - }) - .where("id", "=", params.id) - .where("status", "=", "draft") - .returningAll() - .executeTakeFirstOrThrow(); - return toApi(r); - }, - { tag: TAG }, - ); - mount( - app, - Api.policyRevisions.preview, - async (_c, { params }) => { - const cur = await db().selectFrom("policy_revision").selectAll().where("id", "=", params.id).executeTakeFirst(); - if (!cur) notFound("策略版本"); - const preview = await previewPolicy(d, cur!.rules); - const r = await db() - .updateTable("policy_revision") - .set({ preview: jsonb(preview) }) - .where("id", "=", params.id) - .returningAll() - .executeTakeFirstOrThrow(); - return toApi(r); - }, - { tag: TAG }, - ); - mount( - app, - Api.policyRevisions.publish, - async (c, { params, body }) => { - const r = await db() - .transaction() - .execute(async (trx) => { - const cur = await trx - .selectFrom("policy_revision") - .selectAll() - .where("id", "=", params.id) - .forUpdate() - .executeTakeFirst(); - if (!cur) notFound("策略版本"); - if (cur!.revision !== body.expectedRevision) conflict(`revision 不匹配(当前 ${cur!.revision})`); - if (cur!.status !== "draft") conflict(`只能发布草案(当前 ${cur!.status})`); - checkRules(cur!.kind, cur!.rules); - return publishRevision(trx, cur!.id, cur!.policy_key, actorOf(c), "superseded"); - }); - return toApi(r); - }, - { tag: TAG }, - ); - mount( - app, - Api.policyRevisions.rollback, - async (c, { params }) => { - const r = await db() - .transaction() - .execute(async (trx) => { - const target = await trx - .selectFrom("policy_revision") - .selectAll() - .where("id", "=", params.id) - .executeTakeFirst(); - if (!target) notFound("策略版本"); - if (target!.status === "published") conflict("该版本已是当前发布版本"); - if (target!.status === "draft") conflict("草案不能作为回滚目标"); - await sql`select pg_advisory_xact_lock(hashtext(${`policy:${target!.policy_key}`}))`.execute(trx); - const max = await trx - .selectFrom("policy_revision") - .select(sql`max(revision)`.as("m")) - .where("policy_key", "=", target!.policy_key) - .executeTakeFirstOrThrow(); - // 回滚 = 以目标版本内容发布一个新版本(历史单调递增,可审计) - const clone = await trx - .insertInto("policy_revision") - .values({ - policy_key: target!.policy_key, - revision: max.m + 1, - kind: target!.kind, - status: "draft", - title: target!.title, - summary: `回滚到 r${target!.revision}`, - scope: jsonb(target!.scope), - rules: jsonb(target!.rules), - based_on_revision_id: target!.id, - created_by: actorOf(c), - }) - .returning("id") - .executeTakeFirstOrThrow(); - return publishRevision(trx, clone.id, target!.policy_key, actorOf(c), "rolled_back"); - }); - return toApi(r); - }, - { tag: TAG }, - ); + // ── 策略版本(/rules/policies 为新路径,/policy-revisions 保留为别名,见 docs/runbook/rules.md)── + for (const P of [Api.rulePolicies, Api.policyRevisions]) { + mount( + app, + P.list, + async (_c, { query }) => { + let q = db().selectFrom("policy_revision").selectAll(); + if (query.policyKey) q = q.where("policy_key", "=", query.policyKey); + if (query.status) q = q.where("status", "=", query.status); + const p = await paginate(q, query); + return { items: toApiList(p.rows), nextCursor: p.nextCursor }; + }, + { tag: TAG }, + ); + mount( + app, + P.get, + async (_c, { params }) => { + const r = await db().selectFrom("policy_revision").selectAll().where("id", "=", params.id).executeTakeFirst(); + return toApi(r ?? notFound("策略版本")); + }, + { tag: TAG }, + ); + const checkRules = (kind: string, rules: unknown) => { + if (kind !== "acquisition") return; + const r = AcquisitionPolicyRules.safeParse(rules); + if (!r.success) + throw new ApiException( + "validation_failed", + `rules 不符合收录策略结构: ${r.error.issues.map((i) => `${i.path.join(".")} ${i.message}`).join("; ")}`, + ); + }; + mount( + app, + P.create, + async (c, { body }) => { + checkRules(body.kind, body.rules); + const r = await db() + .transaction() + .execute(async (trx) => { + await sql`select pg_advisory_xact_lock(hashtext(${`policy:${body.policyKey}`}))`.execute(trx); + const prev = await trx + .selectFrom("policy_revision") + .select(["kind"]) + .select(sql`max(revision)`.as("max")) + .where("policy_key", "=", body.policyKey) + .groupBy("kind") + .executeTakeFirst(); + if (prev && prev.kind !== body.kind) conflict(`policyKey ${body.policyKey} 已属于 ${prev.kind}`); + return trx + .insertInto("policy_revision") + .values({ + policy_key: body.policyKey, + revision: (prev?.max ?? 0) + 1, + kind: body.kind, + status: "draft", + title: body.title, + summary: body.summary ?? null, + scope: jsonb(body.scope), + rules: jsonb(body.rules), + based_on_revision_id: body.basedOnRevisionId ?? null, + created_by: actorOf(c), + }) + .returningAll() + .executeTakeFirstOrThrow(); + }); + return toApi(r); + }, + { tag: TAG }, + ); + mount( + app, + P.update, + async (_c, { params, body }) => { + const cur = await db().selectFrom("policy_revision").selectAll().where("id", "=", params.id).executeTakeFirst(); + if (!cur) notFound("策略版本"); + if (cur!.status !== "draft") conflict("只能修改草案。要改已发布版本,新建草案"); + if (body.rules !== undefined) checkRules(cur!.kind, body.rules); + const r = await db() + .updateTable("policy_revision") + .set({ + ...(body.title !== undefined ? { title: body.title } : {}), + ...(body.summary !== undefined ? { summary: body.summary } : {}), + ...(body.scope !== undefined ? { scope: jsonb(body.scope) } : {}), + ...(body.rules !== undefined ? { rules: jsonb(body.rules) } : {}), + ...(body.basedOnRevisionId !== undefined ? { based_on_revision_id: body.basedOnRevisionId } : {}), + preview: null, + }) + .where("id", "=", params.id) + .where("status", "=", "draft") + .returningAll() + .executeTakeFirstOrThrow(); + return toApi(r); + }, + { tag: TAG }, + ); + mount( + app, + P.preview, + async (_c, { params }) => { + const cur = await db().selectFrom("policy_revision").selectAll().where("id", "=", params.id).executeTakeFirst(); + if (!cur) notFound("策略版本"); + const preview = await previewPolicy(d, cur!.rules); + const r = await db() + .updateTable("policy_revision") + .set({ preview: jsonb(preview) }) + .where("id", "=", params.id) + .returningAll() + .executeTakeFirstOrThrow(); + return toApi(r); + }, + { tag: TAG }, + ); + mount( + app, + P.publish, + async (c, { params, body }) => { + const r = await db() + .transaction() + .execute(async (trx) => { + const cur = await trx + .selectFrom("policy_revision") + .selectAll() + .where("id", "=", params.id) + .forUpdate() + .executeTakeFirst(); + if (!cur) notFound("策略版本"); + if (cur!.revision !== body.expectedRevision) conflict(`revision 不匹配(当前 ${cur!.revision})`); + if (cur!.status !== "draft") conflict(`只能发布草案(当前 ${cur!.status})`); + checkRules(cur!.kind, cur!.rules); + return publishRevision(trx, cur!.id, cur!.policy_key, actorOf(c), "superseded"); + }); + return toApi(r); + }, + { tag: TAG }, + ); + mount( + app, + P.rollback, + async (c, { params }) => { + const r = await db() + .transaction() + .execute(async (trx) => { + const target = await trx + .selectFrom("policy_revision") + .selectAll() + .where("id", "=", params.id) + .executeTakeFirst(); + if (!target) notFound("策略版本"); + if (target!.status === "published") conflict("该版本已是当前发布版本"); + if (target!.status === "draft") conflict("草案不能作为回滚目标"); + await sql`select pg_advisory_xact_lock(hashtext(${`policy:${target!.policy_key}`}))`.execute(trx); + const max = await trx + .selectFrom("policy_revision") + .select(sql`max(revision)`.as("m")) + .where("policy_key", "=", target!.policy_key) + .executeTakeFirstOrThrow(); + // 回滚 = 以目标版本内容发布一个新版本(历史单调递增,可审计) + const clone = await trx + .insertInto("policy_revision") + .values({ + policy_key: target!.policy_key, + revision: max.m + 1, + kind: target!.kind, + status: "draft", + title: target!.title, + summary: `回滚到 r${target!.revision}`, + scope: jsonb(target!.scope), + rules: jsonb(target!.rules), + based_on_revision_id: target!.id, + created_by: actorOf(c), + }) + .returning("id") + .executeTakeFirstOrThrow(); + return publishRevision(trx, clone.id, target!.policy_key, actorOf(c), "rolled_back"); + }); + return toApi(r); + }, + { tag: TAG }, + ); + } - // ── 偏好标签 ── + // ── 偏好标签(已并入规则组;旧端点保留为别名,见 docs/runbook/rules.md)── + // 标签 = legacy_kind 非空的规则组;指纹沿用旧口径 hash(kind+scope+rule),旧抑制记录照常命中。 + // rule.ruleString(可选)会编译成规则:hard → 硬排除,soft → 单层软偏好;其余只作记录。 const fingerprint = (b: { kind: string; scope?: unknown; rule?: unknown }) => hashJson({ kind: b.kind, scope: b.scope ?? {}, rule: b.rule ?? {} }); + const tagStatus = (r: RuleGroupRow) => (r.deleted_at ? "deleted" : r.enabled ? "active" : "disabled"); + const toTag = (r: RuleGroupRow): PreferenceTag => ({ + id: r.id, + createdAt: r.created_at.toISOString(), + updatedAt: r.updated_at.toISOString(), + label: r.name, + kind: (r.legacy_kind ?? "soft") as PreferenceTag["kind"], + scope: (r.scope ?? {}) as Record, + rule: (r.legacy_rule ?? {}) as Record, + fingerprint: r.fingerprint, + origin: r.source === "chat" ? "conversation" : r.source === "agent" ? "agent_inferred" : "manual", + description: r.description, + status: tagStatus(r), + revision: r.revision, + evidence: (r.evidence ?? {}) as Record, + policyRevisionId: null, + }); + const sourceOf = (o: string | undefined) => + o === "conversation" ? "chat" : o === "agent_inferred" ? "agent" : "manual"; + /** 旧标签 rule → 规则组层 / 硬排除 */ + const compileTag = async (kind: string, rule: unknown) => { + const rs = (rule as { ruleString?: unknown } | null)?.ruleString; + if (typeof rs !== "string" || !rs.trim() || (kind !== "hard" && kind !== "soft")) + return { layers: [], hardExcludes: [] }; + const atoms = await db().selectFrom("rule_atom").select("key").execute(); + const keys = atoms.map((a) => a.key); + return kind === "hard" + ? { + layers: [], + hardExcludes: + normalizeGroupInput({ hardExcludes: rs.split(">").map((x) => x.trim()) }, keys).hardExcludes ?? [], + } + : { layers: normalizeGroupInput({ ruleString: rs }, keys).layers ?? [], hardExcludes: [] }; + }; + const tagSnapshot = (r: RuleGroupRow) => ({ + id: r.id, + label: r.name, + kind: r.legacy_kind ?? "soft", + status: tagStatus(r), + scope: r.scope, + rule: r.legacy_rule ?? {}, + description: r.description, + revision: r.revision, + }); + const liveTag = (trx: Exec, id: string) => + trx + .selectFrom("rule_group") + .selectAll() + .where("id", "=", id) + .where("legacy_kind", "is not", null) + .where("deleted_at", "is", null); mount( app, Api.preferenceTags.list, async (_c, { query }) => { - let q = db().selectFrom("preference_tag").selectAll(); - if (query.status) q = q.where("status", "=", query.status); - else q = q.where("status", "!=", "deleted"); - if (query.kind) q = q.where("kind", "=", query.kind); + let q = db().selectFrom("rule_group").selectAll().where("legacy_kind", "is not", null); + if (query.status === "deleted") q = q.where("deleted_at", "is not", null); + else { + q = q.where("deleted_at", "is", null); + if (query.status) q = q.where("enabled", "=", query.status === "active"); + } + if (query.kind) q = q.where("legacy_kind", "=", query.kind); const p = await paginate(q, query); - return { items: toApiList(p.rows), nextCursor: p.nextCursor }; + return { items: p.rows.map(toTag), nextCursor: p.nextCursor }; }, { tag: TAG }, ); @@ -648,8 +719,13 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { app, Api.preferenceTags.get, async (_c, { params }) => { - const r = await db().selectFrom("preference_tag").selectAll().where("id", "=", params.id).executeTakeFirst(); - return toApi(r ?? notFound("偏好标签")); + const r = await db() + .selectFrom("rule_group") + .selectAll() + .where("id", "=", params.id) + .where("legacy_kind", "is not", null) + .executeTakeFirst(); + return toTag(r ?? notFound("偏好标签")); }, { tag: TAG }, ); @@ -658,45 +734,43 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { Api.preferenceTags.create, async (c, { body }) => { const fp = fingerprint(body); - const origin = body.origin ?? "manual"; + const source = sourceOf(body.origin); + const compiled = await compileTag(body.kind, body.rule); const r = await db() .transaction() .execute(async (trx) => { - const sup = await trx - .selectFrom("preference_suppression") - .selectAll() + await sql`select pg_advisory_xact_lock(hashtext(${`rule_group:${fp}`}))`.execute(trx); + await checkSuppression(trx, fp, source, actorOf(c)); + const dup = await trx + .selectFrom("rule_group") + .select("id") .where("fingerprint", "=", fp) - .where("active", "=", true) + .where("deleted_at", "is", null) .executeTakeFirst(); - if (sup) { - // Agent 不得据旧证据自动加回;用户手动(或对话确认)重新添加即解除抑制 - if (origin === "agent_inferred") conflict("用户已取消该偏好,Agent 不能自动加回"); - await trx - .updateTable("preference_suppression") - .set({ active: false, lifted_at: new Date(), lifted_by: actorOf(c) }) - .where("id", "=", sup.id) - .execute(); - } + if (dup) conflict("已存在同一语义的偏好标签"); const ins = await trx - .insertInto("preference_tag") + .insertInto("rule_group") .values({ - label: body.label, - kind: body.kind, - scope: jsonb(body.scope), - rule: jsonb(body.rule), - fingerprint: fp, - origin, + name: body.label, description: body.description ?? null, + kind: "acquisition", + scope: jsonb(body.scope), + layers: JSON.stringify(compiled.layers), + hard_excludes: JSON.stringify(compiled.hardExcludes), + ladder_required: body.kind !== "soft", + rule_string: compiled.layers.length ? serializeLayers(compiled.layers) : null, + source, + fingerprint: fp, evidence: jsonb(body.evidence), + legacy_kind: body.kind, + legacy_rule: jsonb(body.rule), }) - .onConflict((oc) => oc.doNothing()) .returningAll() - .executeTakeFirst(); - if (!ins) conflict("已存在同一语义的偏好标签"); - await auditPref(trx, c, "created", ins!, []); - return ins!; + .executeTakeFirstOrThrow(); + await auditPref(trx, c, "created", tagSnapshot(ins), []); + return ins; }); - return toApi(r); + return toTag(r); }, { tag: TAG }, ); @@ -704,54 +778,56 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { app, Api.preferenceTags.update, async (c, { params, body }) => { + const by = actorOf(c); const r = await db() .transaction() .execute(async (trx) => { - const cur = await trx - .selectFrom("preference_tag") - .selectAll() - .where("id", "=", params.id) - .forUpdate() - .executeTakeFirst(); - if (!cur || cur.status === "deleted") notFound("偏好标签"); - const next = { kind: cur!.kind, scope: body.scope ?? cur!.scope, rule: body.rule ?? cur!.rule }; + const cur = await liveTag(trx, params.id).forUpdate().executeTakeFirst(); + if (!cur) notFound("偏好标签"); + const kind = body.kind ?? cur!.legacy_kind!; + const next = { kind, scope: body.scope ?? cur!.scope, rule: body.rule ?? cur!.legacy_rule ?? {} }; const fp = fingerprint(next); - const status = body.status ?? cur!.status; - if (status !== "active" && cur!.status === "active") - await suppress(trx, cur!.id, cur!.fingerprint, actorOf(c), `status:${status}`); - if (status === "active" && cur!.status !== "active") { - await trx - .updateTable("preference_suppression") - .set({ active: false, lifted_at: new Date(), lifted_by: actorOf(c) }) - .where("fingerprint", "=", fp) - .where("active", "=", true) - .execute(); - } - if (fp !== cur!.fingerprint && cur!.status === "active") - await suppress(trx, cur!.id, cur!.fingerprint, actorOf(c), "edited"); - const changed = (["label", "kind", "scope", "rule", "description", "evidence", "status"] as const).filter( - (k) => body[k] !== undefined && JSON.stringify(body[k]) !== JSON.stringify(toApi(cur!)[k]), - ); - const next_ = await trx - .updateTable("preference_tag") + const status = body.status ?? tagStatus(cur!); + if (status === "deleted") throw new ApiException("bad_request", "删除用 DELETE"); + const enabled = status === "active"; + if (!enabled && cur!.enabled) await suppressRuleGroup(trx, cur!.id, cur!.fingerprint, by, `status:${status}`); + if (enabled && !cur!.enabled) await liftSuppressions(trx, fp, by); + if (fp !== cur!.fingerprint && cur!.enabled) + await suppressRuleGroup(trx, cur!.id, cur!.fingerprint, by, "edited"); + const compiled = + body.rule !== undefined || body.kind !== undefined + ? await compileTag(kind, next.rule) + : { layers: cur!.layers as RuleLayer[], hardExcludes: cur!.hard_excludes as RuleExpr[] }; + const before = toTag(cur!); + const row = await trx + .updateTable("rule_group") .set({ - ...(body.label !== undefined ? { label: body.label } : {}), - ...(body.kind !== undefined ? { kind: body.kind } : {}), - ...(body.scope !== undefined ? { scope: jsonb(body.scope) } : {}), - ...(body.rule !== undefined ? { rule: jsonb(body.rule) } : {}), + ...(body.label !== undefined ? { name: body.label } : {}), ...(body.description !== undefined ? { description: body.description } : {}), ...(body.evidence !== undefined ? { evidence: jsonb(body.evidence) } : {}), + scope: jsonb(next.scope), + legacy_kind: kind, + legacy_rule: jsonb(next.rule), + layers: JSON.stringify(compiled.layers), + hard_excludes: JSON.stringify(compiled.hardExcludes), + ladder_required: kind !== "soft", + rule_string: compiled.layers.length ? serializeLayers(compiled.layers) : null, fingerprint: fp, - status, + enabled, + suppressed_at: enabled ? null : (cur!.suppressed_at ?? new Date()), revision: cur!.revision + 1, }) - .where("id", "=", params.id) + .where("id", "=", cur!.id) .returningAll() .executeTakeFirstOrThrow(); - await auditPref(trx, c, "updated", next_, changed); - return next_; + const after = toTag(row); + const changed = (["label", "kind", "scope", "rule", "description", "evidence", "status"] as const).filter( + (k) => body[k] !== undefined && JSON.stringify(after[k]) !== JSON.stringify(before[k]), + ); + await auditPref(trx, c, "updated", tagSnapshot(row), changed); + return row; }); - return toApi(r); + return toTag(r); }, { tag: TAG }, ); @@ -762,20 +838,22 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { await db() .transaction() .execute(async (trx) => { - const cur = await trx - .selectFrom("preference_tag") - .selectAll() - .where("id", "=", params.id) - .forUpdate() - .executeTakeFirst(); - if (!cur || cur.status === "deleted") notFound("偏好标签"); - await trx - .updateTable("preference_tag") - .set({ status: "deleted", revision: cur!.revision + 1 }) - .where("id", "=", params.id) - .execute(); - await auditPref(trx, c, "deleted", { ...cur!, status: "deleted", revision: cur!.revision + 1 }, []); - await suppress(trx, cur!.id, cur!.fingerprint, actorOf(c), "deleted"); + const cur = await liveTag(trx, params.id).forUpdate().executeTakeFirst(); + if (!cur) notFound("偏好标签"); + const row = await trx + .updateTable("rule_group") + .set({ + deleted_at: new Date(), + enabled: false, + default_order: null, + suppressed_at: cur!.suppressed_at ?? new Date(), + revision: cur!.revision + 1, + }) + .where("id", "=", cur!.id) + .returningAll() + .executeTakeFirstOrThrow(); + await auditPref(trx, c, "deleted", tagSnapshot(row), []); + await suppressRuleGroup(trx, cur!.id, cur!.fingerprint, actorOf(c), "deleted"); }); return { ok: true as const }; }, @@ -786,7 +864,12 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { app, Api.preferenceTags.revisions, async (_c, { params }) => { - const tag = await db().selectFrom("preference_tag").select("id").where("id", "=", params.id).executeTakeFirst(); + const tag = await db() + .selectFrom("rule_group") + .select("id") + .where("id", "=", params.id) + .where("legacy_kind", "is not", null) + .executeTakeFirst(); if (!tag) notFound("偏好标签"); const rows = await db() .selectFrom("audit_event") @@ -829,76 +912,81 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { { tag: TAG }, ); - // ── 抑制记录 ── - mount( - app, - Api.preferenceSuppressions.list, - async (_c, { query }) => { - let q = db().selectFrom("preference_suppression").selectAll(); - if (query.active) q = q.where("active", "=", query.active === "true"); - const p = await paginate(q, query); - return { items: toApiList(p.rows), nextCursor: p.nextCursor }; - }, - { tag: TAG }, - ); - mount( - app, - Api.preferenceSuppressions.get, - async (_c, { params }) => { - const r = await db() - .selectFrom("preference_suppression") - .selectAll() - .where("id", "=", params.id) - .executeTakeFirst(); - return toApi(r ?? notFound("抑制记录")); - }, - { tag: TAG }, - ); - mount( - app, - Api.preferenceSuppressions.create, - async (c, { body }) => { - const r = await db() - .insertInto("preference_suppression") - .values({ - preference_tag_id: body.preferenceTagId ?? null, - fingerprint: body.fingerprint, - reason: body.reason ?? null, - evidence: jsonb(body.evidence), - suppressed_by: actorOf(c), - }) - .returningAll() - .executeTakeFirstOrThrow(); - return toApi(r); - }, - { tag: TAG }, - ); - mount( - app, - Api.preferenceSuppressions.update, - async (c, { params, body }) => { - const r = await db() - .updateTable("preference_suppression") - .set({ - active: false, - lifted_at: new Date(), - lifted_by: actorOf(c), - ...(body.reason ? { reason: body.reason } : {}), - }) - .where("id", "=", params.id) - .returningAll() - .executeTakeFirst(); - return toApi(r ?? notFound("抑制记录")); - }, - { tag: TAG }, - ); -} + // ── 抑制记录(/rules/suppressions 为新路径,/preference-suppressions 保留为别名)── + // preferenceTagId:旧标签行的 id;新建的规则组只有 ruleGroupId,为兼容旧界面同时回填到 preferenceTagId + const supToApi = (r: PreferenceSuppressionRow) => { + const o = toApi(r); + return { ...o, preferenceTagId: r.preference_tag_id ?? r.rule_group_id }; + }; + for (const S of [Api.ruleSuppressions, Api.preferenceSuppressions]) { + mount( + app, + S.list, + async (_c, { query }) => { + let q = db().selectFrom("preference_suppression").selectAll(); + if (query.active) q = q.where("active", "=", query.active === "true"); + const p = await paginate(q, query); + return { items: p.rows.map(supToApi), nextCursor: p.nextCursor }; + }, + { tag: TAG }, + ); + mount( + app, + S.get, + async (_c, { params }) => { + const r = await db() + .selectFrom("preference_suppression") + .selectAll() + .where("id", "=", params.id) + .executeTakeFirst(); + return supToApi(r ?? notFound("抑制记录")); + }, + { tag: TAG }, + ); + mount( + app, + S.create, + async (c, { body }) => { + const gid: string | null = body.ruleGroupId ?? body.preferenceTagId ?? null; + const r = await db() + .insertInto("preference_suppression") + .values({ + preference_tag_id: gid ? sql`(select id from preference_tag where id = ${gid})` : null, + rule_group_id: gid ? sql`(select id from rule_group where id = ${gid})` : null, + fingerprint: body.fingerprint, + reason: body.reason ?? null, + evidence: jsonb(body.evidence), + suppressed_by: actorOf(c), + }) + .returningAll() + .executeTakeFirstOrThrow(); + return supToApi(r); + }, + { tag: TAG }, + ); + mount( + app, + S.update, + async (c, { params, body }) => { + const r = await db() + .updateTable("preference_suppression") + .set({ + active: false, + lifted_at: new Date(), + lifted_by: actorOf(c), + ...(body.reason ? { reason: body.reason } : {}), + }) + .where("id", "=", params.id) + .returningAll() + .executeTakeFirst(); + return supToApi(r ?? notFound("抑制记录")); + }, + { tag: TAG }, + ); + } -async function suppress(trx: Transaction, tagId: string, fingerprint: string, by: string, reason: string) { - await trx - .insertInto("preference_suppression") - .values({ preference_tag_id: tagId, fingerprint, reason, evidence: jsonb({ tagId }), suppressed_by: by }) - .execute(); + // 规则组 /rules/*:含 /rules/{id},必须最后注册(静态的 /rules/policies、/rules/suppressions 在前) + registerRuleGroupRoutes(app, d); } /** 发布:旧 published → superseded/rolled_back;吊销旧授权;新版本生成持续授权 automation_grant */ diff --git a/apps/server/src/modules/acquisition-organize/services/acquisition.ts b/apps/server/src/modules/acquisition-organize/services/acquisition.ts index 39043dd6..ce13f61c 100644 --- a/apps/server/src/modules/acquisition-organize/services/acquisition.ts +++ b/apps/server/src/modules/acquisition-organize/services/acquisition.ts @@ -29,6 +29,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 { entityOriginalLanguage, loadRuleCatalog, type RuleCatalog, resolveRuleChain } from "./rules.ts"; import { clearDeferral, rangeOf, @@ -192,6 +193,7 @@ export async function createSlotsFromTemplates( fallback: jsonb(t.fallback), upgrade_threshold: jsonb(t.upgradeThreshold), completion: jsonb(t.completion), + rule_group_ids: JSON.stringify(t.ruleGroupIds ?? []), }) .onConflict((oc) => oc.doNothing()) .execute(); @@ -222,6 +224,10 @@ interface Ctx { now: Date; out: EvaluateOutcome; draft: boolean; + /** 规则组目录(一次评估内复用) */ + rules?: Promise; + /** 作品原语言缓存(规则组 CNSUB/CNVOI 放行) */ + langs?: Map>; } async function note(c: Ctx, base: Omit) { @@ -464,8 +470,8 @@ async function evaluateRelease(c: Ctx, rel: SourceReleaseRow, eventUnits: readon } } -/** 站点字段 → 评估上下文(做种数、促销下载系数、副标题/标签、站点 id) */ -function contextOf(rel: SourceReleaseRow): EvaluationContext { +/** 站点字段 → 评估上下文(做种数、促销下载系数、副标题/标签、站点 id、发布后分钟数) */ +export function contextOf(rel: SourceReleaseRow, now: Date = new Date()): EvaluationContext { const promo = (rel.promotion ?? {}) as { kind?: string; raw?: { downloadFactor?: number | null } }; const factor = typeof promo.raw?.downloadFactor === "number" @@ -485,6 +491,9 @@ function contextOf(rel: SourceReleaseRow): EvaluationContext { labels: rel.tags_raw ?? [], ...(rel.seeders !== null && rel.seeders !== undefined ? { seeders: rel.seeders } : {}), ...(factor !== undefined ? { downloadVolumeFactor: factor } : {}), + ...(rel.source_published_at + ? { publishedMinutesAgo: Math.max(0, (now.getTime() - rel.source_published_at.getTime()) / 60_000) } + : {}), }; } @@ -521,7 +530,26 @@ async function evaluateForSlot(c: Ctx, p: SlotInput): Promise<{ planId: string | return done; } const spec = slotSpecFromRow(toApi(slot)); - const ctx = contextOf(rel); + const ctx = contextOf(rel, now); + // 规则组:版本目标 ruleGroupIds → 订阅 ruleGroupIds → 订阅默认链;开升级叠加洗版默认链 + c.rules ??= loadRuleCatalog(db); + const chain = resolveRuleChain(await c.rules, { + kind: "acquisition", + slotRuleGroupIds: (slot.rule_group_ids ?? []) as string[], + subRuleGroupIds: params.ruleGroupIds, + upgrade: !!spec.upgrade?.enabled, + mediaKind: subKindOf(p.entity.kind), + entityId: p.entity.id, + siteId: rel.site_id, + subscriptionId: sub?.id, + }); + if (chain.groups.length) { + spec.rules = { groups: chain.groups, atoms: chain.atoms }; + c.langs ??= new Map(); + if (!c.langs.has(p.entity.id)) c.langs.set(p.entity.id, entityOriginalLanguage(db, p.entity.id)); + const lang = await c.langs.get(p.entity.id); + if (lang) ctx.originalLanguage = lang; + } const inflight = await slotInflightPlans(db, slot.id); const inflightKeys = inflight.flatMap((x) => x.unitKeys); const defer = async (reason: string, unitKeys: string[], mode: AcquisitionMode | null = null) => { diff --git a/apps/server/src/modules/acquisition-organize/services/rules.ts b/apps/server/src/modules/acquisition-organize/services/rules.ts new file mode 100644 index 00000000..270eff2e --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/services/rules.ts @@ -0,0 +1,347 @@ +import type { + RuleExpr as ApiRuleExpr, + RuleAtom, + RuleGroup, + RuleGroupCreate, + RuleGroupKind, + RulePreviewRequest, + RulePreviewResult, +} from "@umt/contracts"; +import type { RuleAtomRow, RuleGroupRow, SourceReleaseRow } from "@umt/db"; +import { + atomsOf, + BUILTIN_ATOM_MAP, + type CustomAtom, + catalogOf, + compileMpPattern, + type EvaluationContext, + evaluateRuleChain, + normalizeExpr, + type ParsedRelease, + parseExpr, + parseRelease, + parseRuleString, + type RuleExpr, + type RuleGroupSpec, + type RuleLayer, + type RuleParseError, + type SlotRuleChain, + serializeExpr, + serializeLayers, + validateKeywordPattern, +} from "@umt/domain"; +import { ApiException } from "../../../http/errors.ts"; +import { hashJson, toApi } from "../util.ts"; +import type { AoDeps, Exec } from "./common.ts"; + +// 规则组(W9-D):目录加载、引用链解析(版本目标 → 订阅 → 默认链,开升级叠加洗版默认)、指纹、输入规范化、试算。 + +export interface RuleCatalog { + /** 未删除的全部规则组(含停用,解析链时再过滤) */ + groups: Map; + atoms: CustomAtom[]; +} + +export function atomFromRow( + r: Pick< + RuleAtomRow, + | "key" + | "name" + | "include" + | "exclude" + | "match_fields" + | "unit_size_mb" + | "min_seeders" + | "publish_delay_minutes" + | "free" + >, +): CustomAtom { + const arr = (v: unknown) => (Array.isArray(v) ? (v as string[]) : []); + const range = (v: unknown) => (v && typeof v === "object" ? (v as { min?: number; max?: number }) : undefined); + const out: CustomAtom = { id: r.key }; + if (r.name) out.name = r.name; + if (arr(r.include).length) out.include = arr(r.include); + if (arr(r.exclude).length) out.exclude = arr(r.exclude); + if (arr(r.match_fields).length) out.matchFields = arr(r.match_fields) as CustomAtom["matchFields"]; + const size = range(r.unit_size_mb); + if (size) out.unitSizeMB = size; + if (r.min_seeders !== null) out.minSeeders = r.min_seeders; + const pd = range(r.publish_delay_minutes); + if (pd) out.publishDelayMinutes = pd; + if (r.free) out.free = true; + return out; +} + +export async function loadRuleCatalog(exec: Exec): Promise { + const [groups, atoms] = await Promise.all([ + exec.selectFrom("rule_group").selectAll().where("deleted_at", "is", null).execute(), + exec.selectFrom("rule_atom").selectAll().execute(), + ]); + return { groups: new Map(groups.map((g) => [g.id, g])), atoms: atoms.map(atomFromRow) }; +} + +export function groupSpecOf(r: RuleGroupRow): RuleGroupSpec { + return { + id: r.id, + name: r.name, + revision: r.revision, + layers: (r.layers ?? []) as RuleLayer[], + hardExcludes: (r.hard_excludes ?? []) as RuleExpr[], + ladderRequired: r.ladder_required, + }; +} + +export function ruleGroupToApi(r: RuleGroupRow): RuleGroup { + const { legacy_rule: _legacy, ...rest } = r; + return toApi(rest); +} + +export const ruleAtomToApi = (r: RuleAtomRow) => toApi(r); + +export interface RuleTarget { + mediaKind?: "movie" | "series" | "variety" | undefined; + entityId?: string | undefined; + siteId?: string | undefined; + subscriptionId?: string | undefined; +} + +interface Scope { + mediaTypes?: string[]; + entityIds?: string[]; + siteIds?: string[]; + subscriptionId?: string; +} + +/** scope 过滤;空 = 全部。私有组(subscriptionId)只对该订阅生效 */ +export function inScope(r: RuleGroupRow, t: RuleTarget): boolean { + const s = (r.scope ?? {}) as Scope; + if (s.subscriptionId && s.subscriptionId !== t.subscriptionId) return false; + if (s.mediaTypes?.length && t.mediaKind && !s.mediaTypes.includes(t.mediaKind)) return false; + if (s.entityIds?.length && (!t.entityId || !s.entityIds.includes(t.entityId))) return false; + if (s.siteIds?.length && t.siteId && !s.siteIds.includes(t.siteId)) return false; + return true; +} + +/** 某类默认链(启用、未删除、default_order 升序) */ +export function defaultChainIds(cat: RuleCatalog, kind: RuleGroupKind): string[] { + return [...cat.groups.values()] + .filter((g) => g.kind === kind && g.default_order !== null && g.enabled) + .sort((a, b) => a.default_order! - b.default_order! || a.created_at.getTime() - b.created_at.getTime()) + .map((g) => g.id); +} + +export interface ResolveInput extends RuleTarget { + /** 版本目标 rule_group_ids(非空时优先) */ + slotRuleGroupIds?: readonly string[] | undefined; + /** 订阅 params.ruleGroupIds */ + subRuleGroupIds?: readonly string[] | undefined; + kind: "acquisition" | "search"; + /** 版本目标开了升级:叠加洗版默认链 */ + upgrade?: boolean | undefined; +} + +/** + * 引用链:订阅私有组 → (版本目标 ruleGroupIds || 订阅 ruleGroupIds || 该类默认链)→ 开升级时叠加洗版默认链。 + * 按 scope 过滤、跳过停用/已删;pri 取链上最后一个带层的组(对齐 MP 多组串联)。 + */ +export function resolveRuleChain(cat: RuleCatalog, input: ResolveInput): SlotRuleChain & { ids: string[] } { + const explicit = input.slotRuleGroupIds?.length ? input.slotRuleGroupIds : (input.subRuleGroupIds ?? []); + const ids: string[] = []; + const push = (id: string) => { + if (!ids.includes(id)) ids.push(id); + }; + if (input.subscriptionId) + for (const g of cat.groups.values()) + if ((g.scope as Scope | null)?.subscriptionId === input.subscriptionId) push(g.id); + for (const id of explicit.length ? explicit : defaultChainIds(cat, input.kind)) push(id); + if (input.kind === "acquisition" && input.upgrade) for (const id of defaultChainIds(cat, "upgrade")) push(id); + const rows = ids + .map((id) => cat.groups.get(id)) + .filter((g): g is RuleGroupRow => !!g && g.enabled && inScope(g, input)); + return { ids: rows.map((r) => r.id), groups: rows.map(groupSpecOf), atoms: cat.atoms }; +} + +/** 语义指纹:类型 + 范围 + 层 + 硬排除(规范化规则串)+ 是否必须命中 */ +export function ruleFingerprint(g: { + kind: string; + scope: unknown; + layers: readonly RuleLayer[]; + hardExcludes: readonly RuleExpr[]; + ladderRequired: boolean; +}): string { + return hashJson({ + kind: g.kind, + scope: g.scope ?? {}, + layers: g.layers.map((l) => serializeExpr(normalizeExpr(l.expr))), + hardExcludes: g.hardExcludes.map((x) => serializeExpr(normalizeExpr(x))).sort(), + ladderRequired: g.ladderRequired, + }); +} + +const fmtErrors = (errs: readonly RuleParseError[]) => + errs.map((e) => `${e.message}(第 ${e.layer + 1} 层,位置 ${e.position})`).join(";"); + +/** 表达式里的原子都要存在(内置且支持,或自定义) */ +function checkAtoms(e: RuleExpr, custom: ReadonlySet): string[] { + const bad: string[] = []; + for (const id of atomsOf(e)) { + const b = BUILTIN_ATOM_MAP.get(id); + if (b ? !b.supported : !custom.has(id)) bad.push(b ? `${id}(不支持)` : `${id}(不存在)`); + } + return bad; +} + +/** 请求体 → 规范化的层 / 硬排除(ruleString 优先于 layers;硬排除元素可为单层规则串) */ +export function normalizeGroupInput( + body: Pick, + customKeys: Iterable, +): { layers?: RuleLayer[]; hardExcludes?: RuleExpr[] } { + const custom = new Set([...customKeys].map((k) => k.toUpperCase())); + const out: { layers?: RuleLayer[]; hardExcludes?: RuleExpr[] } = {}; + if (body.ruleString !== undefined) { + if (body.ruleString.trim() === "") out.layers = []; + else { + const p = parseRuleString(body.ruleString, { customAtoms: custom }); + if (p.errors.length) throw new ApiException("validation_failed", `规则串解析失败:${fmtErrors(p.errors)}`); + out.layers = p.layers; + } + } else if (body.layers !== undefined) { + out.layers = body.layers.map((l) => ({ + ...(l.name ? { name: l.name } : {}), + expr: normalizeExpr(l.expr as RuleExpr), + })); + } + if (body.hardExcludes !== undefined) { + out.hardExcludes = body.hardExcludes.map((x: string | ApiRuleExpr, i: number) => { + if (typeof x !== "string") return normalizeExpr(x as RuleExpr); + if (x.includes(">")) throw new ApiException("validation_failed", `硬排除第 ${i + 1} 条不能分层(去掉 >)`); + const p = parseExpr(x, { customAtoms: custom }); + if (!p.expr) throw new ApiException("validation_failed", `硬排除第 ${i + 1} 条解析失败:${fmtErrors(p.errors)}`); + return p.expr; + }); + } + const bad = [...(out.layers ?? []).map((l) => l.expr), ...(out.hardExcludes ?? [])].flatMap((e) => + checkAtoms(e, custom), + ); + if (bad.length) throw new ApiException("validation_failed", `条件无效:${[...new Set(bad)].join("、")}`); + return out; +} + +export const ruleStringOf = (layers: readonly RuleLayer[]) => (layers.length ? serializeLayers(layers) : null); + +/** 自定义条件正则:MP 语法(去掉 (?i))+ 防灾难回溯白名单 */ +export function checkAtomPatterns(ps: readonly string[] | undefined): void { + for (const p of ps ?? []) { + const stripped = p.replace(/\(\?i\)/g, ""); + const why = validateKeywordPattern(stripped) ?? (compileMpPattern(p) ? null : "invalid"); + if (why) throw new ApiException("validation_failed", `正则不可用(${why}):${p}`); + } +} + +/** 作品原语言(元数据字段 originalLanguage,最近一次) */ +export async function entityOriginalLanguage(exec: Exec, entityId: string | null): Promise { + if (!entityId) return undefined; + const r = await exec + .selectFrom("metadata_field") + .select("value") + .where("media_entity_id", "=", entityId) + .where("field", "=", "originalLanguage") + .orderBy("fetched_at", "desc") + .limit(1) + .executeTakeFirst(); + return typeof r?.value === "string" && r.value ? r.value : undefined; +} + +/** 试算:对最近 N 条已识别资源(或指定资源 / 粘贴标题)跑本组 */ +export async function previewRuleGroup( + d: AoDeps, + row: RuleGroupRow, + body: RulePreviewRequest, + helpers: { + parsedOf: (r: SourceReleaseRow) => ParsedRelease; + contextOf: (r: SourceReleaseRow) => EvaluationContext; + }, +): Promise { + const db = d.ctx.db; + const cat = await loadRuleCatalog(db); + const spec = groupSpecOf(row); + if (body.draft) { + const n = normalizeGroupInput( + body.draft, + cat.atoms.map((a) => a.id), + ); + if (n.layers) spec.layers = n.layers; + if (n.hardExcludes) spec.hardExcludes = n.hardExcludes; + if (body.draft.ladderRequired !== undefined) spec.ladderRequired = body.draft.ladderRequired; + } + const catalog = catalogOf(cat.atoms); + const samples: { + id: string | null; + title: string; + subtitle: string | null; + parsed: ParsedRelease; + ctx: EvaluationContext; + }[] = []; + for (const t of body.titles ?? []) + samples.push({ + id: null, + title: t.title, + subtitle: t.subtitle ?? null, + parsed: parseRelease(t.title, t.subtitle ? { subtitle: t.subtitle } : {}), + ctx: t.subtitle ? { subtitle: t.subtitle } : {}, + }); + // 指定资源 → 只取这些;否则取最近 N 条(只粘贴标题且没给 limit 时不取) + if (body.sourceReleaseIds?.length || !body.titles?.length || body.limit !== undefined) { + let q = db.selectFrom("source_release").selectAll(); + q = body.sourceReleaseIds?.length + ? q.where("id", "in", body.sourceReleaseIds) + : q + .where("resolution_status", "=", "resolved") + .orderBy("first_seen_at", "desc") + .limit(body.limit ?? 50); + const langs = new Map(); + for (const r of await q.execute()) { + const ctx = helpers.contextOf(r); + if (r.media_entity_id) { + if (!langs.has(r.media_entity_id)) + langs.set(r.media_entity_id, await entityOriginalLanguage(db, r.media_entity_id)); + const lang = langs.get(r.media_entity_id); + if (lang) ctx.originalLanguage = lang; + } + samples.push({ id: r.id, title: r.title, subtitle: r.subtitle, parsed: helpers.parsedOf(r), ctx }); + } + } + const items = samples.map((s) => { + const res = evaluateRuleChain([spec], s.parsed, s.ctx, catalog); + const g = res.groups[0]!; + const reasons = [ + ...(g.excludedBy.length ? ["hard_exclude"] : []), + ...(g.pri !== null && g.level === null && spec.ladderRequired !== false ? ["no_layer_matched"] : []), + ]; + return { + sourceReleaseId: s.id, + title: s.title, + subtitle: s.subtitle, + passed: res.passed, + pri: res.passed ? (res.pri ?? 100) : null, + level: g.level, + excludedBy: g.excludedBy, + reasons, + trace: res.trace, + }; + }); + const hist = new Map(); + for (const it of items) hist.set(it.pri, (hist.get(it.pri) ?? 0) + 1); + return { + ruleGroupId: row.id, + revision: row.revision, + evaluatedAt: d.ports.now().toISOString(), + sampleSize: items.length, + passed: items.filter((i) => i.passed).length, + rejected: items.filter((i) => !i.passed).length, + priHistogram: [...hist.entries()] + .sort((a, b) => (b[0] ?? -1) - (a[0] ?? -1)) + .map(([pri, count]) => ({ pri, count })), + items, + }; +} diff --git a/apps/server/test/rules/rule-groups.test.ts b/apps/server/test/rules/rule-groups.test.ts new file mode 100644 index 00000000..c8e96d66 --- /dev/null +++ b/apps/server/test/rules/rule-groups.test.ts @@ -0,0 +1,282 @@ +import { makeTorrent } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + createHarness, + fakeDisk, + type Harness, + pgAvailable, + seedBase, + seedEntity, + seedRelease, + seedSlot, +} from "../../src/modules/acquisition-organize/__tests__/harness.ts"; +import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts"; + +// W9-D 规则组:关键字表、规则串解析/序列化、CRUD + 修订、抑制记录、试算、默认链、评估接入、旧端点别名。 + +const hasPg = await pgAvailable(); + +describe.skipIf(!hasPg)("规则组 API(真实 PG)", () => { + let h: Harness; + const disk = fakeDisk(1_000_000_000); + let base: Awaited>; + + beforeAll(async () => { + disk.setPct(10); + h = await createHarness("rulegroups", { diskUsage: disk.fn }); + base = await seedBase(h.db, { + libraryPath: "/tmp/nonexistent-lib", + cachePath: "/tmp", + qbUrl: "http://127.0.0.1:9", + }); + }, 60_000); + afterAll(async () => { + await h?.close(); + }); + + it("关键字表对齐 MP;规则串解析 / 序列化 / 错误位置", async () => { + const kw = await h.api("GET", "/rules/keywords"); + expect(kw.status).toBe(200); + const byId = new Map(kw.body.data.items.map((k: { id: string }) => [k.id, k])); + expect(byId.get("4K")).toMatchObject({ builtin: true, supported: true, basis: "field" }); + expect(byId.get("CNSUB")).toMatchObject({ mp: { tmdb: { original_language: "zh,cn" } } }); + expect(byId.get("FLAC")).toMatchObject({ supported: false }); + + const p = await h.api("POST", "/rules/parse", { ruleString: "specsub & cnsub > CNSUB & 4K > 1080P & !BLU" }); + expect(p.status).toBe(200); + expect(p.body.data.ruleString).toBe("SPECSUB & CNSUB > CNSUB & 4K > 1080P & !BLU"); + expect(p.body.data.layers).toHaveLength(3); + const bad = await h.api("POST", "/rules/parse", { ruleString: "4K & CNSUBB > (1080P" }); + expect(bad.body.data.ruleString).toBeNull(); + expect(bad.body.data.errors[0]).toMatchObject({ code: "unknown_atom", position: 5, suggestions: ["CNSUB"] }); + const ser = await h.api("POST", "/rules/serialize", { layers: p.body.data.layers }); + expect(ser.body.data.ruleString).toBe("SPECSUB & CNSUB > CNSUB & 4K > 1080P & !BLU"); + }); + + it("默认三套组;CRUD + revision 乐观锁 + 修订历史;名称唯一", async () => { + const defs = await h.api("GET", "/rules/defaults"); + expect(defs.body.data.acquisition).toHaveLength(1); + expect(defs.body.data.upgrade).toHaveLength(1); + expect(defs.body.data.search).toHaveLength(1); + + const c = await h.api("POST", "/rules", { name: "4K 优先", ruleString: "4K > 1080P", hardExcludes: ["BLU"] }); + expect(c.status).toBe(201); + expect(c.body.data).toMatchObject({ + kind: "acquisition", + source: "manual", + ruleString: "4K > 1080P", + ladderRequired: true, + revision: 1, + hardExcludes: [{ op: "atom", id: "BLU" }], + }); + expect((await h.api("POST", "/rules", { name: "4K 优先" })).status).toBe(409); + expect((await h.api("POST", "/rules", { name: "坏", ruleString: "4K & NOPE" })).status).toBe(422); + const id = c.body.data.id; + expect((await h.api("PATCH", `/rules/${id}`, { expectedRevision: 5, ruleString: "4K" })).status).toBe(409); + const u = await h.api("PATCH", `/rules/${id}`, { expectedRevision: 1, ruleString: "4K & !REMUX > 1080P" }); + expect(u.body.data).toMatchObject({ revision: 2, ruleString: "4K & !REMUX > 1080P" }); + const rev = await h.api("GET", `/rules/${id}/revisions`); + expect(rev.body.data.items.map((r: { action: string }) => r.action)).toEqual([ + "rule_group.updated", + "rule_group.created", + ]); + expect(rev.body.data.items[0].changedFields).toEqual(["layers"]); + const list = await h.api("GET", "/rules?kind=acquisition"); + expect(list.body.data.items.some((g: { id: string }) => g.id === id)).toBe(true); + }); + + it("自定义条件:不能覆盖内置;被引用不能删", async () => { + expect((await h.api("POST", "/rules/atoms", { key: "4K" })).status).toBe(409); + expect((await h.api("POST", "/rules/atoms", { key: "BADRE", include: ["(a+)+"] })).status).toBe(422); + const a = await h.api("POST", "/rules/atoms", { key: "sizemovie", unitSizeMb: { min: 2000, max: 30000 } }); + expect(a.status).toBe(201); + expect(a.body.data.key).toBe("SIZEMOVIE"); + const g = await h.api("POST", "/rules", { name: "体积", ruleString: "SIZEMOVIE" }); + expect(g.status).toBe(201); + expect((await h.api("DELETE", `/rules/atoms/${a.body.data.id}`)).status).toBe(409); + const kw = await h.api("GET", "/rules/keywords"); + expect(kw.body.data.items.find((k: { id: string }) => k.id === "SIZEMOVIE")).toMatchObject({ builtin: false }); + }); + + it("抑制:停用/删除写抑制记录;Agent 不能加回(409);手动新建解除", async () => { + const body = { name: "不要 HDR", source: "agent", hardExcludes: ["HDR"] }; + const c = await h.api("POST", "/rules", body); + expect(c.status).toBe(201); + expect((await h.api("POST", "/rules", { ...body, name: "不要 HDR 2" })).status).toBe(409); + const off = await h.api("PATCH", `/rules/${c.body.data.id}`, { enabled: false }); + expect(off.body.data.suppressedAt).not.toBeNull(); + const sup = await h.api("GET", "/rules/suppressions?active=true"); + expect(sup.body.data.items.map((s: { fingerprint: string }) => s.fingerprint)).toContain(c.body.data.fingerprint); + expect((await h.api("DELETE", `/rules/${c.body.data.id}`)).status).toBe(200); + expect((await h.api("POST", "/rules", { ...body, name: "又来" })).status).toBe(409); + const again = await h.api("POST", "/rules", { ...body, name: "用户自己加", source: "manual" }); + expect(again.status).toBe(201); + const still = await h.api("GET", "/preference-suppressions?active=true"); + expect(still.body.data.items.map((s: { fingerprint: string }) => s.fingerprint)).not.toContain( + c.body.data.fingerprint, + ); + }); + + it("试算:最近资源 + 粘贴标题,给 pri / 命中层 / 拒绝原因", async () => { + const e = await seedEntity(h.db, { kind: "movie", title: "试算片", year: 2024 }); + for (const [i, t] of [ + "Try.2024.2160p.WEB-DL.H265-A", + "Try.2024.1080p.BluRay.x264-B", + "Try.2024.720p.WEB-DL.x264-C", + ].entries()) + await seedRelease(h.db, { + siteId: base.siteId, + entityId: e, + title: t, + size: 1000, + hash: `${"ab".repeat(19)}${i}${i}`, + }); + const g = await h.api("POST", "/rules", { name: "试算组", ruleString: "4K > 1080P", hardExcludes: ["BLU"] }); + const pv = await h.api("POST", `/rules/${g.body.data.id}/preview`, { + limit: 20, + titles: [{ title: "X.2023.1080p.Blu-ray.AVC.DTS-HD.MA-GRP" }], + sourceReleaseIds: [], + }); + expect(pv.status).toBe(200); + const byTitle = new Map(pv.body.data.items.map((i: { title: string }) => [i.title, i])); + expect(byTitle.get("Try.2024.2160p.WEB-DL.H265-A")).toMatchObject({ passed: true, pri: 100, level: 0 }); + expect(byTitle.get("Try.2024.1080p.BluRay.x264-B")).toMatchObject({ passed: true, pri: 99 }); + expect(byTitle.get("Try.2024.720p.WEB-DL.x264-C")).toMatchObject({ passed: false, reasons: ["no_layer_matched"] }); + expect(byTitle.get("X.2023.1080p.Blu-ray.AVC.DTS-HD.MA-GRP")).toMatchObject({ + passed: false, + excludedBy: ["BLU"], + reasons: ["hard_exclude"], + }); + const draft = await h.api("POST", `/rules/${g.body.data.id}/preview`, { + titles: [{ title: "Try.2024.720p.WEB-DL.x264-C" }], + draft: { ruleString: "720P" }, + }); + expect(draft.body.data.items[0]).toMatchObject({ passed: true, pri: 100 }); + }); + + it("评估接入:版本目标引用规则组 → 硬排除拒绝、pri 进候选评估 trace;被引用的组不能删", async () => { + const e = await seedEntity(h.db, { kind: "movie", title: "规则片", year: 2024 }); + const slot = await seedSlot(h.db, { entityId: e }); + const g = await h.api("POST", "/rules", { name: "片源", ruleString: "4K > 1080P", hardExcludes: ["REMUX"] }); + const up = await h.api("PATCH", `/acquisition-slots/${slot}`, { + expectedRevision: 1, + ruleGroupIds: [g.body.data.id], + }); + expect(up.body.data.ruleGroupIds).toEqual([g.body.data.id]); + expect( + ( + await h.api("PATCH", `/acquisition-slots/${slot}`, { + expectedRevision: 2, + ruleGroupIds: ["7d3f7f5e-8a8c-4b8e-9f0a-1b2c3d4e5f60"], + }) + ).status, + ).toBe(400); + + const t1 = makeTorrent({ name: "Rule.2024.1080p.BluRay.REMUX.mkv", length: 1000, pieceSeed: 41 }); + const r1 = await seedRelease(h.db, { + siteId: base.siteId, + entityId: e, + title: "Rule.2024.1080p.BluRay.REMUX.AVC-EPSiLON", + size: 1000, + hash: t1.hash, + }); + const ev1 = await evaluateAcquisition(h.d, { sourceReleaseId: r1 }); + expect(ev1.plans).toHaveLength(0); + const cand = async (rid: string) => + ( + await h.db + .selectFrom("artifact") + .select("content") + .where("subject_id", "=", slot) + .where("title", "=", "candidate_evaluation") + .execute() + ) + .map( + (a) => + a.content as { + sourceReleaseId: string; + decision: string; + pri: number | null; + trace: string[]; + rejected: { code: string }[]; + reasons: string[]; + }, + ) + .find((x) => x.sourceReleaseId === rid)!; + const c1 = await cand(r1); + expect(c1.decision).toBe("reject"); + expect(c1.rejected.map((x) => x.code)).toContain("rule.hard_exclude"); + expect(c1.trace.join("\n")).toContain("规则组「片源」@r1 硬排除 REMUX 命中"); + + const t2 = makeTorrent({ name: "Rule.2024.1080p.WEB-DL.mkv", length: 1000, pieceSeed: 42 }); + const r2 = await seedRelease(h.db, { + siteId: base.siteId, + entityId: e, + title: "Rule.2024.1080p.WEB-DL.H264-GRP", + size: 1000, + hash: t2.hash, + }); + const ev2 = await evaluateAcquisition(h.d, { sourceReleaseId: r2 }); + expect(ev2.plans).toHaveLength(1); + const c2 = await cand(r2); + expect(c2).toMatchObject({ decision: "accept", pri: 99 }); + expect(c2.reasons).toContain(`rule_group:${g.body.data.id}@1`); + + expect((await h.api("DELETE", `/rules/${g.body.data.id}`)).status).toBe(409); + }); + + it("默认链:订阅默认组有层时对未引用规则组的版本目标生效;PUT /rules/defaults 校验类型", async () => { + const defs = (await h.api("GET", "/rules/defaults")).body.data; + const dflt = defs.acquisition[0]; + expect((await h.api("PUT", "/rules/defaults", { ...defs, acquisition: defs.upgrade })).status).toBe(400); + await h.api("PATCH", `/rules/${dflt}`, { ruleString: "WEBDL" }); + const e = await seedEntity(h.db, { kind: "movie", title: "默认片", year: 2024 }); + const slot = await seedSlot(h.db, { entityId: e }); + const t = makeTorrent({ name: "Dflt.2024.1080p.BluRay.mkv", length: 1000, pieceSeed: 51 }); + const r = await seedRelease(h.db, { + siteId: base.siteId, + entityId: e, + title: "Dflt.2024.1080p.BluRay.x264-GRP", + size: 1000, + hash: t.hash, + }); + const ev = await evaluateAcquisition(h.d, { sourceReleaseId: r }); + expect(ev.plans).toHaveLength(0); + const a = await h.db + .selectFrom("artifact") + .select("content") + .where("subject_id", "=", slot) + .where("title", "=", "candidate_evaluation") + .executeTakeFirstOrThrow(); + expect((a.content as { rejected: { code: string }[] }).rejected.map((x) => x.code)).toContain( + "rule.no_layer_matched", + ); + // 复原,避免影响其它用例 + await h.api("PATCH", `/rules/${dflt}`, { ruleString: "" }); + }); + + it("旧端点别名:偏好标签落在规则组;/rules/policies 与 /policy-revisions 同一数据", async () => { + const tag = await h.api("POST", "/preference-tags", { + label: "不要国配", + kind: "hard", + rule: { ruleString: "CNVOI" }, + origin: "conversation", + }); + expect(tag.status).toBe(201); + const g = await h.api("GET", `/rules/${tag.body.data.id}`); + expect(g.body.data).toMatchObject({ + legacyKind: "hard", + source: "chat", + hardExcludes: [{ op: "atom", id: "CNVOI" }], + }); + const pol = await h.api("POST", "/rules/policies", { + policyKey: "w9d", + kind: "acquisition", + title: "t", + rules: {}, + }); + expect(pol.status).toBe(201); + const old = await h.api("GET", `/policy-revisions/${pol.body.data.id}`); + expect(old.body.data.policyKey).toBe("w9d"); + }); +}); diff --git a/packages/contracts/src/api.ts b/packages/contracts/src/api.ts index cc619364..05cd8600 100644 --- a/packages/contracts/src/api.ts +++ b/packages/contracts/src/api.ts @@ -751,6 +751,166 @@ export const Api = { R, ), + // ── 规则(W9-D):规则组 = 收录过滤条件 + 偏好标签合一。静态路径先于 /rules/{id} 注册 ── + rules: { + keywords: ep({ + method: "get", + path: "/rules/keywords", + scope: "read", + summary: "关键字表:内置(对齐 MP,含判定依据与上游正则)+ 自定义条件", + response: policy.RuleKeywordList, + }), + parse: ep({ + method: "post", + path: "/rules/parse", + scope: "read", + summary: "规则串 → 结构化层(`>` 分层,层内 ! & | 与括号);错误逐条给位置与相近名", + body: policy.RuleParseRequest, + response: policy.RuleParseResult, + }), + serialize: ep({ + method: "post", + path: "/rules/serialize", + scope: "read", + summary: "结构化层 → 规范化规则串", + body: policy.RuleSerializeRequest, + response: policy.RuleSerializeResult, + }), + defaults: ep({ + method: "get", + path: "/rules/defaults", + scope: "read", + summary: "三套默认规则组链:订阅 / 洗版 / 搜索", + response: policy.RuleDefaults, + }), + setDefaults: ep({ + method: "put", + path: "/rules/defaults", + scope: "admin", + summary: "设置默认规则组链(按数组顺序串联,pri 取最后一个带层的组)", + body: policy.RuleDefaults, + response: policy.RuleDefaults, + }), + ...crud( + "/rules", + "规则组", + { + row: policy.RuleGroup, + create: policy.RuleGroupCreate, + update: policy.RuleGroupUpdate, + filters: { + kind: policy.RuleGroupKind.optional(), + source: policy.RuleGroupSource.optional(), + enabled: z.enum(["true", "false"]).optional(), + includeDeleted: z.enum(["true", "false"]).optional(), + q: z.string().max(80).optional(), + }, + }, + R, + ), + /** 删除 = 标记 deleted_at + 写抑制记录 */ + remove: del("/rules/{id}", "删除规则组(留抑制记录;被引用时 409)"), + revisions: ep({ + method: "get", + path: "/rules/{id}/revisions", + scope: "read", + summary: "规则组修订历史(新→旧;取自审计事件)", + params: IdParam, + response: policy.RuleGroupRevisionList, + }), + preview: ep({ + method: "post", + path: "/rules/{id}/preview", + scope: "read", + summary: "试算:对最近 N 条已识别资源(或指定资源/标题)跑本组,给 pri、命中层与拒绝原因", + params: IdParam, + body: policy.RulePreviewRequest, + response: policy.RulePreviewResult, + }), + }, + ruleAtoms: { + ...crud( + "/rules/atoms", + "自定义条件", + { row: policy.RuleAtom, create: policy.RuleAtomCreate, update: policy.RuleAtomUpdate }, + R, + ), + remove: del("/rules/atoms/{id}", "删除自定义条件(被规则组引用时 409)"), + }, + /** /policy-revisions 的新路径(旧路径保留为别名,见 docs/runbook/rules.md) */ + rulePolicies: { + list: ep({ + method: "get", + path: "/rules/policies", + scope: "read", + summary: "收录策略版本列表", + query: listQuery({ policyKey: z.string().optional(), status: policy.PolicyStatus.optional() }), + response: cursorPage(policy.PolicyRevision), + }), + get: ep({ + method: "get", + path: "/rules/policies/{id}", + scope: "read", + summary: "收录策略版本", + params: IdParam, + response: policy.PolicyRevision, + }), + create: ep({ + method: "post", + path: "/rules/policies", + scope: "admin", + summary: "新建策略草案", + body: policy.PolicyRevisionCreate, + response: policy.PolicyRevision, + }), + update: ep({ + method: "patch", + path: "/rules/policies/{id}", + scope: "admin", + summary: "修改策略草案", + params: IdParam, + body: policy.PolicyRevisionUpdate, + response: policy.PolicyRevision, + }), + preview: ep({ + method: "post", + path: "/rules/policies/{id}/preview", + scope: "read", + summary: "预览策略草案", + params: IdParam, + response: policy.PolicyRevision, + }), + publish: ep({ + method: "post", + path: "/rules/policies/{id}/publish", + scope: "admin", + summary: "发布策略版本", + params: IdParam, + body: z.object({ expectedRevision: z.number().int().min(1) }), + response: policy.PolicyRevision, + }), + rollback: ep({ + method: "post", + path: "/rules/policies/{id}/rollback", + scope: "admin", + summary: "回滚到该策略版本", + params: IdParam, + response: policy.PolicyRevision, + }), + }, + /** /preference-suppressions 的新路径(旧路径保留为别名) */ + ruleSuppressions: crud( + "/rules/suppressions", + "规则抑制记录", + { + row: policy.PreferenceSuppression, + create: policy.PreferenceSuppressionCreate, + update: policy.PreferenceSuppressionUpdate, + filters: { active: z.enum(["true", "false"]).optional() }, + }, + R, + ), + // ── 执行 ── tasks: { ...crud( diff --git a/packages/contracts/src/policy.ts b/packages/contracts/src/policy.ts index 8fb747a9..6934cb02 100644 --- a/packages/contracts/src/policy.ts +++ b/packages/contracts/src/policy.ts @@ -101,6 +101,8 @@ export const SubscriptionParams = z .optional(), stopAfter: z.string().optional(), downloaderInstanceId: Id.optional(), + /** 引用的规则组(有序,串联取交集,pri 取最后一个带层的组);空 = 订阅默认组 */ + ruleGroupIds: z.array(Id).optional(), }) .passthrough(); export type SubscriptionParams = z.infer; @@ -266,6 +268,8 @@ export const AcquisitionSlot = EntityBase.extend({ lastEvaluatedAt: Timestamp.nullable(), nextEvaluateAt: Timestamp.nullable(), revision: z.number().int().min(1), + /** 引用的规则组(有序);空 = 订阅的 ruleGroupIds,再空 = 订阅默认组(开升级时叠加洗版默认组) */ + ruleGroupIds: z.array(Id), /** 派生字段(列表返回):当前资产的交付特征摘要,无资产为 null */ currentAsset: z .object({ @@ -293,6 +297,7 @@ export const AcquisitionSlotCreate = AcquisitionSlot.pick({ completion: true, distinctGroup: true, requireDistinctAsset: true, + ruleGroupIds: true, }).partial({ contentUnitId: true, subscriptionId: true, @@ -304,6 +309,7 @@ export const AcquisitionSlotCreate = AcquisitionSlot.pick({ completion: true, distinctGroup: true, requireDistinctAsset: true, + ruleGroupIds: true, }); export type AcquisitionSlotCreate = z.infer; /** 规则可改;status/currentAsset/reservedPlan 只由状态机推进,不经 Update */ @@ -407,6 +413,8 @@ export type PreferenceTagUpdate = z.infer; /** 手动取消标签留下的抑制记录:Agent 不得据同一旧证据自动加回 */ export const PreferenceSuppression = EntityBase.extend({ preferenceTagId: Id.nullable(), + /** 被停用/删除的规则组(偏好标签迁移后与 preferenceTagId 相同) */ + ruleGroupId: Id.nullable().optional(), fingerprint: z.string().min(1), reason: z.string().nullable(), evidence: JsonObject, @@ -433,6 +441,8 @@ export type PreferenceSuppressionUpdate = z.infer; + +// ── 规则组(W9-D):收录策略的过滤条件 + 偏好标签合一 ── + +/** 条件表达式:内置关键字 / 自定义条件的 AND / OR / NOT 树(与 @umt/domain RuleExpr 同形) */ +export type RuleExpr = + | { op: "atom"; id: string } + | { op: "not"; arg: RuleExpr } + | { op: "and"; args: RuleExpr[] } + | { op: "or"; args: RuleExpr[] }; + +function isRuleExpr(v: unknown, depth = 0): v is RuleExpr { + if (depth > 32 || !v || typeof v !== "object") return false; + const o = v as Record; + switch (o.op) { + case "atom": + return typeof o.id === "string" && /^[A-Za-z0-9]{1,32}$/.test(o.id); + case "not": + return isRuleExpr(o.arg, depth + 1); + case "and": + case "or": + return Array.isArray(o.args) && o.args.length >= 1 && o.args.every((a) => isRuleExpr(a, depth + 1)); + default: + return false; + } +} +/** 递归结构在 refine 里校验(OpenAPI 只展开一层) */ +export const RuleExpr = z + .object({ + op: z.enum(["atom", "not", "and", "or"]), + /** op=atom:关键字或自定义条件 id */ + id: z + .string() + .regex(/^[A-Za-z0-9]{1,32}$/) + .optional(), + /** op=not */ + arg: JsonObject.optional(), + /** op=and/or */ + args: z.array(JsonObject).optional(), + }) + .refine(isRuleExpr, { + message: "条件表达式结构不对:{op:atom,id} / {op:not,arg} / {op:and|or,args[]}", + }) as unknown as z.ZodType; + +/** 优先级阶梯的一层;第 k 层(0 起)命中 pri = 100 - k */ +export const RuleLayer = z.object({ name: z.string().max(40).optional(), expr: RuleExpr }); +export type RuleLayer = z.infer; + +/** acquisition 订阅 / upgrade 洗版 / search 搜索 */ +export const RuleGroupKind = z.enum(["acquisition", "upgrade", "search"]); +export type RuleGroupKind = z.infer; +/** 来源:manual 手动 / chat 对话中用户明确说的 / agent Agent 推断 */ +export const RuleGroupSource = z.enum(["manual", "chat", "agent"]); +export type RuleGroupSource = z.infer; + +/** 适用范围;空 = 全部。subscriptionId = 订阅私有组 */ +export const RuleGroupScope = z + .object({ + mediaTypes: z.array(SubscriptionMediaKind).optional(), + entityIds: z.array(Id).optional(), + siteIds: z.array(Id).optional(), + subscriptionId: Id.optional(), + }) + .passthrough(); +export type RuleGroupScope = z.infer; + +export const RuleGroup = EntityBase.extend({ + name: z.string().min(1).max(80), + description: z.string().nullable(), + kind: RuleGroupKind, + scope: RuleGroupScope, + layers: z.array(RuleLayer), + /** 任一命中即拒绝 */ + hardExcludes: z.array(RuleExpr), + /** 一层都不命中即拒绝(MP 语义);false = 软偏好,不中只是 pri 低 */ + ladderRequired: z.boolean(), + /** layers 的规范化规则串(无层为 null) */ + ruleString: z.string().nullable(), + source: RuleGroupSource, + enabled: z.boolean(), + revision: z.number().int().min(1), + /** 语义指纹(kind + scope + 层 + 硬排除),停用/删除后写抑制记录 */ + fingerprint: z.string(), + evidence: JsonObject, + /** 非 null = 属于该类默认链,按此排序 */ + defaultOrder: z.number().int().nullable(), + /** 引用本组的订阅缺省是否自动补全其他版本;null = 不表态 */ + autoFill: z.boolean().nullable(), + /** 由偏好标签迁来/经旧端点写入时的旧 kind */ + legacyKind: PreferenceKind.nullable(), + suppressedAt: Timestamp.nullable(), + deletedAt: Timestamp.nullable(), +}); +export type RuleGroup = z.infer; + +/** 层与硬排除可直接给规则串:ruleString 覆盖 layers;hardExcludes 的元素可以是表达式或单层规则串 */ +export const RuleGroupCreate = z.object({ + name: z.string().min(1).max(80), + description: z.string().max(500).nullable().optional(), + kind: RuleGroupKind.optional(), + scope: RuleGroupScope.optional(), + layers: z.array(RuleLayer).max(50).optional(), + ruleString: z.string().max(2000).optional(), + hardExcludes: z + .array(z.union([z.string().min(1).max(500), RuleExpr])) + .max(50) + .optional(), + ladderRequired: z.boolean().optional(), + source: RuleGroupSource.optional(), + enabled: z.boolean().optional(), + evidence: JsonObject.optional(), + autoFill: z.boolean().nullable().optional(), +}); +export type RuleGroupCreate = z.infer; +export const RuleGroupUpdate = RuleGroupCreate.omit({ source: true }) + .partial() + .extend({ expectedRevision: z.number().int().min(1).optional() }); +export type RuleGroupUpdate = z.infer; + +/** 规则组修订历史(审计事件回放,新→旧) */ +export const RuleGroupRevisionList = z.object({ + items: z.array( + z.object({ + id: Id, + at: Timestamp, + action: z.string(), + actorType: z.string(), + actorId: z.string().nullable(), + revision: z.number().int().nullable(), + changedFields: z.array(z.string()), + snapshot: JsonObject, + }), + ), +}); +export type RuleGroupRevisionList = z.infer; + +export const RuleParseError = z.object({ + code: z.enum([ + "empty", + "empty_layer", + "unknown_atom", + "unsupported_atom", + "unbalanced_paren", + "unexpected_token", + "invalid_char", + ]), + message: z.string(), + /** 在整个规则串里的字符偏移(0 起) */ + position: z.number().int(), + /** 第几层(0 起) */ + layer: z.number().int(), + atom: z.string().optional(), + suggestions: z.array(z.string()).optional(), +}); +export type RuleParseError = z.infer; +export const RuleParseRequest = z.object({ ruleString: z.string().max(2000) }); +export const RuleParseResult = z.object({ + layers: z.array(RuleLayer), + /** 规范化后的规则串(有错误时为 null) */ + ruleString: z.string().nullable(), + errors: z.array(RuleParseError), +}); +export type RuleParseResult = z.infer; +export const RuleSerializeRequest = z.object({ layers: z.array(RuleLayer).max(50) }); +export const RuleSerializeResult = z.object({ ruleString: z.string() }); + +/** 试算:缺省取最近 N 条已识别资源;也可指定资源或粘贴标题;draft 覆盖未保存的修改 */ +export const RulePreviewRequest = z.object({ + limit: z.number().int().min(1).max(200).optional(), + sourceReleaseIds: z.array(Id).max(200).optional(), + titles: z + .array(z.object({ title: z.string().min(1).max(500), subtitle: z.string().max(500).optional() })) + .max(50) + .optional(), + draft: RuleGroupCreate.pick({ layers: true, ruleString: true, hardExcludes: true, ladderRequired: true }) + .partial() + .optional(), +}); +export type RulePreviewRequest = z.infer; +export const RulePreviewItem = z.object({ + sourceReleaseId: Id.nullable(), + title: z.string(), + subtitle: z.string().nullable(), + passed: z.boolean(), + /** 命中层的 pri;被拒绝为 null;无层为 100 */ + pri: z.number().int().nullable(), + /** 命中的层(0 起) */ + level: z.number().int().nullable(), + excludedBy: z.array(z.string()), + /** 拒绝原因摘要:hard_exclude / no_layer_matched */ + reasons: z.array(z.string()), + trace: z.array(z.string()), +}); +export const RulePreviewResult = z.object({ + ruleGroupId: Id, + revision: z.number().int(), + evaluatedAt: Timestamp, + sampleSize: z.number().int(), + passed: z.number().int(), + rejected: z.number().int(), + /** 各 pri 的条数(pri=null 为被拒绝) */ + priHistogram: z.array(z.object({ pri: z.number().int().nullable(), count: z.number().int() })), + items: z.array(RulePreviewItem), +}); +export type RulePreviewResult = z.infer; + +/** 关键字表:内置(对齐 MP)+ 自定义条件 */ +export const RuleKeyword = z.object({ + id: z.string(), + name: z.string(), + description: z.string(), + builtin: z.boolean(), + /** false = 音乐等未迁移条件,规则串里出现会报错 */ + supported: z.boolean(), + /** 我们的判定依据:field 解析字段 / regex MP 正则 / site 站点事实 / custom 自定义条件 */ + basis: z.enum(["field", "regex", "site", "custom"]), + /** MP 上游定义(include/exclude/match/tmdb/downloadvolumefactor) */ + mp: JsonObject.nullable(), +}); +export type RuleKeyword = z.infer; +export const RuleKeywordList = z.object({ items: z.array(RuleKeyword) }); + +/** 自定义条件(对齐 MP 自定义规则;不允许覆盖内置关键字) */ +export const RuleAtom = EntityBase.extend({ + /** 大写存储;写入时不区分大小写 */ + key: z + .string() + .max(32) + .regex(/^(?:[A-Za-z][A-Za-z0-9]*|[0-9]+[A-Za-z][A-Za-z0-9]*)$/), + name: z.string().nullable(), + description: z.string().nullable(), + include: z.array(z.string()), + exclude: z.array(z.string()), + matchFields: z.array(z.enum(["title", "subtitle", "labels"])), + /** MB,剧集按每集 */ + unitSizeMb: z.object({ min: z.number().min(0).optional(), max: z.number().min(0).optional() }).nullable(), + minSeeders: z.number().int().min(0).nullable(), + publishDelayMinutes: z.object({ min: z.number().min(0).optional(), max: z.number().min(0).optional() }).nullable(), + free: z.boolean(), +}); +export type RuleAtom = z.infer; +export const RuleAtomCreate = RuleAtom.pick({ + key: true, + name: true, + description: true, + include: true, + exclude: true, + matchFields: true, + unitSizeMb: true, + minSeeders: true, + publishDelayMinutes: true, + free: true, +}).partial({ + name: true, + description: true, + include: true, + exclude: true, + matchFields: true, + unitSizeMb: true, + minSeeders: true, + publishDelayMinutes: true, + free: true, +}); +export type RuleAtomCreate = z.infer; +export const RuleAtomUpdate = RuleAtomCreate.omit({ key: true }).partial(); +export type RuleAtomUpdate = z.infer; + +/** 三套默认链:订阅 / 洗版(开升级时叠加)/ 搜索;按数组顺序串联 */ +export const RuleDefaults = z.object({ + acquisition: z.array(Id), + upgrade: z.array(Id), + search: z.array(Id), +}); +export type RuleDefaults = z.infer; diff --git a/packages/contracts/test/db-sync.test.ts b/packages/contracts/test/db-sync.test.ts index f2d3889d..b1d08e49 100644 --- a/packages/contracts/test/db-sync.test.ts +++ b/packages/contracts/test/db-sync.test.ts @@ -112,6 +112,9 @@ const ROWS: Record ScheduledJobRun: { row: C.ScheduledJobRun }, // W7-A Pi 上游(0027):secret_ref 不出 API PiUpstream: { row: C.PiUpstreamRow, dbOnly: ["secretRef"] }, + // W9-D 规则组(0033):legacy_rule 只供旧 /preference-tags 别名回放 + RuleGroup: { row: C.RuleGroup, dbOnly: ["legacyRule"] }, + RuleAtom: { row: C.RuleAtom }, }; /** 引擎自有表(@umt/tasks 内部使用,不经 API 暴露),不要求 Row schema */ @@ -199,6 +202,8 @@ const ENUMS: Record = { PolicyRevisionStatus: C.PolicyStatus, PreferenceTagKind: C.PreferenceKind, PreferenceTagOrigin: C.PreferenceOrigin, + RuleGroupKind: C.RuleGroupKind, + RuleGroupSource: C.RuleGroupSource, PreferenceTagStatus: C.PreferenceStatus, ReconcileRunKind: C.ReconcileKind, ReconcileRunMode: C.ReconcileMode, diff --git a/packages/db/migrations/0033_rule_groups.ts b/packages/db/migrations/0033_rule_groups.ts new file mode 100644 index 00000000..ee2e5fc7 --- /dev/null +++ b/packages/db/migrations/0033_rule_groups.ts @@ -0,0 +1,90 @@ +import type { Kysely } from "kysely"; +import { dropTables, entityTable, execDdl, oneOf } from "../src/migration-helpers.ts"; + +// W9-D 规则组:收录策略的过滤条件 + 偏好标签合一(docs/design/subscriptions.md §6)。 +// rule_group = 结构化层(每层一个 AND/OR/NOT 表达式,第 k 层 pri = 100 - k)+ 硬排除;规则串只是它的文本表示。 +// rule_atom = 自定义条件(对齐 MP 自定义规则 7 字段,不允许覆盖内置关键字)。 +// 偏好标签原样迁成规则组(同 id、同指纹,legacy_kind/legacy_rule 保留旧语义),旧端点改为别名;抑制记录挂到规则组。 +// 版本目标 rule_group_ids 有序引用规则组;default_order 非空 = 该类(订阅/洗版/搜索)默认链的一员。 + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl( + db, + ` +${entityTable( + "rule_group", + ` + name text not null check (length(name) between 1 and 80), + description text, + kind text not null default 'acquisition' ${oneOf("kind", ["acquisition", "upgrade", "search"])}, + scope jsonb not null default '{}'::jsonb, + layers jsonb not null default '[]'::jsonb, + hard_excludes jsonb not null default '[]'::jsonb, + ladder_required boolean not null default true, + rule_string text, + source text not null default 'manual' ${oneOf("source", ["manual", "chat", "agent"])}, + enabled boolean not null default true, + revision integer not null default 1 check (revision >= 1), + fingerprint text not null, + evidence jsonb not null default '{}'::jsonb, + default_order integer, + auto_fill boolean, + legacy_kind text check (legacy_kind is null or legacy_kind in ('soft', 'hard', 'version', 'target')), + legacy_rule jsonb, + suppressed_at timestamptz, + deleted_at timestamptz`, +)} +create index rule_group_live_idx on rule_group (kind, default_order) where deleted_at is null; +create index rule_group_fingerprint_idx on rule_group (fingerprint) where deleted_at is null; + +${entityTable( + "rule_atom", + ` + key text not null unique check (key ~ '^([A-Z][A-Z0-9]*|[0-9]+[A-Z][A-Z0-9]*)$'), + name text, + description text, + include jsonb not null default '[]'::jsonb, + exclude jsonb not null default '[]'::jsonb, + match_fields jsonb not null default '[]'::jsonb, + unit_size_mb jsonb, + min_seeders integer check (min_seeders is null or min_seeders >= 0), + publish_delay_minutes jsonb, + free boolean not null default false`, +)} + +alter table acquisition_slot add column rule_group_ids jsonb not null default '[]'::jsonb check (jsonb_typeof(rule_group_ids) = 'array'); +alter table preference_suppression add column rule_group_id uuid references rule_group (id) on delete set null; +create index preference_suppression_rule_group_idx on preference_suppression (rule_group_id) where rule_group_id is not null; + +insert into rule_group (id, name, description, kind, scope, layers, hard_excludes, ladder_required, source, enabled, + revision, fingerprint, evidence, legacy_kind, legacy_rule, suppressed_at, deleted_at, + created_at, updated_at) +select id, label, description, 'acquisition', scope, '[]'::jsonb, '[]'::jsonb, kind <> 'soft', + case origin when 'manual' then 'manual' when 'conversation' then 'chat' else 'agent' end, + status = 'active', revision, fingerprint, evidence, kind, rule, + case when status <> 'active' then updated_at end, + case when status = 'deleted' then updated_at end, + created_at, updated_at + from preference_tag; +update preference_suppression set rule_group_id = preference_tag_id where preference_tag_id is not null; + +insert into rule_group (name, description, kind, source, fingerprint, default_order) values + ('订阅默认', '订阅没有指定规则组时使用', 'acquisition', 'manual', 'seed:acquisition', 0), + ('洗版默认', '版本目标开启升级时额外叠加', 'upgrade', 'manual', 'seed:upgrade', 0), + ('搜索默认', '手动搜索与 Agent 搜索资源时使用', 'search', 'manual', 'seed:search', 0); +`, + ); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl( + db, + ` +alter table preference_suppression drop column if exists rule_group_id; +alter table acquisition_slot drop column if exists rule_group_ids; +${dropTables(["rule_atom", "rule_group"])} +`, + ); +} diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index 418ff71c..d2f35280 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -206,6 +206,8 @@ export type ResolutionAttemptOutcome = | "manual" | "skipped_locked"; export type ResolutionAttemptActorType = "user" | "agent" | "system" | "api_key"; +export type RuleGroupKind = "acquisition" | "upgrade" | "search"; +export type RuleGroupSource = "manual" | "chat" | "agent"; export type ScheduledJobBoardGroup = "collect" | "download" | "organize" | "maintain" | "agent"; export type ScheduledJobKind = "task" | "agent_check"; export type ScheduledJobLastStatus = "ok" | "failed" | "skipped" | "running"; @@ -289,6 +291,7 @@ export interface AcquisitionSlotTable { created_at: GeneratedTimestamptz; updated_at: GeneratedTimestamptz; reserved_plan_id: string | null; + rule_group_ids: Generated; } export interface AdminSessionTable { @@ -1056,6 +1059,7 @@ export interface PreferenceSuppressionTable { lifted_by: string | null; created_at: GeneratedTimestamptz; updated_at: GeneratedTimestamptz; + rule_group_id: string | null; } export interface PreferenceTagTable { @@ -1108,6 +1112,47 @@ export interface ResolutionAttemptTable { updated_at: GeneratedTimestamptz; } +export interface RuleAtomTable { + id: Generated; + key: string; + name: string | null; + description: string | null; + include: Generated; + exclude: Generated; + match_fields: Generated; + unit_size_mb: Json | null; + min_seeders: number | null; + publish_delay_minutes: Json | null; + free: Generated; + created_at: GeneratedTimestamptz; + updated_at: GeneratedTimestamptz; +} + +export interface RuleGroupTable { + id: Generated; + name: string; + description: string | null; + kind: Generated; + scope: Generated; + layers: Generated; + hard_excludes: Generated; + ladder_required: Generated; + rule_string: string | null; + source: Generated; + enabled: Generated; + revision: Generated; + fingerprint: string; + evidence: Generated; + default_order: number | null; + auto_fill: boolean | null; + legacy_kind: string | null; + legacy_rule: Json | null; + suppressed_at: Timestamptz | null; + deleted_at: Timestamptz | null; + created_at: GeneratedTimestamptz; + updated_at: GeneratedTimestamptz; +} + export interface ScheduledJobTable { id: string; name: string; @@ -1515,6 +1560,8 @@ export interface DB { preference_tag: PreferenceTagTable; reconcile_run: ReconcileRunTable; resolution_attempt: ResolutionAttemptTable; + rule_atom: RuleAtomTable; + rule_group: RuleGroupTable; scheduled_job: ScheduledJobTable; scheduled_job_run: ScheduledJobRunTable; secret: SecretTable; @@ -1682,6 +1729,12 @@ export type ReconcileRunUpdate = Updateable; export type ResolutionAttemptRow = Selectable; export type NewResolutionAttempt = Insertable; export type ResolutionAttemptUpdate = Updateable; +export type RuleAtomRow = Selectable; +export type NewRuleAtom = Insertable; +export type RuleAtomUpdate = Updateable; +export type RuleGroupRow = Selectable; +export type NewRuleGroup = Insertable; +export type RuleGroupUpdate = Updateable; export type ScheduledJobRow = Selectable; export type NewScheduledJob = Insertable; export type ScheduledJobUpdate = Updateable; diff --git a/packages/db/test/schema.test.ts b/packages/db/test/schema.test.ts index 7091c810..5e31c387 100644 --- a/packages/db/test/schema.test.ts +++ b/packages/db/test/schema.test.ts @@ -99,6 +99,9 @@ const EXPECTED_TABLES = [ // W7-C 定时任务注册表(0028) "scheduled_job", "scheduled_job_run", + // W9-D 规则组(0033) + "rule_group", + "rule_atom", ].sort(); /** 不走 entityTable() 的表:追加型/引擎自有表,没有 uuid id + created_at/updated_at 三件套 */