From d3ab54f2bd903e12940396afdee0151e620e51c2 Mon Sep 17 00:00:00 2001 From: root Date: Sun, 27 Sep 2026 21:10:02 +0800 Subject: [PATCH] feat(server): W4-A integration endpoints and contract gaps - POST /acquisition-plans: manual submit via download.plan -> download.submit (same authorization/audit/idempotency path as the agent), slotId or entityId - read-only /media-assets, /asset-locations, /asset-unit-coverage, /path-bindings, /reconcile-runs, /server-mappings, /artifacts, GET /title-aliases/{id}; unimplemented write endpoints removed from api.ts; registry test now checks both directions - candidate evaluation upsert per (slot, source_release) + migration 0022 dedupe/unique index - organization records expose downloadTaskId (+ list filter); inventory adds subscribed/candidateReasons; feed sources expose derived possibleGap - GET /events?after=&limit= JSON replay page for API key callers (same sse_event table) - preference tag revision history from audit_event; subscriptions/identities/aliases lists carry mediaEntityTitle; subscription q/mediaKind search; slot subscriptionId filter + currentAsset summary; policy preview samples carry title/reasons - DecisionSummary, IdentityRevisionList, ReleaseIdentityResult, PolicyPreviewSummary exported from contracts Co-Authored-By: Claude Opus 5.5 --- apps/server/src/http/sse.ts | 59 +++++- .../src/modules/acquisition-organize/index.ts | 2 + .../acquisition-organize/routes/assets.ts | 100 ++++++++++ .../acquisition-organize/routes/library.ts | 38 ++++ .../routes/organization.ts | 1 + .../acquisition-organize/routes/rules.ts | 186 +++++++++++++++++- .../acquisition-organize/services/common.ts | 36 +--- .../acquisition-organize/services/organize.ts | 1 + apps/server/src/modules/agent/routes.ts | 130 ++++++++++-- .../src/modules/sources-identity/defs.ts | 1 + .../sources-identity/routes/identity.ts | 46 ++++- .../sources-identity/routes/sources.ts | 14 +- .../src/modules/sources-identity/rows.ts | 15 ++ .../modules/sources-identity/service/feeds.ts | 16 +- apps/server/test/api-registry.test.ts | 5 + apps/server/test/e2e/e6-w4a-endpoints.test.ts | 148 ++++++++++++++ .../test/sources-identity/ingest.test.ts | 9 + apps/server/test/sse.test.ts | 26 +++ apps/web/app/api/feeds.ts | 12 +- apps/web/app/api/inventory.ts | 31 ++- apps/web/app/components/library/slot-card.tsx | 14 +- .../workbench/release-detail-sheet.tsx | 5 +- .../app/components/workbench/release-list.tsx | 18 +- apps/web/app/routes/console/sources.tsx | 6 +- packages/contracts/src/api.ts | 106 +++++++--- packages/contracts/src/content.ts | 2 + packages/contracts/src/events.ts | 22 +++ packages/contracts/src/execution.ts | 25 +++ packages/contracts/src/identity.ts | 2 + packages/contracts/src/organization.ts | 21 ++ packages/contracts/src/policy.ts | 63 ++++++ packages/contracts/src/sources.ts | 7 + packages/contracts/test/db-sync.test.ts | 10 +- .../db/migrations/0022_candidate_dedupe.ts | 29 +++ 34 files changed, 1083 insertions(+), 123 deletions(-) create mode 100644 apps/server/src/modules/acquisition-organize/routes/assets.ts create mode 100644 apps/server/test/e2e/e6-w4a-endpoints.test.ts create mode 100644 packages/db/migrations/0022_candidate_dedupe.ts diff --git a/apps/server/src/http/sse.ts b/apps/server/src/http/sse.ts index 696a3be9..b97be08d 100644 --- a/apps/server/src/http/sse.ts +++ b/apps/server/src/http/sse.ts @@ -1,5 +1,6 @@ -import { type EventEnvelope, EventType } from "@umt/contracts"; -import { Hono } from "hono"; +import { createRoute, OpenAPIHono } from "@hono/zod-openapi"; +import { Api, apiResponse, type EventEnvelope, type EventReplayPage, EventType } from "@umt/contracts"; +import type { Context } from "hono"; import { streamSSE } from "hono/streaming"; import { ApiException } from "./errors.ts"; @@ -79,14 +80,60 @@ export interface SseOptions { keepaliveMs?: number; } -/** GET /events?after=(也接受 Last-Event-ID 头) */ +const REPLAY_DEFAULT_LIMIT = 100; + +/** JSON 补读还是 SSE:带 limit,或 Accept 只要 JSON(不含 text/event-stream)时走 JSON */ +function wantsJson(accept: string | undefined, limit: string | undefined): boolean { + if (limit !== undefined) return true; + const a = (accept ?? "").toLowerCase(); + return a.includes("application/json") && !a.includes("text/event-stream"); +} + +/** + * GET /events?after=(也接受 Last-Event-ID 头): + * - 控制台(cookie 会话)用 EventSource 走 SSE 流; + * - API Key 调用者用 `GET /events?after=&limit=` 取 JSON 页(同一 sse_event 表),按 nextAfter 续读。 + */ export function sseRoutes({ bus, keepaliveMs = 15_000 }: SseOptions) { - const app = new Hono(); - app.get("/events", (c) => { + const app = new OpenAPIHono(); + const def = Api.events.list; + const route = createRoute({ + method: "get", + path: def.path, + tags: ["events"], + summary: def.summary, + responses: { + 200: { + description: "JSON 补读页(带 limit 或 Accept: application/json);否则为 text/event-stream", + content: { + "application/json": { schema: apiResponse(def.response) }, + "text/event-stream": { schema: { type: "string" } }, + }, + }, + }, + }); + app.openapi(route, (async (c: Context) => { const raw = c.req.query("after") ?? c.req.header("last-event-id") ?? "0"; const after = Number(raw); if (!Number.isInteger(after) || after < 0) throw new ApiException("bad_request", "after 必须是非负整数"); + const limitRaw = c.req.query("limit"); + if (wantsJson(c.req.header("accept"), limitRaw)) { + const limit = limitRaw === undefined ? REPLAY_DEFAULT_LIMIT : Number(limitRaw); + if (!Number.isInteger(limit) || limit < 1 || limit > 1000) + throw new ApiException("bad_request", "limit 必须是 1–1000 的整数"); + const r = await bus.since(after, limit + 1); + const items = r.events.slice(0, limit); + const data: EventReplayPage = { + items, + nextAfter: items.at(-1)?.seq ?? after, + hasMore: r.events.length > limit, + latestSeq: r.latestSeq, + gap: r.gap, + }; + return c.json({ ok: true as const, data }); + } + return streamSSE(c, async (stream) => { const queue: EventEnvelope[] = []; let wake: (() => void) | null = null; @@ -140,6 +187,6 @@ export function sseRoutes({ bus, keepaliveMs = 15_000 }: SseOptions) { unsubscribe(); } }); - }); + }) as never); return app; } diff --git a/apps/server/src/modules/acquisition-organize/index.ts b/apps/server/src/modules/acquisition-organize/index.ts index 7924c389..edcdb828 100644 --- a/apps/server/src/modules/acquisition-organize/index.ts +++ b/apps/server/src/modules/acquisition-organize/index.ts @@ -3,6 +3,7 @@ import type { ServerContext, ServerModule } from "../types.ts"; import { registerAoCapabilities } from "./capabilities.ts"; import { registerAoTasks } from "./handlers.ts"; import { type AoOptions, type AoPorts, createPorts } from "./ports.ts"; +import { registerAssetRoutes } from "./routes/assets.ts"; import { registerDashboardRoutes } from "./routes/dashboard.ts"; import { registerDownloadRoutes } from "./routes/downloads.ts"; import { registerLibraryRoutes } from "./routes/library.ts"; @@ -67,6 +68,7 @@ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): Se registerOrganizationRoutes(app, d); registerLibraryRoutes(app, d); registerDashboardRoutes(app, d); + registerAssetRoutes(app, d); }, registerTasks(registry, ctx) { registerAoTasks(registry, lazy(ctx)); diff --git a/apps/server/src/modules/acquisition-organize/routes/assets.ts b/apps/server/src/modules/acquisition-organize/routes/assets.ts new file mode 100644 index 00000000..006ca3f3 --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/routes/assets.ts @@ -0,0 +1,100 @@ +import { Api, type EndpointDef } from "@umt/contracts"; +import type { ModuleApp } from "../../types.ts"; +import type { AoDeps } from "../services/common.ts"; +import { notFound, toApi, toApiList } from "../util.ts"; +import { mount, paginate } from "./helpers.ts"; + +// 资产 / 位置 / 覆盖 / 路径绑定 / 校验运行 / 库存映射:只读列表与详情。 +// 这些表只由整理、对账、媒体服务器同步流程写入,不开放直接修改(api.ts 已移除对应写端点)。 + +const TAG = "assets"; + +type Table = + | "media_asset" + | "asset_unit_coverage" + | "asset_location" + | "path_binding" + | "reconcile_run" + | "server_mapping"; + +interface ReadOnly { + list: EndpointDef; + get: EndpointDef; + table: Table; + what: string; + /** query 字段 → 列名 */ + filters: Record; +} + +const READ_ONLY: ReadOnly[] = [ + { + ...Api.mediaAssets, + table: "media_asset", + what: "资产", + filters: { mediaEntityId: "media_entity_id", status: "status" }, + }, + { + ...Api.assetUnitCoverage, + table: "asset_unit_coverage", + what: "资产覆盖", + filters: { mediaAssetId: "media_asset_id", contentUnitId: "content_unit_id" }, + }, + { + ...Api.assetLocations, + table: "asset_location", + what: "资产位置", + filters: { mediaAssetId: "media_asset_id", status: "status" }, + }, + { + ...Api.pathBindings, + table: "path_binding", + what: "路径绑定", + filters: { mediaEntityId: "media_entity_id", status: "status" }, + }, + { + ...Api.reconcileRuns, + table: "reconcile_run", + what: "校验运行", + filters: { kind: "kind", status: "status" }, + }, + { + ...Api.serverMappings, + table: "server_mapping", + what: "库存映射", + filters: { + mediaServerInstanceId: "media_server_instance_id", + mediaEntityId: "media_entity_id", + matchStatus: "match_status", + }, + }, +]; + +export function registerAssetRoutes(app: ModuleApp, d: AoDeps) { + const db = () => d.ctx.db; + for (const r of READ_ONLY) { + mount( + app, + r.list, + async (_c, { query }) => { + // biome-ignore lint/suspicious/noExplicitAny: 表名在 READ_ONLY 里枚举,列名随表变化 + let q = (db() as any).selectFrom(r.table).selectAll(); + for (const [key, col] of Object.entries(r.filters)) { + const v = (query as Record)[key]; + if (v !== undefined) q = q.where(col, "=", v); + } + const p = await paginate(q, query); + return { items: toApiList(p.rows as object[]), nextCursor: p.nextCursor }; + }, + { tag: TAG }, + ); + mount( + app, + r.get, + async (_c, { params }) => { + const row = await db().selectFrom(r.table).selectAll().where("id", "=", params.id).executeTakeFirst(); + return toApi((row as object | undefined) ?? notFound(r.what)); + }, + { tag: TAG }, + ); + } +} diff --git a/apps/server/src/modules/acquisition-organize/routes/library.ts b/apps/server/src/modules/acquisition-organize/routes/library.ts index 23623905..93da034c 100644 --- a/apps/server/src/modules/acquisition-organize/routes/library.ts +++ b/apps/server/src/modules/acquisition-organize/routes/library.ts @@ -317,6 +317,42 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) { .where("dt.status", "in", [...ACTIVE_DL]) .execute(), ]); + const [subs, cands] = await Promise.all([ + db() + .selectFrom("subscription") + .select("media_entity_id") + .where("media_entity_id", "in", ids) + .where("status", "=", "active") + .execute(), + db() + .selectFrom("artifact as a") + .innerJoin("acquisition_slot as s", (j) => + j.on(sql`a.subject_id = s.id and a.subject_type = 'acquisition_slot'`), + ) + .select(["s.media_entity_id", "s.name as slot_name", "a.content"]) + .where("a.title", "=", CANDIDATE_ARTIFACT_TITLE) + .where("s.media_entity_id", "in", ids) + .orderBy("a.updated_at", "desc") + .limit(50 * ids.length) + .execute(), + ]); + const subscribed = new Set(subs.map((r) => r.media_entity_id)); + const reasonsOf = (entityId: string) => + cands + .filter((c) => c.media_entity_id === entityId) + .slice(0, 50) + .map((c) => { + const ev = c.content as CandidateEvaluation; + return { + slotId: ev.slotId, + slotName: c.slot_name, + sourceReleaseId: ev.sourceReleaseId, + decision: ev.decision, + reasons: [...new Set([...(ev.rejected ?? []).map((r) => r.message), ...(ev.reasons ?? [])])], + planId: ev.planId ?? null, + evaluatedAt: ev.evaluatedAt, + }; + }); return { items: ids.map((entityId) => ({ entityId, @@ -340,6 +376,8 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) { downloading: dts .filter((t) => t.entity_id === entityId) .map((t) => ({ downloadTaskId: t.id, name: t.name, status: t.status, progress: t.progress })), + subscribed: subscribed.has(entityId), + candidateReasons: reasonsOf(entityId), })), }; }, diff --git a/apps/server/src/modules/acquisition-organize/routes/organization.ts b/apps/server/src/modules/acquisition-organize/routes/organization.ts index 48bc1b44..0a2e7dae 100644 --- a/apps/server/src/modules/acquisition-organize/routes/organization.ts +++ b/apps/server/src/modules/acquisition-organize/routes/organization.ts @@ -23,6 +23,7 @@ export function registerOrganizationRoutes(app: ModuleApp, d: AoDeps) { if (query.status) q = q.where("status", "=", query.status); if (query.trigger) q = q.where("trigger", "=", query.trigger); if (query.mediaEntityId) q = q.where("media_entity_id", "=", query.mediaEntityId); + if (query.downloadTaskId) q = q.where("download_task_id", "=", query.downloadTaskId); if (query.q) { const like = `%${String(query.q).replace(/[\\%_]/g, (m) => `\\${m}`)}%`; q = q.where((eb) => diff --git a/apps/server/src/modules/acquisition-organize/routes/rules.ts b/apps/server/src/modules/acquisition-organize/routes/rules.ts index 77131441..4a82589f 100644 --- a/apps/server/src/modules/acquisition-organize/routes/rules.ts +++ b/apps/server/src/modules/acquisition-organize/routes/rules.ts @@ -10,6 +10,8 @@ import { import { type DB, sql, type Transaction } from "@umt/db"; import { evaluateCandidate, parseRelease } 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"; @@ -34,6 +36,48 @@ async function kickEvaluate(d: AoDeps, input: { subscriptionId?: string; acquisi } } +/** 偏好标签写入审计(修订历史端点据此回放):payload 带写入后快照与改动字段 */ +async function auditPref( + trx: Transaction, + c: Context, + action: "created" | "updated" | "deleted", + row: { + id: string; + label: string; + kind: string; + status: string; + scope: unknown; + rule: unknown; + description: string | null; + revision: number; + }, + 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: `preference_tag.${action}`, + target_type: "preference_tag", + target_id: row.id, + payload: jsonb({ + revision: row.revision, + changedFields, + snapshot: { + label: row.label, + kind: row.kind, + status: row.status, + scope: row.scope, + rule: row.rule, + description: row.description, + }, + }), + }) + .execute(); +} + export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { const db = () => d.ctx.db; @@ -42,10 +86,32 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { app, Api.subscriptions.list, async (_c, { query }) => { - let q = db().selectFrom("subscription").selectAll(); - if (query.status) q = q.where("status", "=", query.status); - if (query.mediaEntityId) q = q.where("media_entity_id", "=", query.mediaEntityId); - const p = await paginate(q, query); + let q = db() + .selectFrom("subscription as s") + .leftJoin("media_entity as e", "e.id", "s.media_entity_id") + .selectAll("s") + .select("e.title as media_entity_title"); + if (query.status) q = q.where("s.status", "=", query.status); + if (query.mediaEntityId) q = q.where("s.media_entity_id", "=", query.mediaEntityId); + if (query.mediaKind) q = q.where("s.media_kind", "=", query.mediaKind); + if (query.q) { + const like = `%${String(query.q).replace(/[\\%_]/g, (m) => `\\${m}`)}%`; + q = q.where((eb) => + eb.or([ + eb("e.title", "ilike", like), + eb("e.original_title", "ilike", like), + eb("s.title_query", "ilike", like), + eb.exists( + eb + .selectFrom("title_alias as t") + .select("t.id") + .whereRef("t.media_entity_id", "=", "s.media_entity_id") + .where("t.alias", "ilike", like), + ), + ]), + ); + } + const p = await paginate(q, query, "s."); return { items: toApiList(p.rows), nextCursor: p.nextCursor }; }, { tag: TAG }, @@ -152,8 +218,34 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { let q = db().selectFrom("acquisition_slot").selectAll(); if (query.mediaEntityId) q = q.where("media_entity_id", "=", query.mediaEntityId); if (query.status) q = q.where("status", "=", query.status); + if (query.subscriptionId) q = q.where("subscription_id", "=", query.subscriptionId); const p = await paginate(q, query); - return { items: toApiList(p.rows), nextCursor: p.nextCursor }; + const assetIds = p.rows.map((r) => r.current_asset_id).filter((x): x is string => !!x); + const assets = assetIds.length + ? await db() + .selectFrom("media_asset") + .select(["id", "delivery", "size_bytes", "status"]) + .where("id", "in", assetIds) + .execute() + : []; + const items = p.rows.map((r) => { + const a = assets.find((x) => x.id === r.current_asset_id); + const dv = (a?.delivery ?? {}) as { resolution?: string; sourceType?: string; releaseGroup?: string }; + return { + ...toApi(r), + currentAsset: a + ? { + id: a.id, + resolution: dv.resolution ?? null, + sourceType: dv.sourceType ?? null, + releaseGroup: dv.releaseGroup ?? null, + sizeBytes: a.size_bytes, + status: a.status, + } + : null, + }; + }); + return { items, nextCursor: p.nextCursor }; }, { tag: TAG }, ); @@ -484,6 +576,7 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { .returningAll() .executeTakeFirst(); if (!ins) conflict("已存在同一语义的偏好标签"); + await auditPref(trx, c, "created", ins!, []); return ins!; }); return toApi(r); @@ -519,7 +612,10 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { } if (fp !== cur!.fingerprint && cur!.status === "active") await suppress(trx, cur!.id, cur!.fingerprint, actorOf(c), "edited"); - return trx + 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") .set({ ...(body.label !== undefined ? { label: body.label } : {}), @@ -535,6 +631,8 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { .where("id", "=", params.id) .returningAll() .executeTakeFirstOrThrow(); + await auditPref(trx, c, "updated", next_, changed); + return next_; }); return toApi(r); }, @@ -559,6 +657,7 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { .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"); }); return { ok: true as const }; @@ -566,6 +665,53 @@ export function registerRuleRoutes(app: ModuleApp, d: AoDeps) { { tag: TAG, status: 200 }, ); + mount( + app, + Api.preferenceTags.revisions, + async (_c, { params }) => { + const tag = await db().selectFrom("preference_tag").select("id").where("id", "=", params.id).executeTakeFirst(); + if (!tag) notFound("偏好标签"); + const rows = await db() + .selectFrom("audit_event") + .selectAll() + .where("target_type", "=", "preference_tag") + .where("target_id", "=", params.id) + .where("action", "like", "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; + }; + const snap = p.snapshot ?? {}; + return { + id: r.id, + at: r.at.toISOString(), + action: r.action.slice("preference_tag.".length), + actorType: r.actor_type, + actorId: r.actor_id, + revision: p.revision ?? null, + changedFields: p.changedFields ?? [], + snapshot: { + label: String(snap.label ?? ""), + kind: String(snap.kind ?? ""), + status: String(snap.status ?? ""), + scope: (snap.scope ?? {}) as Record, + rule: (snap.rule ?? {}) as Record, + description: (snap.description as string | null | undefined) ?? null, + }, + }; + }), + }; + }, + { tag: TAG }, + ); + // ── 抑制记录 ── mount( app, @@ -696,7 +842,14 @@ export async function previewPolicy(d: AoDeps, rules: unknown) { .execute(); let matched = 0; let excluded = 0; - const samples: { sourceReleaseId: string; template: string; decision: string }[] = []; + const samples: { + sourceReleaseId: string; + title: string; + sizeBytes: number | null; + template: string; + decision: string; + reasons: string[]; + }[] = []; for (const rel of rels) { const parsed = { ...parseRelease(rel.title, rel.subtitle ? { subtitle: rel.subtitle } : {}), @@ -715,7 +868,24 @@ export async function previewPolicy(d: AoDeps, rules: unknown) { const ev = evaluateCandidate(spec, parsed, null); if (ev.decision === "accept") matched++; else excluded++; - if (samples.length < 20) samples.push({ sourceReleaseId: rel.id, template: t.name, decision: ev.decision }); + if (samples.length < 20) { + const reasons = + ev.decision === "reject" + ? ev.violations.map((v) => `${v.field}:${v.code}`) + : ev.decision === "unknown_field" + ? ev.missing.map((m) => `missing:${m}`) + : "reasons" in ev + ? (ev.reasons ?? []) + : []; + samples.push({ + sourceReleaseId: rel.id, + title: rel.title, + sizeBytes: rel.size_bytes, + template: t.name, + decision: ev.decision, + reasons, + }); + } } } return { diff --git a/apps/server/src/modules/acquisition-organize/services/common.ts b/apps/server/src/modules/acquisition-organize/services/common.ts index 3fd81bd7..3ea3c3fa 100644 --- a/apps/server/src/modules/acquisition-organize/services/common.ts +++ b/apps/server/src/modules/acquisition-organize/services/common.ts @@ -56,6 +56,10 @@ export async function upsertDecision(d: AoDeps, exec: Exec, input: DecisionInput export const CANDIDATE_ARTIFACT_TITLE = "candidate_evaluation"; +/** + * 同一 (Slot, 来源资源) 只保留最新一条评估(唯一索引 artifact_candidate_slot_release); + * 重复评估覆盖内容与 updated_at,不再追加新行。 + */ export async function recordCandidate(exec: Exec, ev: CandidateEvaluation): Promise { const content = { ...ev }; const contentHash = hashJson({ @@ -66,32 +70,12 @@ export async function recordCandidate(exec: Exec, ev: CandidateEvaluation): Prom reasons: ev.reasons, planId: ev.planId, }); - const dup = await exec - .selectFrom("artifact") - .select("id") - .where("subject_type", "=", "acquisition_slot") - .where("subject_id", "=", ev.slotId) - .where("content_hash", "=", contentHash) - .executeTakeFirst(); - if (dup) { - await exec - .updateTable("artifact") - .set({ content: jsonb(content), updated_at: new Date() }) - .where("id", "=", dup.id) - .execute(); - return; - } - await exec - .insertInto("artifact") - .values({ - kind: "evidence", - subject_type: "acquisition_slot", - subject_id: ev.slotId, - title: CANDIDATE_ARTIFACT_TITLE, - content: jsonb(content), - content_hash: contentHash, - }) - .execute(); + await sql` + insert into artifact (kind, subject_type, subject_id, title, content, content_hash) + values ('evidence', 'acquisition_slot', ${ev.slotId}, ${CANDIDATE_ARTIFACT_TITLE}, ${jsonb(content)}::jsonb, ${contentHash}) + on conflict (subject_id, (content->>'sourceReleaseId')) + where title = 'candidate_evaluation' and subject_type = 'acquisition_slot' + do update set content = excluded.content, content_hash = excluded.content_hash, updated_at = now()`.execute(exec); } // ── 业务任务行(task 表:时间线投影;引擎只管推进) ── diff --git a/apps/server/src/modules/acquisition-organize/services/organize.ts b/apps/server/src/modules/acquisition-organize/services/organize.ts index cbe1ffa3..fbbddfd2 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize.ts @@ -1222,6 +1222,7 @@ export async function loadRecordsPublic(exec: Exec, rows: OrganizationRecordRow[ return { id: r.id, taskId: r.task_id, + downloadTaskId: r.download_task_id, operationIds: ops.filter((o) => o.organization_record_id === r.id).map((o) => o.id), media: e ? { diff --git a/apps/server/src/modules/agent/routes.ts b/apps/server/src/modules/agent/routes.ts index 7b17ec67..b095850c 100644 --- a/apps/server/src/modules/agent/routes.ts +++ b/apps/server/src/modules/agent/routes.ts @@ -4,6 +4,7 @@ import { Api, CAPABILITY_CATALOG, type CapabilityDescriptor, + type DownloadPlanOutput, type KnownCapabilityId, } from "@umt/contracts"; import { type AgentRunRow, sql } from "@umt/db"; @@ -341,23 +342,106 @@ export function registerAgentRoutes(app: ModuleApp, ctx: ServerContext) { planId: typeof o.planId === "string" ? o.planId : typeof o.plan?.id === "string" ? o.plan.id : null, }; } catch (err) { - if (!isCapabilityError(err)) throw err; - const code = - err.code === "unauthorized" - ? "forbidden" - : err.code === "not_found" - ? "not_found" - : err.code === "conflict" - ? "conflict" - : err.code === "precondition" || err.code === "budget" - ? "precondition_failed" - : err.code === "invalid_input" - ? "validation_failed" - : "internal"; - throw new ApiException(code, err.message); + capabilityHttpError(err); } }); + // ── 手动提交下载:同一条 download.plan → download.submit 能力路径(授权/审计/幂等与 Agent 一致)── + route(app, Api.acquisitionPlans.create, async ({ body, actorId, c }) => { + const rel = await db() + .selectFrom("source_release") + .select(["id", "media_entity_id"]) + .where("id", "=", body.sourceReleaseId) + .executeTakeFirst(); + if (!rel) throw new ApiException("not_found", "来源资源不存在"); + let slotIds: string[]; + if (body.slotId) { + const s = await db().selectFrom("acquisition_slot").select("id").where("id", "=", body.slotId).executeTakeFirst(); + if (!s) throw new ApiException("not_found", "收录目标(Slot)不存在"); + slotIds = [s.id]; + } else { + const entityId = body.entityId ?? rel.media_entity_id; + if (!entityId) + throw new ApiException("precondition_failed", "来源资源尚未识别到本地作品,请先在“身份与元数据”指定作品"); + slotIds = ( + await db() + .selectFrom("acquisition_slot") + .select("id") + .where("media_entity_id", "=", entityId) + .orderBy("created_at") + .execute() + ).map((r) => r.id); + } + const rejected: { slotId: string | null; code: string; message: string }[] = []; + if (slotIds.length === 0) + rejected.push({ slotId: null, code: "no_slot", message: "该作品没有版本目标(Slot),请先在订阅里建目标" }); + const invokeCtx = { + actorId, + taskId: `http:${c.get("requestId") ?? "req"}`, + scopeIds: [], + signal: c.req.raw.signal, + source: "web" as const, + }; + try { + for (const slotId of slotIds) { + const out = (await ctx.capabilities.invoke( + "download.plan", + { slotId, sourceReleaseId: rel.id }, + invokeCtx, + )) as DownloadPlanOutput; + if (!out.plan) { + for (const r of out.rejected) rejected.push({ slotId, ...r }); + continue; + } + const sub = (await ctx.capabilities.invoke("download.submit", { planId: out.plan.id }, invokeCtx)) as { + taskId: string; + }; + const plan = await db() + .selectFrom("acquisition_plan") + .selectAll() + .where("id", "=", out.plan.id) + .executeTakeFirstOrThrow(); + await audit(ctx, actorId, "acquisition_plan.manual_submit", plan.id, { + sourceReleaseId: rel.id, + slotId, + taskId: sub.taskId, + note: body.note ?? null, + }); + return { submitted: true, plan: camel(plan), slotId, taskId: sub.taskId, rejected }; + } + } catch (err) { + capabilityHttpError(err); + } + return { submitted: false, plan: null, slotId: null, taskId: null, rejected }; + }); + + // ── 产物(整理计划 / 汇报 / 证据):只读 ── + route(app, Api.artifacts.list, async ({ query: raw }) => { + const query = raw as ListQuery & { agentRunId?: string; taskId?: string }; + let q = db().selectFrom("artifact").selectAll(); + if (query.agentRunId) q = q.where("agent_run_id", "=", query.agentRunId); + if (query.taskId) q = q.where("task_id", "=", query.taskId); + if (query.kind) q = q.where("kind", "=", query.kind as never); + const cur = decode(query.cursor); + if (cur) q = q.where(sql`(created_at, id) < (${cur.at}::timestamptz, ${cur.id}::uuid)`); + const rows = await q + .orderBy("created_at", "desc") + .orderBy("id", "desc") + .limit(query.limit + 1) + .execute(); + const items = rows.slice(0, query.limit); + const last = items.at(-1); + return { + items: items.map(camel), + nextCursor: rows.length > query.limit && last ? encode(last.created_at, last.id) : null, + }; + }); + route(app, Api.artifacts.get, async ({ params }) => { + const r = await db().selectFrom("artifact").selectAll().where("id", "=", params.id).executeTakeFirst(); + if (!r) throw new ApiException("not_found", "产物不存在"); + return camel(r); + }); + // ── 业务任务 ── route(app, Api.tasks.list, async ({ query: raw }) => { const query = raw as ListQuery; @@ -413,6 +497,24 @@ export function registerAgentRoutes(app: ModuleApp, ctx: ServerContext) { }); } +/** 能力错误 → HTTP 错误码 */ +function capabilityHttpError(err: unknown): never { + if (!isCapabilityError(err)) throw err; + const code = + err.code === "unauthorized" + ? "forbidden" + : err.code === "not_found" + ? "not_found" + : err.code === "conflict" + ? "conflict" + : err.code === "precondition" || err.code === "budget" + ? "precondition_failed" + : err.code === "invalid_input" + ? "validation_failed" + : "internal"; + throw new ApiException(code, err.message); +} + async function audit(ctx: ServerContext, actorId: string, action: string, targetId: string, payload: object) { await ctx.db .insertInto("audit_event") diff --git a/apps/server/src/modules/sources-identity/defs.ts b/apps/server/src/modules/sources-identity/defs.ts index 1b99ac6d..ba0e407b 100644 --- a/apps/server/src/modules/sources-identity/defs.ts +++ b/apps/server/src/modules/sources-identity/defs.ts @@ -243,6 +243,7 @@ export const D = { { mediaEntityId: Id.optional(), q: z.string().max(200).optional() }, TI, ), + get: get("/title-aliases/{id}", "别名详情", TitleAlias, TI), create: write("post", "/title-aliases", "organize.execute", "新增别名", TitleAliasCreate, TitleAlias, TI, false), remove: write("delete", "/title-aliases/{id}", "organize.execute", "删除别名", undefined, OkResult, TI), }, diff --git a/apps/server/src/modules/sources-identity/routes/identity.ts b/apps/server/src/modules/sources-identity/routes/identity.ts index 9c8f2f87..15b37257 100644 --- a/apps/server/src/modules/sources-identity/routes/identity.ts +++ b/apps/server/src/modules/sources-identity/routes/identity.ts @@ -17,7 +17,7 @@ import { ReleaseIdentityAssign, type TitleAlias, } from "@umt/contracts"; -import { sql } from "@umt/db"; +import { type DB, type Kysely, sql } from "@umt/db"; import { normalizeTitle } from "@umt/domain"; import { z } from "zod"; import { ApiException } from "../../../http/errors.ts"; @@ -44,6 +44,30 @@ const T_ID = ["identity"]; const Accepted = z.object({ taskId: Id }); const FieldParam = z.object({ id: Id, field: EntityMetadataField }); +/** 列表带作品标题:按 media_entity_id 批量取;只挂在内容单元上的行以 `cu:` 为键取所属作品标题 */ +async function entityTitles( + db: Kysely, + rows: readonly { media_entity_id: string | null; content_unit_id?: string | null }[], +): Promise> { + const out = new Map(); + const ids = [...new Set(rows.map((r) => r.media_entity_id).filter((x): x is string => !!x))]; + const cus = [...new Set(rows.filter((r) => !r.media_entity_id && r.content_unit_id).map((r) => r.content_unit_id!))]; + if (ids.length) { + for (const e of await db.selectFrom("media_entity").select(["id", "title"]).where("id", "in", ids).execute()) + out.set(e.id, e.title); + } + if (cus.length) { + const units = await db + .selectFrom("content_unit as u") + .innerJoin("media_entity as e", "e.id", "u.media_entity_id") + .select(["u.id", "e.title"]) + .where("u.id", "in", cus) + .execute(); + for (const u of units) out.set(`cu:${u.id}`, u.title); + } + return out; +} + export function identityRoutes(app: ModuleApp, rt: () => SourcesRuntime) { const db = () => rt().ctx.db; @@ -329,7 +353,16 @@ export function identityRoutes(app: ModuleApp, rt: () => SourcesRuntime) { .limit(query.limit + 1) .execute(); const p = page(rows, off, query.limit); - return { items: toApiList(p.items), nextCursor: p.nextCursor }; + const titles = await entityTitles(db(), p.items); + return { + items: p.items.map((r) => ({ ...toApi(r), mediaEntityTitle: titles.get(r.media_entity_id) ?? null })), + nextCursor: p.nextCursor, + }; + }); + route(app, D.titleAliases.get, async ({ params }) => { + const row = await db().selectFrom("title_alias").selectAll().where("id", "=", params.id).executeTakeFirst(); + if (!row) throw new ApiException("not_found", "别名不存在"); + return toApi(row); }); route(app, D.titleAliases.create, async ({ body, actor }) => { const id = await resolveEntityId(db(), body.mediaEntityId); @@ -379,7 +412,14 @@ export function identityRoutes(app: ModuleApp, rt: () => SourcesRuntime) { .limit(query.limit + 1) .execute(); const p = page(rows, off, query.limit); - return { items: toApiList(p.items), nextCursor: p.nextCursor }; + const titles = await entityTitles(db(), p.items); + return { + items: p.items.map((r) => ({ + ...toApi(r), + mediaEntityTitle: titles.get(r.media_entity_id ?? `cu:${r.content_unit_id}`) ?? null, + })), + nextCursor: p.nextCursor, + }; }); route(app, D.externalIdentities.get, async ({ params }) => { const row = await db().selectFrom("external_identity").selectAll().where("id", "=", params.id).executeTakeFirst(); diff --git a/apps/server/src/modules/sources-identity/routes/sources.ts b/apps/server/src/modules/sources-identity/routes/sources.ts index ef15a375..3f22173c 100644 --- a/apps/server/src/modules/sources-identity/routes/sources.ts +++ b/apps/server/src/modules/sources-identity/routes/sources.ts @@ -22,7 +22,7 @@ import { ApiException } from "../../../http/errors.ts"; import type { ModuleApp } from "../../types.ts"; import { D } from "../defs.ts"; import { route } from "../http.ts"; -import { decodeCursor, jsonb, page, toApi, toApiList } from "../rows.ts"; +import { decodeCursor, feedToApi, jsonb, page, toApi, toApiList } from "../rows.ts"; import type { SourcesRuntime } from "../runtime.ts"; import { audit, closeDecisions } from "../service/common.ts"; import { assignReleaseIdentity } from "../service/identity-admin.ts"; @@ -328,9 +328,9 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) { .limit(query.limit + 1) .execute(); const p = page(rows, off, query.limit); - return { items: toApiList(p.items), nextCursor: p.nextCursor }; + return { items: p.items.map((r) => feedToApi(r)), nextCursor: p.nextCursor }; }); - route(app, D.feedSources.get, async ({ params }) => toApi(await getFeed(params.id))); + route(app, D.feedSources.get, async ({ params }) => feedToApi(await getFeed(params.id))); route(app, D.feedSources.create, async ({ body, actor }) => { if (body.siteId) await getSite(body.siteId); if (body.siteAccountId) await checkAccount(body.siteAccountId, body.siteId ?? null); @@ -352,7 +352,7 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) { .returningAll() .executeTakeFirstOrThrow(); await audit(db(), actor, "feed_source.created", { type: "feed_source", id: row.id }, { url: row.url_display }); - return toApi(row); + return feedToApi(row); }); route(app, D.feedSources.update, async ({ params, body, actor }) => { const feed = await getFeed(params.id); @@ -402,7 +402,7 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) { fields: Object.keys(patch).filter((k) => k !== "url_secret_ref"), }, ); - return toApi(row); + return feedToApi(row); }); for (const enabled of [true, false]) { route( @@ -432,7 +432,7 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) { type: "feed_source", id: params.id, }); - return toApi(row); + return feedToApi(row); }, ); } @@ -485,7 +485,7 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) { await audit(trx, actor, "feed_source.reset", { type: "feed_source", id: params.id }); return r; }); - return toApi(row); + return feedToApi(row); }, ); route(app, D.feedSources.fetchNow, async ({ params }) => { diff --git a/apps/server/src/modules/sources-identity/rows.ts b/apps/server/src/modules/sources-identity/rows.ts index 27dde3d9..4d91532c 100644 --- a/apps/server/src/modules/sources-identity/rows.ts +++ b/apps/server/src/modules/sources-identity/rows.ts @@ -35,3 +35,18 @@ export function page(rows: T[], offset: number, limit: number): { items: T[]; /** jsonb 写入:数组/对象都先 JSON.stringify(pg 会把 JS 数组当 PG 数组) */ export const jsonb = (v: unknown): string => JSON.stringify(v ?? null); + +/** feed_source 行 → FeedSource(补派生字段 possibleGap,取自 cursor.gap) */ +export function feedToApi(row: { cursor: unknown }): T { + const gap = (row.cursor as { gap?: { detectedAt?: unknown; previousFetchedAt?: unknown; backfillQueued?: unknown } }) + ?.gap; + const possibleGap = + gap && typeof gap.detectedAt === "string" + ? { + detectedAt: gap.detectedAt, + previousFetchedAt: typeof gap.previousFetchedAt === "string" ? gap.previousFetchedAt : null, + backfillQueued: gap.backfillQueued === true, + } + : null; + return { ...toApi>(row), possibleGap } as T; +} diff --git a/apps/server/src/modules/sources-identity/service/feeds.ts b/apps/server/src/modules/sources-identity/service/feeds.ts index 48fc0dc2..1bcada56 100644 --- a/apps/server/src/modules/sources-identity/service/feeds.ts +++ b/apps/server/src/modules/sources-identity/service/feeds.ts @@ -17,9 +17,11 @@ import { clip, type Exec, openDecision } from "./common.ts"; // FeedFetchState / FeedCursor 持久化在 feed_source 行;enclosure 只存脱敏值(适配器已脱敏)。 /** feed_source.cursor 的形状:重叠窗口游标 + 熔断打开时间 */ -interface StoredCursor { +export interface StoredCursor { window?: FeedCursor; circuitOpenedAt?: string; + /** 最近一次疑似缺口(窗口零重叠);下一次成功抓取重新衔接后清空 */ + gap?: { detectedAt: string; previousFetchedAt: string | null; backfillQueued: boolean }; } const AUTH_BACKOFF_MS = 24 * 3600_000; @@ -144,9 +146,21 @@ export async function ingestFeed( const lastError = auth ? { ...(state.lastError ?? {}), kind: "auth", message: "Feed 凭证失效或过期(401/403/过期提示),已停止自动重试" } : (state.lastError ?? null); + // 疑似缺口:本次 ok 且零重叠 → 记下;本次 ok 且衔接上 → 清空;抓取失败 → 保留上次状态 + const gap: StoredCursor["gap"] = + result.kind === "ok" + ? out.possibleGap + ? { + detectedAt: now.toISOString(), + previousFetchedAt: cursor.window?.lastFetchedAt ?? null, + backfillQueued: !!site?.id, + } + : undefined + : cursor.gap; const nextCursor: StoredCursor = { window, ...(state.circuit === "open" && state.circuitOpenedAt ? { circuitOpenedAt: state.circuitOpenedAt } : {}), + ...(gap ? { gap } : {}), }; await trx .updateTable("feed_source") diff --git a/apps/server/test/api-registry.test.ts b/apps/server/test/api-registry.test.ts index 0eb3b51b..66f515a1 100644 --- a/apps/server/test/api-registry.test.ts +++ b/apps/server/test/api-registry.test.ts @@ -34,5 +34,10 @@ describe("Api 注册表覆盖", () => { const registered = new Set(listEndpoints().map((e) => `${e.method.toUpperCase()} ${e.path}`)); const missing = served.filter((k) => !registered.has(k)).sort(); expect(missing, `以下路由未登记到 packages/contracts/src/api.ts:\n${missing.join("\n")}`).toEqual([]); + + // 反向:api.ts 登记的每条端点都必须有实现(OpenAPI 里可见) + const servedSet = new Set(served); + const unimplemented = [...registered].filter((k) => !servedSet.has(k)).sort(); + expect(unimplemented, `以下端点登记在 api.ts 但未实现:\n${unimplemented.join("\n")}`).toEqual([]); }); }); diff --git a/apps/server/test/e2e/e6-w4a-endpoints.test.ts b/apps/server/test/e2e/e6-w4a-endpoints.test.ts new file mode 100644 index 00000000..8261c04f --- /dev/null +++ b/apps/server/test/e2e/e6-w4a-endpoints.test.ts @@ -0,0 +1,148 @@ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + completeDownload, + createE2E, + type E2E, + pgAvailable, + seedBase, + seedMovieTarget, + sql, + waitFor, +} from "./harness.ts"; + +// E6 W4-A 前后端集成端点:手动提交下载(与自动同流程)、候选评估去重、库存叠加补字段、 +// 整理记录 downloadTaskId、只读资产/产物端点、偏好标签修订历史、列表带作品标题、事件 JSON 补读。 + +const hasPg = await pgAvailable(); + +describe.skipIf(!hasPg)("E6 W4-A 集成端点", () => { + let e: E2E; + let target: Awaited>; + + beforeAll(async () => { + e = await createE2E("e6"); + const base = await seedBase(e); + target = await seedMovieTarget(e, base.siteId); + }, 120_000); + + afterAll(async () => { + await e?.close(); + }, 60_000); + + it("POST /acquisition-plans:硬条件不满足 → 不提交并给出原因;重复评估只留一条候选记录", async () => { + for (let i = 0; i < 2; i++) { + const r = await e.api("POST", "/acquisition-plans", { sourceReleaseId: target.r720, slotId: target.slotId }); + expect(r.status).toBe(200); + expect(r.body.data).toMatchObject({ submitted: false, plan: null, taskId: null }); + expect(r.body.data.rejected.length).toBeGreaterThan(0); + expect(r.body.data.rejected[0].slotId).toBe(target.slotId); + } + const n = await sql<{ n: number }>`select count(*)::int as n from artifact + where title = 'candidate_evaluation' and subject_id = ${target.slotId} + and content->>'sourceReleaseId' = ${target.r720}`.execute(e.db); + expect(n.rows[0]?.n).toBe(1); + }); + + it("GET /inventory:subscribed 与 candidateReasons", async () => { + const r = await e.api("GET", `/inventory?entityIds=${target.entityId}`); + expect(r.status).toBe(200); + const entry = r.body.data.items[0]; + expect(entry.subscribed).toBe(true); + const c = entry.candidateReasons.find((x: { sourceReleaseId: string }) => x.sourceReleaseId === target.r720); + expect(c).toMatchObject({ slotId: target.slotId, slotName: "1080p", decision: "reject" }); + expect(c.reasons.length).toBeGreaterThan(0); + }); + + it("POST /acquisition-plans:按作品选 Slot,走 download.plan → download.submit,qB 收到种子;整理记录带 downloadTaskId", async () => { + const r = await e.api("POST", "/acquisition-plans", { + sourceReleaseId: target.r1080, + entityId: target.entityId, + note: "工作台手动", + }); + expect(r.status).toBe(200); + expect(r.body.data).toMatchObject({ submitted: true, slotId: target.slotId }); + expect(r.body.data.plan.status).toMatch(/reserved|submitted/); + expect(r.body.data.taskId).toBeTruthy(); + const audits = await e.db + .selectFrom("audit_event") + .select(["action", "target_id"]) + .where("action", "in", ["capability.invoke", "acquisition_plan.manual_submit"]) + .execute(); + expect(audits.filter((a) => a.target_id === "download.submit")).toHaveLength(1); + expect(audits.some((a) => a.action === "acquisition_plan.manual_submit")).toBe(true); + + await completeDownload(e, "Dune.Part.Two.2024.1080p.BluRay.x264-GRP"); + const rec = await waitFor( + async () => { + const l = await e.api("GET", "/organization-records"); + return l.body.data.items.find((x: { status: string }) => x.status === "succeeded"); + }, + 60_000, + "整理记录成功", + ); + const dt = await e.db.selectFrom("download_task").select("id").executeTakeFirstOrThrow(); + expect(rec.downloadTaskId).toBe(dt.id); + const filtered = await e.api("GET", `/organization-records?downloadTaskId=${dt.id}`); + expect(filtered.body.data.items.map((x: { id: string }) => x.id)).toEqual([rec.id]); + }, 120_000); + + it("只读资产端点与 Slot 当前资产摘要", async () => { + const assets = await e.api("GET", `/media-assets?mediaEntityId=${target.entityId}`); + expect(assets.status).toBe(200); + expect(assets.body.data.items.length).toBeGreaterThan(0); + const a = assets.body.data.items[0]; + expect((await e.api("GET", `/media-assets/${a.id}`)).body.data.id).toBe(a.id); + const locs = await e.api("GET", `/asset-locations?mediaAssetId=${a.id}`); + expect(locs.body.data.items.length).toBeGreaterThan(0); + expect((await e.api("GET", `/path-bindings?mediaEntityId=${target.entityId}`)).body.data.items.length).toBe(1); + expect((await e.api("GET", "/media-assets/00000000-0000-4000-8000-000000000000")).status).toBe(404); + expect((await e.api("PATCH", `/media-assets/${a.id}`, { status: "missing" })).status).toBe(404); + + const sub = await e.db.selectFrom("subscription").select("id").executeTakeFirstOrThrow(); + const slots = await e.api("GET", `/acquisition-slots?subscriptionId=${sub.id}`); + expect(slots.body.data.items).toHaveLength(1); + expect(slots.body.data.items[0].currentAsset).toMatchObject({ resolution: "1080p" }); + }); + + it("GET /artifacts:列表与详情", async () => { + const list = await e.api("GET", "/artifacts?kind=evidence"); + expect(list.status).toBe(200); + expect(list.body.data.items.length).toBeGreaterThan(0); + const id = list.body.data.items[0].id; + const one = await e.api("GET", `/artifacts/${id}`); + expect(one.body.data).toMatchObject({ id, kind: "evidence" }); + }); + + it("列表带作品标题;订阅按标题 / 类型搜索", async () => { + const subs = await e.api("GET", "/subscriptions?q=Dune&mediaKind=movie"); + expect(subs.body.data.items).toHaveLength(1); + expect(subs.body.data.items[0].mediaEntityTitle).toBe("沙丘2"); + expect((await e.api("GET", "/subscriptions?mediaKind=series")).body.data.items).toHaveLength(0); + await e.api("POST", "/title-aliases", { mediaEntityId: target.entityId, alias: "沙丘:第二部", source: "user" }); + const aliases = await e.api("GET", `/title-aliases?mediaEntityId=${target.entityId}`); + expect(aliases.body.data.items[0].mediaEntityTitle).toBe("沙丘2"); + const one = await e.api("GET", `/title-aliases/${aliases.body.data.items[0].id}`); + expect(one.status).toBe(200); + }); + + it("偏好标签修订历史:创建 / 修改 / 删除各一条(新→旧)", async () => { + const c = await e.api("POST", "/preference-tags", { label: "偏好 FLAC", kind: "soft", rule: { audio: "flac" } }); + expect(c.status).toBe(201); + const id = c.body.data.id; + await e.api("PATCH", `/preference-tags/${id}`, { label: "偏好无损音轨" }); + await e.api("DELETE", `/preference-tags/${id}`); + const r = await e.api("GET", `/preference-tags/${id}/revisions`); + expect(r.status).toBe(200); + expect(r.body.data.items.map((x: { action: string }) => x.action)).toEqual(["deleted", "updated", "created"]); + expect(r.body.data.items[1]).toMatchObject({ changedFields: ["label"], revision: 2 }); + expect(r.body.data.items[1].snapshot.label).toBe("偏好无损音轨"); + }); + + it("GET /events?after=&limit= 返回 JSON 补读页", async () => { + e.bus.publishSync("test.ping", { n: 1 }); + const r = await e.api("GET", "/events?after=0&limit=5"); + expect(r.status).toBe(200); + expect(r.body.data.items.length).toBeGreaterThan(0); + expect(r.body.data.nextAfter).toBe(r.body.data.items.at(-1).seq); + }); +}); diff --git a/apps/server/test/sources-identity/ingest.test.ts b/apps/server/test/sources-identity/ingest.test.ts index 8baf0fb4..6be24ec3 100644 --- a/apps/server/test/sources-identity/ingest.test.ts +++ b/apps/server/test/sources-identity/ingest.test.ts @@ -138,6 +138,15 @@ describe.skipIf(!up)("ingestFeed", () => { at: new Date(), }); expect(route).toMatchObject({ task: "snapshotSite", input: { siteId, kind: "backfill", maxPages: 3 } }); + // 响应派生字段 possibleGap;下一次衔接上的抓取后清空 + const got = (await (await h.req(`/feed-sources/${f.id}`)).json()) as { + data: { possibleGap: { backfillQueued: boolean } | null }; + }; + expect(got.data.possibleGap).toMatchObject({ backfillQueued: true }); + s.setItems([item("gap-9", "900209", "C.2020.1080p-X"), item("gap-10", "900210", "D.2020.1080p-X")]); + await ingestFeed(h.rt, f.id); + const after = (await (await h.req(`/feed-sources/${f.id}`)).json()) as { data: { possibleGap: unknown } }; + expect(after.data.possibleGap).toBeNull(); }); it("登录失效(403):账号标失效 + 待处理,不重试", async () => { diff --git a/apps/server/test/sse.test.ts b/apps/server/test/sse.test.ts index 986c03e5..992a9eaa 100644 --- a/apps/server/test/sse.test.ts +++ b/apps/server/test/sse.test.ts @@ -39,3 +39,29 @@ describe("GET /api/v1/events", () => { await reader.cancel(); }); }); + +describe("GET /api/v1/events JSON 补读", () => { + it("带 limit 返回 JSON 页,按 nextAfter 续读", async () => { + const { app, bus } = testApp(); + for (let i = 0; i < 5; i++) bus.publishSync("t", { i }); + const h = { Authorization: `Bearer ${DEV_KEY}` }; + const r1 = await app.request("/api/v1/events?after=1&limit=2", { headers: h }); + expect(r1.headers.get("content-type")).toContain("application/json"); + const b1 = (await r1.json()) as { data: { items: { seq: number }[]; nextAfter: number; hasMore: boolean } }; + expect(b1.data.items.map((e) => e.seq)).toEqual([2, 3]); + expect(b1.data).toMatchObject({ nextAfter: 3, hasMore: true }); + const r2 = await app.request(`/api/v1/events?after=${b1.data.nextAfter}&limit=10`, { headers: h }); + const b2 = (await r2.json()) as { data: { items: { seq: number }[]; hasMore: boolean; latestSeq: number } }; + expect(b2.data.items.map((e) => e.seq)).toEqual([4, 5]); + expect(b2.data).toMatchObject({ hasMore: false, latestSeq: 5 }); + }); + + it("Accept: application/json 也走 JSON;非法 limit 400", async () => { + const { app, bus } = testApp(); + bus.publishSync("a", 1); + const h = { Authorization: `Bearer ${DEV_KEY}` }; + const r = await app.request("/api/v1/events", { headers: { ...h, Accept: "application/json" } }); + expect(((await r.json()) as { data: { items: unknown[] } }).data.items).toHaveLength(1); + expect((await app.request("/api/v1/events?limit=0", { headers: h })).status).toBe(400); + }); +}); diff --git a/apps/web/app/api/feeds.ts b/apps/web/app/api/feeds.ts index c885d1ea..cb3e5767 100644 --- a/apps/web/app/api/feeds.ts +++ b/apps/web/app/api/feeds.ts @@ -72,14 +72,16 @@ export function useFetchFeedNow() { } /** - * 疑似缺口:Feed 窗口与上次零重叠时服务端发 feed.gap_suspected 并排补漏。 - * 列表字段里没有显式标记,这里按 cursor 里的 gapSuspectedAt(若服务端写入)或"距上次成功超过 3 个周期"推断,并标"疑似"。 + * 疑似缺口:服务端在 Feed 窗口与上次零重叠时写 possibleGap 并排站点补漏;下一次抓取衔接上后清空。 + * 没有服务端标记但"距上次成功超过 3 个周期"时,只作推断提示(标"可能")。 */ export function feedGapHint(f: FeedSource, now = Date.now()): string | null { - const gapAt = typeof f.cursor.gapSuspectedAt === "string" ? f.cursor.gapSuspectedAt : null; - if (gapAt) return `疑似缺口(${new Date(gapAt).toLocaleString("zh-CN", { hour12: false })} 检测,已排补漏)`; + if (f.possibleGap) { + const at = new Date(f.possibleGap.detectedAt).toLocaleString("zh-CN", { hour12: false }); + return `疑似缺口(${at} 检测,${f.possibleGap.backfillQueued ? "已排补漏" : "Feed 未关联站点,无法自动补漏"})`; + } if (!f.enabled || !f.lastSuccessAt) return null; const gapMs = now - new Date(f.lastSuccessAt).getTime(); - if (gapMs > f.pollIntervalSec * 1000 * 3) return "距上次成功已超过 3 个周期,可能漏项(疑似)"; + if (gapMs > f.pollIntervalSec * 1000 * 3) return "距上次成功已超过 3 个周期,可能漏项"; return null; } diff --git a/apps/web/app/api/inventory.ts b/apps/web/app/api/inventory.ts index 234aef61..c4fa12e9 100644 --- a/apps/web/app/api/inventory.ts +++ b/apps/web/app/api/inventory.ts @@ -16,11 +16,29 @@ export function useInventory(entityIds: string[]) { } /** 叠加标记:只来自本地身份/Slot/任务,不从网页标题猜确定事实 */ -export type OverlayFlag = { key: string; label: string; tone: "success" | "warning" | "neutral" }; +export type OverlayFlag = { key: string; label: string; tone: "success" | "warning" | "neutral"; title?: string }; -export function inventoryFlags(entry: InventoryEntry | undefined, releaseResolution?: string | null): OverlayFlag[] { +const DECISION_FLAG: Record = { + accept: { label: "已选中", tone: "success" }, + upgrade: { label: "可升级", tone: "success" }, + reject: { label: "未选", tone: "warning" }, + unknown_field: { label: "字段未知", tone: "warning" }, + skipped: { label: "被跳过", tone: "neutral" }, +}; + +/** 该资源在各 Slot 上的最近一次评估(服务端每个 (Slot, 资源) 只保留一条) */ +export function releaseCandidates(entry: InventoryEntry | undefined, releaseId: string) { + return entry?.candidateReasons.filter((c) => c.sourceReleaseId === releaseId) ?? []; +} + +export function inventoryFlags( + entry: InventoryEntry | undefined, + releaseResolution?: string | null, + releaseId?: string, +): OverlayFlag[] { if (!entry) return []; const flags: OverlayFlag[] = []; + if (entry.subscribed) flags.push({ key: "sub", label: "已订阅", tone: "neutral" }); const available = entry.versions.filter((v) => v.available); if (available.length) { const res = [...new Set(available.map((v) => v.resolution ?? "未知分辨率"))].join(" / "); @@ -35,6 +53,15 @@ export function inventoryFlags(entry: InventoryEntry | undefined, releaseResolut if (entry.downloading.length) { flags.push({ key: "dl", label: `下载中 ${entry.downloading.length}`, tone: "neutral" }); } + for (const c of releaseId ? releaseCandidates(entry, releaseId) : []) { + const d = DECISION_FLAG[c.decision] ?? { label: c.decision, tone: "neutral" as const }; + flags.push({ + key: `cand-${c.slotId}`, + label: `${c.slotName}:${d.label}`, + tone: d.tone, + title: c.reasons.join(";"), + }); + } if (releaseResolution && releaseResolution === "resolved" && flags.length === 0) { flags.push({ key: "tracked", label: "已识别,无版本目标", tone: "neutral" }); } diff --git a/apps/web/app/components/library/slot-card.tsx b/apps/web/app/components/library/slot-card.tsx index 0cfe0ceb..037f6353 100644 --- a/apps/web/app/components/library/slot-card.tsx +++ b/apps/web/app/components/library/slot-card.tsx @@ -26,6 +26,10 @@ export function SlotCard({ const hard = summarizeHard(slot.hardConditions, slot.fallback); const soft = summarizeSoft(slot.softPreferences); const upgrade = summarizeUpgrade(slot.upgradeThreshold); + // 服务端按 (Slot, 来源资源) 只保留最新一次评估;这里再按资源去重兜底(列表已按评估时间新→旧) + const latest = candidates.filter( + (c, i) => candidates.findIndex((x) => x.sourceReleaseId === c.sourceReleaseId) === i, + ); return (
@@ -70,12 +74,12 @@ export function SlotCard({
{currentAsset ?? "无"}
- {candidates.length ? ( + {latest.length ? (
-

最近候选 · 为什么选 / 不选

+

最近候选 · 为什么选 / 不选(每个资源显示最近一次评估)

    - {candidates.map((c) => ( -
  • + {latest.map((c) => ( +
  • {c.title} @@ -92,7 +96,7 @@ export function SlotCard({ {c.reasons.length ? (

    {c.reasons.join(";")}

    ) : null} -

    {formatDateTime(c.evaluatedAt)}

    +

    评估于 {formatDateTime(c.evaluatedAt)}

    • ))}
diff --git a/apps/web/app/components/workbench/release-detail-sheet.tsx b/apps/web/app/components/workbench/release-detail-sheet.tsx index d029a3ce..28991bf4 100644 --- a/apps/web/app/components/workbench/release-detail-sheet.tsx +++ b/apps/web/app/components/workbench/release-detail-sheet.tsx @@ -65,7 +65,8 @@ function LocalInventory({ release, entry }: { release: WorkbenchRelease; entry: ) : ( <>

- {slots.length ? `${slots.length} 个版本目标(已订阅)` : "没有版本目标(未订阅)"} + {slots.length ? `${slots.length} 个版本目标` : "没有版本目标"} + {entry ? (entry.subscribed ? " · 已订阅" : " · 未订阅") : ""}

{candidates.length ? (
    @@ -73,7 +74,7 @@ function LocalInventory({ release, entry }: { release: WorkbenchRelease; entry: const d = DECISION_LABEL[c.decision]; const slot = slots.find((s) => s.id === c.slotId); return ( -
  • +
  • {d.label} {slot ? `目标 ${slot.name}` : "版本目标"} diff --git a/apps/web/app/components/workbench/release-list.tsx b/apps/web/app/components/workbench/release-list.tsx index 3e0784a4..42368f38 100644 --- a/apps/web/app/components/workbench/release-list.tsx +++ b/apps/web/app/components/workbench/release-list.tsx @@ -44,7 +44,7 @@ export function InventoryOverlay({ : "未识别到本地作品"; return {label}; } - const flags = inventoryFlags(entry, release.resolutionStatus); + const flags = inventoryFlags(entry, release.resolutionStatus, release.id); return ( @@ -53,11 +53,17 @@ export function InventoryOverlay({ {loading ? 库存加载中… : null} {!loading && entry && flags.length === 0 ? 本地无版本 : null} - {flags.map((f) => ( - - {f.label} - - ))} + {flags.map((f) => + f.title ? ( + + {f.label} + + ) : ( + + {f.label} + + ), + )} ); } diff --git a/apps/web/app/routes/console/sources.tsx b/apps/web/app/routes/console/sources.tsx index 702b27a0..03b90c8f 100644 --- a/apps/web/app/routes/console/sources.tsx +++ b/apps/web/app/routes/console/sources.tsx @@ -114,7 +114,11 @@ export default function SourcesPage() { return (
    {HEALTH[f.healthStatus].label} - {gap ? 疑似缺口 : null} + {gap ? ( + + {f.possibleGap ? "疑似缺口" : "可能漏项"} + + ) : null}
    ); }, diff --git a/packages/contracts/src/api.ts b/packages/contracts/src/api.ts index 6e538884..a19edfe4 100644 --- a/packages/contracts/src/api.ts +++ b/packages/contracts/src/api.ts @@ -5,6 +5,7 @@ import * as auth from "./auth.ts"; import { CapabilityDescriptor, KnownCapabilityId } from "./capability.ts"; import { type ApiScope, CursorQuery, cursorPage, HealthStatus, Id, IdParam } from "./common.ts"; import * as content from "./content.ts"; +import * as events from "./events.ts"; import * as execution from "./execution.ts"; import * as identity from "./identity.ts"; import * as instances from "./instances.ts"; @@ -111,10 +112,16 @@ const instanceTest = (path: string, summary: string) => ep({ method: "post", path, scope: "admin", summary, params: IdParam, response: instances.InstanceTestResult }); /** 入队后台任务的受理结果(taskId = 引擎 run id) */ -const Accepted = z.object({ taskId: Id }); +export const TaskAccepted = z.object({ taskId: Id }); +export type TaskAccepted = z.infer; -/** 人工指定来源资源身份的结果 */ -const ReleaseIdentityResult = z.object({ mediaEntityId: Id, contentUnitIds: z.array(Id) }); +/** 人工指定来源资源身份的结果(待识别池 assign / source-releases identity) */ +export const ReleaseIdentityResult = z.object({ mediaEntityId: Id, contentUnitIds: z.array(Id) }); +export type ReleaseIdentityResult = z.infer; + +/** 身份修订历史 */ +export const IdentityRevisionList = z.object({ items: z.array(identity.IdentityRevision) }); +export type IdentityRevisionList = z.infer; /** 字段覆盖/取消覆盖的结果 */ const FieldOverrideResult = z.object({ @@ -128,7 +135,7 @@ const FieldParam = z.object({ id: Id, field: content.EntityMetadataField }); const EngineRunIdParam = z.object({ id: z.string().min(1).max(200) }); /** 待处理事项聚合分组 */ -const DecisionSummaryGroup = z.object({ +export const DecisionSummaryGroup = z.object({ kind: agent.DecisionKind, mediaEntityId: Id.nullable(), mediaEntityTitle: z.string().nullable(), @@ -136,6 +143,9 @@ const DecisionSummaryGroup = z.object({ maxSeverity: z.enum(["info", "warning", "critical"]), latestAt: z.string(), }); +export type DecisionSummaryGroup = z.infer; +export const DecisionSummary = z.object({ total: z.number().int(), groups: z.array(DecisionSummaryGroup) }); +export type DecisionSummary = z.infer; export const Api = { // ── 内容 / 身份 ── @@ -192,7 +202,7 @@ export const Api = { scope: "organize.execute", summary: "刷新元数据(字段级;用户锁优先,不动路径)", params: IdParam, - response: Accepted, + response: TaskAccepted, }), setField: ep({ method: "put", @@ -217,7 +227,6 @@ export const Api = { "内容单元", { row: content.ContentUnit, - create: content.ContentUnitCreate, update: content.ContentUnitUpdate, filters: { mediaEntityId: Id.optional(), seasonNumber: z.coerce.number().int().optional() }, }, @@ -230,7 +239,6 @@ export const Api = { { row: content.TitleAlias, create: content.TitleAliasCreate, - update: content.TitleAliasUpdate, filters: { mediaEntityId: Id.optional(), q: z.string().max(200).optional() }, }, R, @@ -259,17 +267,16 @@ export const Api = { scope: "read", summary: "身份修订历史", params: IdParam, - response: z.object({ items: z.array(identity.IdentityRevision) }), + response: IdentityRevisionList, }), }, - // ── 文件 ── + // ── 文件(只读;写入只经整理/对账流程,不开放直接改) ── mediaAssets: crud( "/media-assets", "资产", { row: assets.MediaAsset, - update: assets.MediaAssetUpdate, filters: { mediaEntityId: Id.optional(), status: assets.AssetStatus.optional() }, }, R, @@ -279,8 +286,6 @@ export const Api = { "资产覆盖", { row: assets.AssetUnitCoverage, - create: assets.AssetUnitCoverageCreate, - update: assets.AssetUnitCoverageUpdate, filters: { mediaAssetId: Id.optional(), contentUnitId: Id.optional() }, }, R, @@ -290,7 +295,6 @@ export const Api = { "资产位置", { row: assets.AssetLocation, - update: assets.AssetLocationUpdate, filters: { mediaAssetId: Id.optional(), status: assets.LocationStatus.optional() }, }, R, @@ -300,8 +304,6 @@ export const Api = { "路径绑定", { row: assets.PathBinding, - create: assets.PathBindingCreate, - update: assets.PathBindingUpdate, filters: { mediaEntityId: Id.optional(), status: assets.BindingStatus.optional() }, }, R, @@ -405,7 +407,7 @@ export const Api = { summary: "立即同步站点快照(受站点限流)", params: IdParam, body: z.object({ maxPages: z.number().int().min(1).max(20).optional() }), - response: Accepted, + response: TaskAccepted, }), snapshots: ep({ method: "get", @@ -544,7 +546,7 @@ export const Api = { scope: "organize.execute", summary: "解除来源资源的人工锁并重新识别", params: IdParam, - response: Accepted, + response: TaskAccepted, }), }, observations: crud( @@ -565,7 +567,13 @@ export const Api = { row: policy.Subscription, create: policy.SubscriptionCreate, update: policy.SubscriptionUpdate, - filters: { status: policy.SubscriptionStatus.optional(), mediaEntityId: Id.optional() }, + filters: { + status: policy.SubscriptionStatus.optional(), + mediaEntityId: Id.optional(), + mediaKind: policy.SubscriptionMediaKind.optional(), + /** 作品标题 / 原名 / 别名 / 未建档检索词,模糊匹配 */ + q: z.string().max(200).optional(), + }, }, R, ), @@ -614,7 +622,11 @@ export const Api = { row: policy.AcquisitionSlot, create: policy.AcquisitionSlotCreate, update: policy.AcquisitionSlotUpdate, - filters: { mediaEntityId: Id.optional(), status: policy.SlotStatus.optional() }, + filters: { + mediaEntityId: Id.optional(), + status: policy.SlotStatus.optional(), + subscriptionId: Id.optional(), + }, }, R, ), @@ -632,6 +644,14 @@ export const Api = { ), /** 删除 = 标记 deleted + 写抑制记录 */ remove: del("/preference-tags/{id}", "删除偏好标签(留抑制记录)"), + revisions: ep({ + method: "get", + path: "/preference-tags/{id}/revisions", + scope: "read", + summary: "偏好标签修订历史(新→旧;取自审计事件)", + params: IdParam, + response: policy.PreferenceTagRevisionList, + }), }, preferenceSuppressions: crud( "/preference-suppressions", @@ -686,15 +706,26 @@ export const Api = { response: z.object({ items: z.array(execution.AssetOperation) }), }), }, - acquisitionPlans: crud( - "/acquisition-plans", - "下载计划", - { - row: execution.AcquisitionPlan, - filters: { slotId: Id.optional(), status: execution.PlanStatus.optional() }, - }, - R, - ), + acquisitionPlans: { + ...crud( + "/acquisition-plans", + "下载计划", + { + row: execution.AcquisitionPlan, + filters: { slotId: Id.optional(), status: execution.PlanStatus.optional() }, + }, + R, + ), + create: ep({ + method: "post", + path: "/acquisition-plans", + scope: "organize.execute", + summary: + "手动提交下载:与自动收录同一流程(download.plan → download.submit,含硬条件/预算/缓存阈值/授权检查);未生成计划时返回拒绝原因", + body: execution.ManualAcquisitionRequest, + response: execution.ManualAcquisitionResult, + }), + }, downloadTasks: { ...crud( "/download-tasks", @@ -753,7 +784,7 @@ export const Api = { scope: "read", summary: "未关闭待处理事项按作品与问题类型聚合", query: z.object({ status: agent.DecisionStatus.optional() }), - response: z.object({ total: z.number().int(), groups: z.array(DecisionSummaryGroup) }), + response: DecisionSummary, }), }, approvals: { @@ -849,7 +880,6 @@ export const Api = { "校验运行", { row: ops.ReconcileRun, - create: ops.ReconcileRunCreate, filters: { kind: ops.ReconcileKind.optional(), status: ops.ReconcileStatus.optional() }, }, R, @@ -859,7 +889,6 @@ export const Api = { "库存映射", { row: ops.ServerMapping, - update: ops.ServerMappingUpdate, filters: { mediaServerInstanceId: Id.optional(), mediaEntityId: Id.optional(), @@ -1056,6 +1085,19 @@ export const Api = { }), }, + // ── 事件 ── + events: { + list: ep({ + method: "get", + path: "/events", + scope: "read", + summary: + "事件流:控制台用 SSE(text/event-stream,cookie 会话);API Key 调用者带 limit 取 JSON 补读页(after=上次 nextAfter)", + query: events.EventReplayQuery, + response: events.EventReplayPage, + }), + }, + // ── 系统 ── system: { health: ep({ @@ -1137,7 +1179,7 @@ export const Api = { scope: "organize.execute", summary: "立即重新识别", params: IdParam, - response: Accepted, + response: TaskAccepted, }), }, diff --git a/packages/contracts/src/content.ts b/packages/contracts/src/content.ts index af53ce5c..6691502c 100644 --- a/packages/contracts/src/content.ts +++ b/packages/contracts/src/content.ts @@ -113,6 +113,8 @@ export const TitleAlias = EntityBase.extend({ kind: AliasKind, source: AliasSource, locked: z.boolean(), + /** 派生字段(列表返回):作品标题 */ + mediaEntityTitle: z.string().nullable().optional(), }); export type TitleAlias = z.infer; /** normalized 由服务端计算,不接受客户端传入 */ diff --git a/packages/contracts/src/events.ts b/packages/contracts/src/events.ts index 91076af7..76e236d9 100644 --- a/packages/contracts/src/events.ts +++ b/packages/contracts/src/events.ts @@ -216,3 +216,25 @@ export function parseEventPayload(type: T, payload: unknown const r = schema.safeParse(payload); return r.success ? (r.data as EventPayload) : null; } + +/** + * 事件补读(JSON,给 API Key 调用者轮询;EventSource 带不了 Bearer)。 + * GET /events 带 limit 参数(或 Accept 只要 application/json)时返回这一页,否则是 SSE 流。 + */ +export const EventReplayQuery = z.object({ + after: z.coerce.number().int().nonnegative().default(0), + limit: z.coerce.number().int().min(1).max(1000).optional(), +}); +export type EventReplayQuery = z.infer; + +export const EventReplayPage = z.object({ + items: z.array(EventEnvelope), + /** 下次请求的 after(本页最后一条 seq;空页时等于请求的 after) */ + nextAfter: z.number().int().nonnegative(), + hasMore: z.boolean(), + /** 当前最大 seq */ + latestSeq: z.number().int().nonnegative(), + /** true:after 早于保留窗口(已裁剪),应重新拉取业务快照后从 latestSeq 续读 */ + gap: z.boolean(), +}); +export type EventReplayPage = z.infer; diff --git a/packages/contracts/src/execution.ts b/packages/contracts/src/execution.ts index 869e5164..792c755f 100644 --- a/packages/contracts/src/execution.ts +++ b/packages/contracts/src/execution.ts @@ -123,6 +123,31 @@ export const AcquisitionPlanCreate = AcquisitionPlan.omit({ ...SERVER_MANAGED, s statusReason: true, }); export type AcquisitionPlanCreate = z.infer; + +/** + * 手动提交下载(工作台下载入口,POST /acquisition-plans):与自动收录同一流程—— + * download.plan(evaluateAcquisition,硬条件/预算/阈值/Slot 忙)→ download.submit(预留 + submitDownload 任务)。 + * 目标 Slot:slotId 优先;否则 entityId(缺省取来源资源已识别的作品)下按创建顺序逐个尝试,第一个生成计划的 Slot 胜出。 + */ +export const ManualAcquisitionRequest = z.object({ + sourceReleaseId: Id, + slotId: Id.optional(), + entityId: Id.optional(), + note: z.string().max(2000).optional(), +}); +export type ManualAcquisitionRequest = z.infer; + +export const ManualAcquisitionResult = z.object({ + /** true = 计划已预留并入队 submitDownload */ + submitted: z.boolean(), + plan: AcquisitionPlan.nullable(), + slotId: Id.nullable(), + /** 业务任务 ID(submitDownload) */ + taskId: Id.nullable(), + /** 未生成计划时逐个 Slot 的拒绝原因 */ + rejected: z.array(z.object({ slotId: Id.nullable(), code: z.string(), message: z.string() })), +}); +export type ManualAcquisitionResult = z.infer; export const AcquisitionPlanUpdate = AcquisitionPlan.pick({ status: true, taskId: true, statusReason: true }).partial(); export type AcquisitionPlanUpdate = z.infer; diff --git a/packages/contracts/src/identity.ts b/packages/contracts/src/identity.ts index 73339310..6dd22815 100644 --- a/packages/contracts/src/identity.ts +++ b/packages/contracts/src/identity.ts @@ -34,6 +34,8 @@ export const ExternalIdentity = EntityBase.extend({ verifyStatus: IdentityVerifyStatus, /** 附属字段,如 TMDB episode_group */ extra: JsonObject, + /** 派生字段(列表返回):作品标题(挂在内容单元上时取所属作品) */ + mediaEntityTitle: z.string().nullable().optional(), }); export type ExternalIdentity = z.infer; diff --git a/packages/contracts/src/organization.ts b/packages/contracts/src/organization.ts index 42d265c2..14692e7b 100644 --- a/packages/contracts/src/organization.ts +++ b/packages/contracts/src/organization.ts @@ -201,6 +201,8 @@ export const PublicOrganizationRecord = z.object({ /** 稳定记录 ID */ id: Id, taskId: Id.nullable(), + /** 来源下载任务(手动整理本地路径时为 null) */ + downloadTaskId: Id.nullable(), /** 相关资产操作账本 ID */ operationIds: z.array(Id), media: z @@ -265,6 +267,7 @@ export const OrganizationListQuery = z.object({ status: OrganizationStatus.optional(), trigger: OrganizationTrigger.optional(), mediaEntityId: Id.optional(), + downloadTaskId: Id.optional(), q: z.string().max(200).optional(), }); export type OrganizationListQuery = z.infer; @@ -384,6 +387,24 @@ export const InventoryEntry = z.object({ downloading: z.array( z.object({ downloadTaskId: Id, name: z.string(), status: DownloadStatus, progress: z.number().min(0).max(1) }), ), + /** 有生效中的订阅(active) */ + subscribed: z.boolean(), + /** + * 候选理由:该作品各 Slot 对来源资源的最近一次评估(每个 (Slot, 资源) 一条,新→旧,最多 50 条)。 + * 工作台按 sourceReleaseId 取当前资源"为什么选 / 不选"。 + */ + candidateReasons: z.array( + z.object({ + slotId: Id, + slotName: z.string(), + sourceReleaseId: Id, + decision: CandidateEvaluation.shape.decision, + /** 拒绝原因 message + 评估说明,已合并去重 */ + reasons: z.array(z.string()), + planId: Id.nullable(), + evaluatedAt: Timestamp, + }), + ), }); export type InventoryEntry = z.infer; diff --git a/packages/contracts/src/policy.ts b/packages/contracts/src/policy.ts index a55753d1..2a585e02 100644 --- a/packages/contracts/src/policy.ts +++ b/packages/contracts/src/policy.ts @@ -72,6 +72,8 @@ export const Subscription = EntityBase.extend({ statusReason: z.string().nullable(), lastCheckedAt: Timestamp.nullable(), createdBy: z.string().nullable(), + /** 派生字段(列表返回):作品标题,未建档时为 null */ + mediaEntityTitle: z.string().nullable().optional(), }); export type Subscription = z.infer; export const SubscriptionCreate = Subscription.pick({ @@ -145,6 +147,18 @@ export const AcquisitionSlot = EntityBase.extend({ lastEvaluatedAt: Timestamp.nullable(), nextEvaluateAt: Timestamp.nullable(), revision: z.number().int().min(1), + /** 派生字段(列表返回):当前资产的交付特征摘要,无资产为 null */ + currentAsset: z + .object({ + id: Id, + resolution: z.string().nullable(), + sourceType: z.string().nullable(), + releaseGroup: z.string().nullable(), + sizeBytes: z.number().int().nullable(), + status: z.string(), + }) + .nullable() + .optional(), }); export type AcquisitionSlot = z.infer; export const AcquisitionSlotCreate = AcquisitionSlot.pick({ @@ -206,6 +220,55 @@ export const PreferenceTag = EntityBase.extend({ policyRevisionId: Id.nullable(), }); export type PreferenceTag = z.infer; + +/** 偏好标签修订历史:取自 audit_event(target_type=preference_tag),新→旧 */ +export const PreferenceTagRevision = z.object({ + /** audit_event id */ + id: Id, + at: Timestamp, + action: z.enum(["created", "updated", "deleted"]), + actorType: z.string(), + actorId: z.string().nullable(), + /** 该次写入后的 revision */ + revision: z.number().int().min(1).nullable(), + /** 本次改动的字段(created/deleted 为空) */ + changedFields: z.array(z.string()), + /** 写入后的快照 */ + snapshot: z.object({ + label: z.string(), + kind: z.string(), + status: z.string(), + scope: JsonObject, + rule: JsonObject, + description: z.string().nullable(), + }), +}); +export type PreferenceTagRevision = z.infer; +export const PreferenceTagRevisionList = z.object({ items: z.array(PreferenceTagRevision) }); +export type PreferenceTagRevisionList = z.infer; + +/** 策略预览结果(policy_revision.preview):最近 200 条来源资源按模板回放 */ +export const PolicyPreviewSummary = z.object({ + evaluatedAt: z.string(), + sampleSize: z.number(), + templates: z.number(), + matched: z.number(), + excluded: z.number(), + samples: z.array( + z.object({ + sourceReleaseId: z.string(), + /** 资源标题(W4 起提供) */ + title: z.string().optional(), + sizeBytes: z.number().nullable().optional(), + template: z.string(), + decision: z.string(), + /** 拒绝/接受理由摘要 */ + reasons: z.array(z.string()).optional(), + }), + ), + valid: z.boolean(), +}); +export type PolicyPreviewSummary = z.infer; /** fingerprint 由服务端计算 */ export const PreferenceTagCreate = PreferenceTag.pick({ label: true, diff --git a/packages/contracts/src/sources.ts b/packages/contracts/src/sources.ts index e2a7838a..49a8dab6 100644 --- a/packages/contracts/src/sources.ts +++ b/packages/contracts/src/sources.ts @@ -105,6 +105,13 @@ export const FeedSource = EntityBase.extend({ backoffUntil: Timestamp.nullable(), healthStatus: FeedHealth, lastError: JsonObject.nullable(), + /** + * 派生字段(非表列):最近一次抓取发现与上次窗口零重叠 → 疑似漏项,已排站点补漏。 + * 下一次抓取窗口重新衔接后清空。站点未登记的 Feed 无法补漏(backfillQueued=false)。 + */ + possibleGap: z + .object({ detectedAt: Timestamp, previousFetchedAt: Timestamp.nullable(), backfillQueued: z.boolean() }) + .nullable(), }); export type FeedSource = z.infer; /** url 为只写明文(可能含 passkey),服务端加密存为 secret 并生成 urlDisplay */ diff --git a/packages/contracts/test/db-sync.test.ts b/packages/contracts/test/db-sync.test.ts index 40b13b3a..25547c3c 100644 --- a/packages/contracts/test/db-sync.test.ts +++ b/packages/contracts/test/db-sync.test.ts @@ -38,7 +38,7 @@ type Obj = z.ZodObject; /** 表(PascalCase)→ Row schema 与允许的差异 */ const ROWS: Record = { AcquisitionPlan: { row: C.AcquisitionPlan }, - AcquisitionSlot: { row: C.AcquisitionSlot }, + AcquisitionSlot: { row: C.AcquisitionSlot, apiOnly: ["currentAsset"] }, AgentRun: { row: C.AgentRun }, AgentRunEvent: { row: C.AgentRunEventRow }, AgentRunMessage: { row: C.AgentRunMessage }, @@ -55,8 +55,8 @@ const ROWS: Record Decision: { row: C.Decision }, DownloadTask: { row: C.DownloadTask }, DownloaderInstance: { row: C.DownloaderInstance }, - ExternalIdentity: { row: C.ExternalIdentity }, - FeedSource: { row: C.FeedSource }, + ExternalIdentity: { row: C.ExternalIdentity, apiOnly: ["mediaEntityTitle"] }, + FeedSource: { row: C.FeedSource, apiOnly: ["possibleGap"] }, IdentityRevision: { row: C.IdentityRevision }, LibraryRoot: { row: C.LibraryRoot }, MediaAsset: { row: C.MediaAsset }, @@ -78,9 +78,9 @@ const ROWS: Record Site: { row: C.Site }, SiteAccount: { row: C.SiteAccount }, SourceRelease: { row: C.SourceRelease }, - Subscription: { row: C.Subscription }, + Subscription: { row: C.Subscription, apiOnly: ["mediaEntityTitle"] }, Task: { row: C.Task }, - TitleAlias: { row: C.TitleAlias }, + TitleAlias: { row: C.TitleAlias, apiOnly: ["mediaEntityTitle"] }, // W2-B 来源/身份/元数据(0018–0020) FieldOverride: { row: C.FieldOverride }, IdentityCandidate: { row: C.IdentityCandidate }, diff --git a/packages/db/migrations/0022_candidate_dedupe.ts b/packages/db/migrations/0022_candidate_dedupe.ts new file mode 100644 index 00000000..7352ca5a --- /dev/null +++ b/packages/db/migrations/0022_candidate_dedupe.ts @@ -0,0 +1,29 @@ +import type { Kysely } from "kysely"; +import { execDdl } from "../src/migration-helpers.ts"; + +// W4-A 候选评估记录去重:同一 (acquisition_slot, source_release) 只保留最新一条(artifact title=candidate_evaluation)。 +// 先清掉历史重复(保留 updated_at 最新的一条),再加唯一表达式索引,写入端按该键 upsert。 + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl( + db, + ` +delete from artifact a + using artifact b + where a.title = 'candidate_evaluation' and b.title = 'candidate_evaluation' + and a.subject_type = 'acquisition_slot' and b.subject_type = 'acquisition_slot' + and a.subject_id = b.subject_id + and (a.content->>'sourceReleaseId') = (b.content->>'sourceReleaseId') + and (a.updated_at, a.id) < (b.updated_at, b.id); +create unique index artifact_candidate_slot_release + on artifact (subject_id, (content->>'sourceReleaseId')) + where title = 'candidate_evaluation' and subject_type = 'acquisition_slot'; +`, + ); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl(db, "drop index if exists artifact_candidate_slot_release;"); +}