Merge branch 'ia2'

Co-Authored-By: Claude Fable 5.1 <[email protected]>

# Conflicts:
#	CHANGELOG.md
#	docs/plan/00-first-version-plan.md
#	packages/agent/src/system-prompt.ts
This commit is contained in:
root committed 2026-09-30 00:30:42 +08:00
commit 7a24d23bbf
46 files changed
+1845 -1503

No files matched your search

+11 -11
View File
@@ -1,6 +1,6 @@
import type { SearchAction } from "@umt/contracts";
// 页面与操作跳转(命令面板静态清单),按 IA-3 定稿导航:概览 / 资源 / 订阅 / 整理 / 系统(定时任务在任务页"计划"Tab)。
// 页面与操作跳转(命令面板静态清单),按 IA-2 导航:概览 / 获取 / 整理 / 系统(任务归系统;RSS 在站点详情里)。
// href 是控制台内路径;action 类带查询参数,由前端页面识别后打开对应表单。
const page = (id: string, group: string, title: string, href: string, keywords: string[] = []): SearchAction => ({
@@ -22,28 +22,28 @@ const action = (id: string, group: string, title: string, href: string, keywords
export const SEARCH_ACTIONS: readonly SearchAction[] = [
page("home", "概览", "首页", "/console", ["home", "dashboard", "概览"]),
page("tasks", "概览", "任务", "/console/tasks", ["task", "运行", "历史"]),
page("agent", "概览", "智能体会话", "/console/agent", ["agent", "会话"]),
page("inbox", "概览", "待处理", "/console/inbox", ["inbox", "收件箱", "decision"]),
page("sites", "资源", "站点", "/console/sites", ["site", "pt", "站点账号"]),
page("sources", "资源", "RSS 订阅", "/console/sources", ["rss", "feed"]),
page("subscriptions", "订阅", "订阅", "/console/subscriptions", ["subscription", "追剧"]),
page("rules", "订阅", "规则", "/console/rules", ["rules", "收录策略", "policy"]),
page("schedules", "概览", "计划", "/console/schedules", ["schedule", "cron", "巡检", "计划", "定时任务"]),
page("sites", "获取", "站点", "/console/sites", ["site", "pt", "站点账号", "rss", "feed", "bt"]),
page("subscriptions", "获取", "订阅", "/console/subscriptions", ["subscription", "追剧"]),
page("rules", "获取", "规则", "/console/rules", ["rules", "收录策略", "policy"]),
page("downloads", "整理", "下载面板", "/console/downloads", ["download", "qb", "下载任务"]),
page("download-plans", "整理", "下载计划", "/console/downloads/plans", ["plan", "计划"]),
page("organize", "整理", "整理记录", "/console/organize", ["organize", "入库", "历史"]),
page("files", "整理", "文件管理", "/console/files", ["files", "文件"]),
page("trash", "整理", "回收站", "/console/files/trash", ["trash", "删除"]),
page("library", "整理", "媒体库", "/console/library", ["library", "媒体库", "作品"]),
page("tasks", "系统", "任务", "/console/tasks", ["task", "运行", "历史"]),
page("settings", "系统", "设置", "/console/settings", ["settings", "配置", "tmdb", "下载器"]),
page("schedules", "概览", "定时任务", "/console/tasks?tab=plan", ["schedule", "cron", "巡检", "计划"]),
page("audit", "系统", "审计", "/console/audit", ["audit", "日志"]),
page("preferences", "系统", "偏好", "/console/preferences", ["preferences", "主题"]),
action("new-subscription", "订阅", "新建订阅", "/console/subscriptions?new=1", ["subscribe", "订阅"]),
action("new-subscription", "获取", "新建订阅", "/console/subscriptions?new=1", ["subscribe", "订阅"]),
action("new-agent-run", "概览", "新建智能体会话", "/console/agent?new=1", ["agent", "委托"]),
action("add-site", "资源", "添加站点", "/console/sites?new=1", ["site", "站点"]),
action("add-feed", "资源", "添加 RSS 订阅", "/console/sources?new=1", ["rss", "feed"]),
action("new-schedule", "概览", "新建巡检", "/console/tasks?tab=plan&new=1", ["巡检", "agent_check"]),
action("add-site", "获取", "添加站点", "/console/sites?new=1", ["site", "站点", "rss", "feed", "bt"]),
// RSS 属于站点(IA-2):添加 RSS 源 = 添加「自定义 RSS 源」类型的站点
action("add-feed", "获取", "添加自定义 RSS 源", "/console/sites?new=rss_feed", ["rss", "feed"]),
action("new-schedule", "概览", "新建巡检", "/console/schedules?new=1", ["巡检", "agent_check"]),
];
/** 标题 / 关键词 / 路径包含查询词(不区分大小写);标题前缀命中排前 */
+4 -2
View File
@@ -14,6 +14,7 @@ import { type SiteRow, sql } from "@umt/db";
import { applyFacets, type FacetInput, normalizeTitle, parseFacetParam, parseRelease } from "@umt/domain";
import type { SourcesRuntime } from "../sources-identity/runtime.ts";
import { upsertSiteRelease } from "../sources-identity/service/snapshot.ts";
import { siteCan } from "../sources-identity/site-kinds.ts";
import { raceAbort } from "./media.ts";
// 跨站资源搜索:选中的已启用站点逐站调适配器 search(每站共享限流器 + 单站超时 + 站点级并发上限);
@@ -63,15 +64,16 @@ export class ReleaseFanout {
private readonly opts: FanoutOptions,
) {}
/** 参与搜索的候选站点:启用且未删除(按名称排序) */
/** 参与搜索的候选站点:启用、未删除且支持搜索(IA-2 按站点类型与定义的获取方式;按名称排序) */
async enabledSites(): Promise<SiteRow[]> {
return this.rt()
const rows = await this.rt()
.ctx.db.selectFrom("site")
.selectAll()
.where("enabled", "=", true)
.where("deleted_at", "is", null)
.orderBy("name")
.execute();
return rows.filter((s) => siteCan(s, "search"));
}
/**
@@ -3,6 +3,7 @@ import { capabilitySpec, type SiteAccountUserData, type SiteLastHealth } from "@
import type { SiteAccountRow, SiteRow } from "@umt/db";
import type { SourcesRuntime } from "./runtime.ts";
import { checkSiteHealth, overallAccountStatus, refreshSiteAccount, userDataHistory } from "./service/account.ts";
import { capabilitiesOf, requireCapability } from "./site-kinds.ts";
// W10-pre2 站点能力:site.list / site.health_check / site.refresh_login / site.user_data,处理器调 W9-A 站点中心的 service。
// 脱敏:输出按白名单逐字段拼装——不含 secret 引用、Cookie、密码、登录名、UA 与站点 config。
@@ -62,6 +63,8 @@ export function registerSiteCapabilities(registry: CapabilityRegistry, rt: () =>
id: s.id,
key: s.key,
name: s.name,
kind: s.kind,
capabilities: capabilitiesOf(s),
adapter: s.adapter,
baseUrl: s.base_url,
enabled: s.enabled,
@@ -83,6 +86,7 @@ export function registerSiteCapabilities(registry: CapabilityRegistry, rt: () =>
registry.register(capabilitySpec("site.refresh_login"), async (input, ctx) => {
const r = rt();
const s = await liveSite(r, input.siteId);
requireCapability(s, "login");
let q = r.ctx.db.selectFrom("site_account").select(["id"]).where("site_id", "=", s.id);
q = input.accountId
? q.where("id", "=", input.accountId)
@@ -10,6 +10,7 @@ import {
OkResult,
ReleaseIdentityAssign,
type SecretPurpose,
SITE_KINDS,
Site,
type SiteAccount,
type SiteAdapter,
@@ -25,7 +26,7 @@ import {
SiteUserDataQuery,
type SourceRelease,
} from "@umt/contracts";
import { sql } from "@umt/db";
import { type FeedSourceRow, sql } from "@umt/db";
import { z } from "zod";
import { ApiException } from "../../../http/errors.ts";
import type { ModuleApp } from "../../types.ts";
@@ -44,9 +45,10 @@ import {
setSiteCredentials,
userDataHistory,
} from "../service/account.ts";
import { audit, closeDecisions } from "../service/common.ts";
import { audit, closeDecisions, type Exec } from "../service/common.ts";
import { feedDefaultsForSite } from "../service/feeds.ts";
import { assignReleaseIdentity } from "../service/identity-admin.ts";
import { ensureRssSite, requireCapability, siteToApi, uniqueSiteKey } from "../site-kinds.ts";
// 站点、站点账号、Feed、来源资源、观察的管理 API。凭证只写(存 secret,对外只给 secret_ref / 脱敏 URL)。
@@ -88,6 +90,7 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
items: listSiteDefinitions().map((d) => ({
key: d.id,
name: d.umt.kind === "nexusphp" ? (d.umt.displayName ?? d.name) : d.umt.displayName,
kind: "pt" as const,
adapter: adapterOf(d),
baseUrl: d.domain,
shippedEnabled: d.enabled !== false,
@@ -113,7 +116,7 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
.limit(query.limit + 1)
.execute();
const p = page(rows, off, query.limit);
const items = toApiList<Record<string, unknown>>(p.items);
const items = p.items.map((r) => siteToApi<Record<string, unknown>>(r));
if (query.include === "accounts" && p.items.length) {
const accounts = await db()
.selectFrom("site_account")
@@ -133,37 +136,70 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
}
return { items, nextCursor: p.nextCursor };
});
route(app, D.sites.get, async ({ params }) => toApi(await getSite(params.id)));
route(app, D.sites.get, async ({ params }) => siteToApi(await getSite(params.id)));
route(app, D.sites.create, async ({ body, actor }) => {
const def = listSiteDefinitions().find((d) => d.id === body.key);
if (def && adapterOf(def) !== body.adapter) {
throw new ApiException("bad_request", `站点 ${body.key} 的适配器类型应为 ${adapterOf(def)}`);
}
if (!def && body.adapter !== "generic_rss") {
throw new ApiException("bad_request", `注册表中没有站点 ${body.key};未适配站点只能以 generic_rss 登记`);
const def = body.key ? listSiteDefinitions().find((d) => d.id === body.key) : undefined;
// 类型缺省:注册表里有定义 = PT 站;否则按旧口径(generic_rss = 自定义 RSS 站点)
const kind = body.kind ?? (def ? "pt" : "rss_feed");
if (!SITE_KINDS[kind].implemented) throw new ApiException("bad_request", `还不支持添加${SITE_KINDS[kind].label}`);
if (kind === "pt") {
if (!body.key) throw new ApiException("bad_request", "PT 站要从站点定义里选(缺 key)");
if (def && adapterOf(def) !== body.adapter) {
throw new ApiException("bad_request", `站点 ${body.key} 的适配器类型应为 ${adapterOf(def)}`);
}
if (!def && body.adapter !== "generic_rss") {
throw new ApiException("bad_request", `注册表中没有站点 ${body.key};未适配站点只能以 generic_rss 登记`);
}
} else if (body.adapter !== "generic_rss") {
throw new ApiException("bad_request", `${SITE_KINDS[kind].label}只能以 generic_rss 登记`);
}
const key = body.key ?? (await uniqueSiteKey(db(), body.rssUrl ?? body.baseUrl));
try {
const row = await db()
.insertInto("site")
.values({
key: body.key,
name: body.name,
adapter: body.adapter,
base_url: body.baseUrl,
enabled: body.enabled ?? true,
rate_limit: jsonb(body.rateLimit ?? {}),
config: jsonb(body.config ?? {}),
snapshot_enabled: body.snapshotEnabled ?? false,
snapshot_interval_sec: body.snapshotIntervalSec ?? null,
proxy: body.proxy ?? false,
search_default: body.searchDefault ?? true,
})
.returningAll()
.executeTakeFirstOrThrow();
await audit(db(), actor, "site.created", { type: "site", id: row.id }, { key: body.key });
return toApi(row);
.transaction()
.execute(async (trx) => {
const row = await trx
.insertInto("site")
.values({
key,
name: body.name,
kind,
adapter: body.adapter,
base_url: body.baseUrl,
enabled: body.enabled ?? true,
rate_limit: jsonb(body.rateLimit ?? {}),
config: jsonb(body.config ?? {}),
snapshot_enabled: kind === "pt" ? (body.snapshotEnabled ?? false) : false,
snapshot_interval_sec: body.snapshotIntervalSec ?? null,
proxy: body.proxy ?? false,
// 不能搜索的站点不进默认搜索集合
search_default: kind === "pt" ? (body.searchDefault ?? true) : false,
})
.returningAll()
.executeTakeFirstOrThrow();
if (body.rssUrl) {
const inferred = await feedDefaultsForSite(trx, row, body.rssUrl, null);
const feed = await insertFeed(trx, {
siteId: row.id,
url: body.rssUrl,
name: inferred.name,
parserVersion: inferred.parserVersion,
pollIntervalSec: inferred.pollIntervalSec,
});
await audit(
trx,
actor,
"feed_source.created",
{ type: "feed_source", id: feed.id },
{ url: feed.url_display },
);
}
return row;
});
await audit(db(), actor, "site.created", { type: "site", id: row.id }, { key, kind });
return siteToApi(row);
} catch (e) {
if ((e as { code?: string }).code === "23505") throw new ApiException("conflict", `站点 ${body.key} 已登记`);
if ((e as { code?: string }).code === "23505") throw new ApiException("conflict", `站点 ${key} 已登记`);
throw e;
}
});
@@ -184,9 +220,9 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
await db().updateTable("site").set(patch).where("id", "=", params.id).execute();
await audit(db(), actor, "site.updated", { type: "site", id: params.id }, { fields: Object.keys(patch) });
}
return toApi(await setSiteEnabled(params.id, body.enabled, actor));
return siteToApi(await setSiteEnabled(params.id, body.enabled, actor));
}
if (!Object.keys(patch).length) return toApi(await getSite(params.id));
if (!Object.keys(patch).length) return siteToApi(await getSite(params.id));
const row = await db()
.updateTable("site")
.set(patch)
@@ -194,7 +230,7 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
.returningAll()
.executeTakeFirstOrThrow();
await audit(db(), actor, "site.updated", { type: "site", id: params.id }, { fields: Object.keys(patch) });
return toApi(row);
return siteToApi(row);
});
for (const enabled of [true, false]) {
route(
@@ -208,7 +244,7 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
params: IdParam,
response: Site,
},
async ({ params, actor }) => toApi(await setSiteEnabled(params.id, enabled, actor)),
async ({ params, actor }) => siteToApi(await setSiteEnabled(params.id, enabled, actor)),
);
}
route(
@@ -225,6 +261,7 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
},
async ({ params, body }) => {
const s = await getSite(params.id);
requireCapability(s, "list");
if (!s.enabled) throw new ApiException("precondition_failed", "站点已关闭");
const r = await rt().ctx.engine.enqueue(
"snapshotSite",
@@ -349,6 +386,7 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
},
async ({ params, body, actor }) => {
const s = await getSite(params.id);
requireCapability(s, "login");
let q = db().selectFrom("site_account").select(["id", "enabled", "auth_kind"]).where("site_id", "=", s.id);
q = body.accountId
? q.where("id", "=", body.accountId)
@@ -409,7 +447,11 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
body: SiteCredentialsPut,
response: SiteCredentialsResult,
},
async ({ params, body, actor }) => setSiteCredentials(rt(), await getSite(params.id), body, actor),
async ({ params, body, actor }) => {
const s = await getSite(params.id);
requireCapability(s, "login");
return setSiteCredentials(rt(), s, body, actor);
},
);
route(
app,
@@ -562,37 +604,33 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
});
route(app, D.feedSources.get, async ({ params }) => feedToApi(await getFeed(params.id)));
route(app, D.feedSources.create, async ({ body, actor }) => {
const site = body.siteId ? await getSite(body.siteId) : null;
if (body.siteAccountId) await checkAccount(body.siteAccountId, body.siteId ?? null);
// RSS 属于站点(IA-2):没给站点时归入同主机的「自定义 RSS 源」站点,没有就建一个
const site = body.siteId ? await getSite(body.siteId) : await ensureRssSite(db(), body.url);
requireCapability(site, "rss");
if (body.siteAccountId) await checkAccount(body.siteAccountId, site.id);
// 站点内添加:账号 / 解析器 / 名称按站点推断,域名不符只提示
const inferred = site ? await feedDefaultsForSite(db(), site, body.url, body.siteAccountId ?? null) : null;
const name = body.name?.trim() || inferred?.name || new URL(body.url).hostname;
const ref = await rt().ctx.secrets.create({ purpose: "feed_url", plaintext: body.url, label: `feed:${name}` });
const row = await db()
.insertInto("feed_source")
.values({
site_id: body.siteId ?? null,
site_account_id: inferred?.siteAccountId ?? body.siteAccountId ?? null,
name,
...(inferred ? { parser_version: inferred.parserVersion } : {}),
purpose: body.purpose ?? "generic",
kind: body.kind ?? "rss",
url_secret_ref: ref,
url_display: redactUrl(body.url),
enabled: body.enabled ?? true,
poll_interval_sec: body.pollIntervalSec ?? inferred?.pollIntervalSec ?? 900,
filter: jsonb(body.filter ?? {}),
})
.returningAll()
.executeTakeFirstOrThrow();
const inferred = await feedDefaultsForSite(db(), site, body.url, body.siteAccountId ?? null);
const row = await insertFeed(db(), {
siteId: site.id,
url: body.url,
name: body.name?.trim() || inferred.name,
siteAccountId: inferred.siteAccountId,
parserVersion: inferred.parserVersion,
purpose: body.purpose,
kind: body.kind,
enabled: body.enabled,
pollIntervalSec: body.pollIntervalSec ?? inferred.pollIntervalSec,
filter: body.filter,
});
await audit(db(), actor, "feed_source.created", { type: "feed_source", id: row.id }, { url: row.url_display });
return inferred?.warnings.length ? { ...feedToApi<FeedSource>(row), warnings: inferred.warnings } : feedToApi(row);
return inferred.warnings.length ? { ...feedToApi<FeedSource>(row), warnings: inferred.warnings } : feedToApi(row);
});
route(app, D.feedSources.update, async ({ params, body, actor }) => {
const feed = await getFeed(params.id);
const patch: Record<string, unknown> = {};
if (body.siteId !== undefined) {
if (body.siteId) await getSite(body.siteId);
if (!body.siteId) throw new ApiException("bad_request", "RSS 必须属于一个站点");
requireCapability(await getSite(body.siteId), "rss");
patch.site_id = body.siteId;
}
if (body.siteAccountId !== undefined) {
@@ -831,6 +869,45 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
if (!row) throw new ApiException("not_found", "站点不存在");
return row;
}
async function insertFeed(
exec: Exec,
f: {
siteId: string;
url: string;
name: string;
pollIntervalSec: number;
siteAccountId?: string | null;
parserVersion?: string;
purpose?: FeedSourceRow["purpose"] | undefined;
kind?: FeedSourceRow["kind"] | undefined;
enabled?: boolean | undefined;
filter?: Record<string, unknown> | undefined;
},
) {
// secret.name 唯一:同站多个 RSS 都叫「站点名 RSS」时不能撞名
const ref = await rt().ctx.secrets.create({
purpose: "feed_url",
plaintext: f.url,
label: `feed:${f.name}:${randomUUID().slice(0, 8)}`,
});
return exec
.insertInto("feed_source")
.values({
site_id: f.siteId,
site_account_id: f.siteAccountId ?? null,
name: f.name,
...(f.parserVersion ? { parser_version: f.parserVersion } : {}),
purpose: f.purpose ?? "generic",
kind: f.kind ?? "rss",
url_secret_ref: ref,
url_display: redactUrl(f.url),
enabled: f.enabled ?? true,
poll_interval_sec: f.pollIntervalSec,
filter: jsonb(f.filter ?? {}),
})
.returningAll()
.executeTakeFirstOrThrow();
}
async function getFeed(id: string) {
const row = await db().selectFrom("feed_source").selectAll().where("id", "=", id).executeTakeFirst();
if (!row) throw new ApiException("not_found", "Feed 不存在");
@@ -3,6 +3,7 @@ import { getSiteDefinition } from "@umt/adapters";
import { sql } from "@umt/db";
import { type SourcesRuntime, siteAccessFor } from "./runtime.ts";
import { archiveStalePool } from "./service/identity-decisions.ts";
import { RSS_KINDS, siteCan } from "./site-kinds.ts";
// worker 后台调度:按 Feed 轮询策略发 feed.due(outbox → ingestFeed)、按站点快照周期入队 snapshotSite、
// 按待识别池 next_review_at 入队复查。幂等键按时间槽生成,多 worker 并发也不会重复。
@@ -39,7 +40,7 @@ export async function tickFeeds(rt: SourcesRuntime): Promise<number> {
const feedsDue = await ctx.db.transaction().execute(async (trx) => {
const due = await sql<{ id: string; poll_interval_sec: number }>`
select f.id, f.poll_interval_sec from feed_source f left join site s on s.id = f.site_id
where f.enabled and (s.id is null or s.enabled)
where f.enabled and (s.id is null or (s.enabled and s.deleted_at is null and s.kind = any(${RSS_KINDS})))
and (f.next_fetch_at is null or f.next_fetch_at <= ${now})
and (f.backoff_until is null or f.backoff_until <= ${now})
and f.health_status <> 'circuit_open'
@@ -72,13 +73,15 @@ export async function tickSnapshots(rt: SourcesRuntime): Promise<number> {
const defaults = await rt.siteDefaults();
const rows = await ctx.db
.selectFrom("site")
.select(["id", "key", "snapshot_interval_sec", "last_snapshot_at"])
.select(["id", "key", "kind", "adapter", "snapshot_interval_sec", "last_snapshot_at"])
.where("enabled", "=", true)
.where("snapshot_enabled", "=", true)
.where("deleted_at", "is", null)
.execute();
const due: Array<{ id: string; interval: number }> = [];
for (const r of rows) {
// 只给支持列表抓取的站点排快照(IA-2:按站点类型与定义的获取方式)
if (!siteCan(r, "list")) continue;
const interval =
r.snapshot_interval_sec ?? siteAccessFor(getSiteDefinition(r.key) ?? null, defaults).snapshotIntervalSec;
if (!r.last_snapshot_at || r.last_snapshot_at.getTime() + interval * 1000 <= now.getTime()) {
@@ -26,6 +26,7 @@ import { type SiteAccountRow, type SiteRow, sql } from "@umt/db";
import { ApiException } from "../../../http/errors.ts";
import { toApi } from "../rows.ts";
import type { SourcesRuntime } from "../runtime.ts";
import { capabilitiesOf, LOGIN_KINDS } from "../site-kinds.ts";
import { type Actor, audit, closeDecisions, type Exec, openDecision, SYSTEM } from "./common.ts";
import { cookieFromBrowser, writeBrowserCookie } from "./credentials.ts";
@@ -929,6 +930,7 @@ export async function tickSiteUserData(rt: SourcesRuntime, slot: string): Promis
.select(["a.id"])
.where("s.enabled", "=", true)
.where("s.deleted_at", "is", null)
.where("s.kind", "in", LOGIN_KINDS)
.where("a.enabled", "=", true)
.where("a.status", "=", "active")
.where("a.auth_kind", "!=", "passkey")
@@ -950,11 +952,12 @@ export async function tickSiteHealth(rt: SourcesRuntime): Promise<number> {
.selectAll()
.where("enabled", "=", true)
.where("deleted_at", "is", null)
.where("adapter", "!=", "generic_rss")
.execute();
let n = 0;
const now = rt.now().getTime();
for (const s of sites) {
// 没有站点适配器的(自定义 RSS 源、未适配站点)没有首页检测;只剩 RSS 的站点跳过
if (!capabilitiesOf(s).some((c) => c !== "rss")) continue;
// 站点访问默认的最短间隔内检测过(手动或定时)就跳过;留 2 分钟余量,免得和定时任务周期错开一轮
const { healthCheckIntervalSec } = await rt.siteAccess(s.key);
if (s.last_health_at && now - s.last_health_at.getTime() < healthCheckIntervalSec * 1000 - 120_000) continue;
@@ -13,6 +13,7 @@ import type { FeedHealth } from "@umt/contracts";
import { type FeedSourceRow, type SiteRow, sql } from "@umt/db";
import { jsonb } from "../rows.ts";
import type { SourcesRuntime } from "../runtime.ts";
import { siteCan } from "../site-kinds.ts";
import { clip, type Exec, openDecision } from "./common.ts";
// ingestFeed:RSS 抓取 → observation(feed+guid 唯一,摘要变化 = 更新)→ source_release(site+站点资源 ID 聚合)。
@@ -154,6 +155,7 @@ export async function ingestFeed(
? await db.selectFrom("site").selectAll().where("id", "=", feed.site_id).executeTakeFirst()
: undefined;
if (site && !site.enabled) return { status: "skipped", reason: "site_disabled", ...empty };
if (site && !siteCan(site, "rss")) return { status: "skipped", reason: "unsupported", ...empty };
if (!feed.url_secret_ref) return { status: "skipped", reason: "no_url", ...empty };
const url = await secrets.resolve(feed.url_secret_ref);
@@ -216,7 +218,8 @@ export async function ingestFeed(
? {
detectedAt: now.toISOString(),
previousFetchedAt: cursor.window?.lastFetchedAt ?? null,
backfillQueued: !!site?.id,
// 补漏靠列表抓取:站点支持 list 才会排补漏(自定义 RSS 源没有列表)
backfillQueued: !!site && siteCan(site, "list"),
}
: undefined
: cursor.gap;
@@ -6,6 +6,7 @@ import { parseRelease } from "@umt/domain";
import { siteReleaseToSourceRelease } from "@umt/mapping";
import { jsonb } from "../rows.ts";
import type { SourcesRuntime } from "../runtime.ts";
import { siteCan } from "../site-kinds.ts";
import { enqueueRelogin } from "./account.ts";
import { type Exec, openDecision } from "./common.ts";
import { hasDetailPage, saveDetailPage } from "./detail-page.ts";
@@ -134,6 +135,10 @@ export async function snapshotSite(rt: SourcesRuntime, input: SnapshotInput): Pr
const site = await db.selectFrom("site").selectAll().where("id", "=", input.siteId).executeTakeFirst();
if (!site) return { status: "skipped", reason: "not_found", ...none };
if (!site.enabled) return { status: "skipped", reason: "site_disabled", ...none };
// 按站点类型的获取方式(IA-2):实时搜索要 search,列表快照 / 补漏要 list
if (!siteCan(site, input.kind === "live_search" ? "search" : "list")) {
return { status: "skipped", reason: "unsupported", ...none };
}
if (site.health_status === "failing" && input.kind !== "manual" && input.kind !== "live_search") {
const invalid = await db
.selectFrom("site_account")
@@ -0,0 +1,110 @@
import { getSiteDefinition } from "@umt/adapters";
import { SITE_KINDS, type SiteCapability, type SiteKind, siteCapabilities } from "@umt/contracts";
import type { SiteRow } from "@umt/db";
import { ApiException } from "../../http/errors.ts";
import { toApi } from "./rows.ts";
import type { Exec } from "./service/common.ts";
// IA-2 站点类型:服务端统一按 capabilities 判断能不能搜索 / 抓列表 / 抓 RSS / 看详情 / 登录 / 取账号数据。
// 实际可用 = 类型允许的 ∩ 站点定义实现的(contracts siteCapabilities)。generic_rss 站点不查定义。
type SiteLike = Pick<SiteRow, "key" | "adapter"> & { kind: SiteKind };
export function capabilitiesOf(site: SiteLike): SiteCapability[] {
const def = site.adapter === "generic_rss" ? null : getSiteDefinition(site.key);
return siteCapabilities(site.kind, def ? def.umt.capabilities : null);
}
export function siteCan(site: SiteLike, cap: SiteCapability): boolean {
return capabilitiesOf(site).includes(cap);
}
const CAP_TEXT: Record<SiteCapability, string> = {
search: "搜索",
list: "列表抓取",
rss: "RSS",
detail: "详情页",
login: "登录",
userinfo: "账号数据",
};
/** 站点不支持该获取方式时抛 400:「名称」是自定义 RSS 源,不支持登录 */
export function requireCapability(site: SiteLike & Pick<SiteRow, "name">, cap: SiteCapability): void {
if (siteCan(site, cap)) return;
throw new ApiException("bad_request", `「${site.name}」是${SITE_KINDS[site.kind].label},不支持${CAP_TEXT[cap]}`);
}
/** site 行 → Site(补派生字段 capabilities) */
export function siteToApi<T>(row: SiteLike & object): T {
return { ...toApi<Record<string, unknown>>(row), capabilities: capabilitiesOf(row) } as T;
}
/** 有登录类能力的类型(健康检测、刷新登录、账号数据的定时任务按这个过滤) */
export const LOGIN_KINDS: SiteKind[] = (Object.keys(SITE_KINDS) as SiteKind[]).filter(
(k) => SITE_KINDS[k].implemented && SITE_KINDS[k].capabilities.includes("login"),
);
/** 能抓 RSS 的类型(Feed 调度按这个过滤) */
export const RSS_KINDS: SiteKind[] = (Object.keys(SITE_KINDS) as SiteKind[]).filter(
(k) => SITE_KINDS[k].implemented && SITE_KINDS[k].capabilities.includes("rss"),
);
/** 主机名 → key 片段(与迁移 0055 一致):小写,非字母数字换成 -,去掉 www. */
export function hostSlug(host: string): string {
const s = host
.toLowerCase()
.replace(/^www\./, "")
.replace(/[^a-z0-9]+/g, "-")
.replace(/^-+|-+$/g, "")
.slice(0, 40)
.replace(/-+$/, "");
return s || "feed";
}
function hostOfUrl(url: string): string | null {
try {
return new URL(url).host || null;
} catch {
return null;
}
}
/** 非 PT 站的 key:rss-<主机>,已占用(含已删除的站点)时加序号 */
export async function uniqueSiteKey(db: Exec, url: string): Promise<string> {
const base = `rss-${hostSlug(hostOfUrl(url) ?? "")}`;
const taken = new Set(
(await db.selectFrom("site").select("key").where("key", "like", `${base}%`).execute()).map((r) => r.key),
);
let key = base;
for (let n = 2; taken.has(key); n++) key = `${base}-${n}`;
return key;
}
/**
* 不带站点新建 RSS(旧 API 调用方):归入同主机的「自定义 RSS 源」站点,没有就建一个。
* 与迁移 0055 同口径:名称 = 主机名,地址 = https://<主机>,不参与默认搜索。
*/
export async function ensureRssSite(db: Exec, url: string): Promise<SiteRow> {
const host = hostOfUrl(url);
if (!host) throw new ApiException("bad_request", "RSS 地址格式不对");
const existing = await db
.selectFrom("site")
.selectAll()
.where("kind", "=", "rss_feed")
.where("deleted_at", "is", null)
.where("base_url", "=", `https://${host}`)
.orderBy("created_at")
.executeTakeFirst();
if (existing) return existing;
return db
.insertInto("site")
.values({
key: await uniqueSiteKey(db, url),
name: host,
kind: "rss_feed",
adapter: "generic_rss",
base_url: `https://${host}`,
search_default: false,
})
.returningAll()
.executeTakeFirstOrThrow();
}
@@ -252,6 +252,7 @@ export const sourcesOutboxRoutes: Record<string, OutboxRouteFn> = {
"feed.gap_suspected": (ev) => {
const p = ev.payload as P;
if (!p.siteId) return null;
// 站点不支持列表抓取时 snapshotSite 自己跳过(reason=unsupported)
return { task: "snapshotSite", input: { siteId: p.siteId, kind: "backfill", maxPages: BACKFILL_MAX_PAGES } };
},
};