import { randomUUID } from "node:crypto"; import type { SiteRelease as AdapterRelease } from "@umt/adapters"; import { SiteError } from "@umt/adapters"; import type { FacetCount, InventoryEntry, ReleaseSearchQuery, SearchInventoryState, SearchRelease, SearchReleaseInventory, SearchSiteCoverage, } from "@umt/contracts"; 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"; import { ruleVerdicts } from "./rule-verdict.ts"; // 跨站资源搜索:选中的已启用站点逐站调适配器 search(每站共享限流器 + 单站超时 + 站点级并发上限); // 按(站点, 查询词)并发合并、短缓存 —— 换站点选择时已查过的站点直接复用;结果 upsert 为资源 // (source_release,新资源照常发 release.observed 进识别流水线)。 // 作品关联与库存叠加在每次请求时现算(不进缓存),识别完成后刷新即可看到。 /** 缓存里的资源条目(不含作品关联与库存叠加) */ export type BaseRelease = Omit; export interface Fanout { items: BaseRelease[]; sites: SearchSiteCoverage[]; fetchedAt: string; } export interface FanoutOptions { timeoutMs: number; concurrency: number; ttlMs: number; now: () => number; } export interface FanoutHit { fanout: Fanout; /** 全部(站点, 查询词)都命中缓存 */ cached: boolean; /** 至少一个(站点, 查询词)与同时进行的相同搜索合并 */ coalesced: boolean; } interface SiteResult { items: BaseRelease[]; coverage: SearchSiteCoverage; fetchedAt: number; } const cacheKey = (q: string) => q.trim().toLowerCase().replace(/\s+/g, " "); const COVERAGE_RANK: Record = { ok: 0, timeout: 1, failed: 2, skipped: 3 }; export class ReleaseFanout { private readonly cache = new Map(); private readonly inflight = new Map>(); constructor( private readonly rt: () => SourcesRuntime, private readonly opts: FanoutOptions, ) {} /** 参与搜索的候选站点:启用、未删除且支持搜索(IA-2 按站点类型与定义的获取方式;按名称排序) */ async enabledSites(): Promise { 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")); } /** * 在给定站点上搜一组查询词:每站每词一次请求(同站多个词的结果按站点资源 ID 去重)。 * 缓存按(站点, 查询词):只缓存成功的结果;失败 / 超时 / 验证拦截 / 跳过不写缓存,下次搜索直接重试。 */ async search(terms: string[], sites: SiteRow[], opts: { refresh?: boolean } = {}): Promise { const words = [...new Map(terms.map((t) => [cacheKey(t), t.trim()])).entries()].filter(([k]) => k); let cached = true; let coalesced = false; const perSite = await mapLimit(sites, Math.max(1, this.opts.concurrency), async (site) => { const parts: SiteResult[] = []; for (const [k, term] of words) { const key = `${site.id}|${k}`; const hit = this.cache.get(key); if (!opts.refresh && hit && hit.expires > this.opts.now()) { parts.push(hit.result); continue; } cached = false; const running = this.inflight.get(key); if (running) { coalesced = true; parts.push(await running); continue; } const p = this.searchSite(this.rt(), site, term).finally(() => this.inflight.delete(key)); this.inflight.set(key, p); const r = await p; if (r.coverage.status === "ok") this.cache.set(key, { expires: this.opts.now() + this.opts.ttlMs, result: r }); parts.push(r); } return mergeSite(site, parts); }); this.prune(); const fetched = perSite.map((r) => r.fetchedAt); return { fanout: { items: perSite.flatMap((r) => r.items), sites: perSite.map((r) => r.coverage), fetchedAt: new Date(fetched.length ? Math.min(...fetched) : this.opts.now()).toISOString(), }, cached: cached && words.length > 0 && sites.length > 0, coalesced, }; } /** 缓存里找一条站点资源(下载时资源行还没落库的兜底) */ findCached(siteId: string, siteReleaseId: string): { release: BaseRelease; raw: AdapterRelease } | null { for (const { result } of this.cache.values()) { const r = result.items.find((i) => i.siteId === siteId && i.siteReleaseId === siteReleaseId); const raw = r ? this.raws.get(r.key) : undefined; if (r && raw) return { release: r, raw }; } return null; } private readonly raws = new Map(); private prune() { const now = this.opts.now(); for (const [k, v] of this.cache) if (v.expires <= now) this.cache.delete(k); while (this.cache.size > 1000) this.cache.delete(this.cache.keys().next().value as string); if (this.raws.size > 20_000) this.raws.clear(); } private async searchSite(rt: SourcesRuntime, site: SiteRow, q: string): Promise { const t0 = performance.now(); const cov = (status: SearchSiteCoverage["status"], count: number, error: string | null): SearchSiteCoverage => ({ siteId: site.id, siteName: site.name, status, count, tookMs: Math.round(performance.now() - t0), error, }); const done = (items: BaseRelease[], coverage: SearchSiteCoverage): SiteResult => ({ items, coverage, fetchedAt: this.opts.now(), }); let adapter: ReturnType; try { adapter = rt.siteAdapter(site); } catch (e) { return done([], cov("skipped", 0, (e as Error).message)); } if (!adapter.capabilities.search) return done([], cov("skipped", 0, "站点不支持搜索")); const signal = AbortSignal.timeout(this.opts.timeoutMs); const timeoutMsg = `${Math.round(this.opts.timeoutMs / 1000)} 秒内没有响应`; let found: AdapterRelease[]; try { found = (await raceAbort(adapter.search(q, { page: 1 }, { signal }), signal, timeoutMsg)).items; } catch (e) { if (signal.aborted) return done([], cov("timeout", 0, timeoutMsg)); const msg = e instanceof SiteError ? `${e.kind}: ${e.message}` : (e as Error).message || "搜索失败"; return done([], cov("failed", 0, msg)); } // 同站同 ID 去重(翻页边界可能重复) const seen = new Set(); found = found.filter((r) => !seen.has(r.siteReleaseId) && !!seen.add(r.siteReleaseId)); const ids = new Map(); try { await rt.ctx.db.transaction().execute(async (trx) => { for (const r of found) ids.set(r.siteReleaseId, (await upsertSiteRelease(rt, trx, site, r)).id); }); } catch (e) { // 落库失败不影响展示;下载时再按缓存补写 ids.clear(); rt.ctx.logger.warn({ err: { message: (e as Error).message }, siteId: site.id }, "搜索结果写入资源表失败"); } await this.resolveUnmatched(rt, [...ids.values()]); const items = found.map((r) => { const base = toBase(site, r, ids.get(r.siteReleaseId) ?? null); this.raws.set(base.key, r); return base; }); return done(items, cov("ok", items.length, null)); } /** * 惰性识别(DEC-1):用户搜索到的资源算「有理由查 TMDB」,未识别且未人工锁定的入队 resolveObservation{trigger:"search"}。 * 同一资源一小时内只入队一次;入队失败只记警告(资源仍经 release.observed 按 observed 处理)。 */ private async resolveUnmatched(rt: SourcesRuntime, sourceReleaseIds: string[]) { if (sourceReleaseIds.length === 0) return; try { const rows = await rt.ctx.db .selectFrom("source_release") .select("id") .where("id", "in", sourceReleaseIds) .where("media_entity_id", "is", null) .where("resolution_locked", "=", false) .limit(RESOLVE_PER_SEARCH) .execute(); const hour = Math.floor(this.opts.now() / 3_600_000); for (const r of rows) await rt.ctx.engine.enqueue( "resolveObservation", { sourceReleaseId: r.id, trigger: "search" }, { idempotencyKey: `resolve:search:${r.id}:${hour}`, traceId: randomUUID() }, ); } catch (e) { rt.ctx.logger.warn({ err: { message: (e as Error).message } }, "搜索结果入队识别失败"); } } } /** 单次站点搜索最多为多少条未识别资源入队识别 */ const RESOLVE_PER_SEARCH = 50; /** 同一站点多个查询词的结果合并:资源按 key 去重;状态取最好的一次(有一次成功就算成功) */ function mergeSite(site: SiteRow, parts: SiteResult[]): SiteResult { if (parts.length === 1) return parts[0]!; if (parts.length === 0) return { items: [], coverage: { siteId: site.id, siteName: site.name, status: "skipped", count: 0, tookMs: 0, error: "没有查询词" }, fetchedAt: Date.now(), }; const seen = new Set(); const items = parts.flatMap((p) => p.items).filter((i) => !seen.has(i.key) && !!seen.add(i.key)); const best = [...parts].sort((a, b) => COVERAGE_RANK[a.coverage.status] - COVERAGE_RANK[b.coverage.status])[0]; return { items, coverage: { siteId: site.id, siteName: site.name, status: best?.coverage.status ?? "skipped", count: items.length, tookMs: parts.reduce((n, p) => n + p.coverage.tookMs, 0), error: best?.coverage.status === "ok" ? null : (best?.coverage.error ?? null), }, fetchedAt: Math.min(...parts.map((p) => p.fetchedAt)), }; } function toBase(site: Pick, r: AdapterRelease, sourceReleaseId: string | null): BaseRelease { const p = parseRelease(r.title, r.subtitle ? { subtitle: r.subtitle } : {}); const pr = r.promotion; return { key: `${site.id}:${r.siteReleaseId}`, sourceReleaseId, siteId: site.id, siteName: site.name, siteReleaseId: r.siteReleaseId, title: r.title, subtitle: r.subtitle ?? null, category: { id: r.category.id, name: r.category.name }, sizeBytes: r.sizeBytes !== null && r.sizeBytes >= 0 ? Math.round(r.sizeBytes) : null, seeders: r.seeders, leechers: r.leechers, snatched: r.snatched, promotion: { type: pr.type, downloadFactor: pr.downloadFactor, uploadFactor: pr.uploadFactor, expiresAt: pr.expiresAt, }, free: pr.downloadFactor === 0 || pr.type === "free" || pr.type === "2xfree", hr: r.hr, uploadedAt: r.uploadedAt, detailUrl: r.detailUrl, observedAt: r.observedAt, parsed: { titleCn: p.titleCn ?? null, titleEn: p.titleEn ?? null, year: p.year ?? null, season: p.season ?? null, seasonEnd: p.seasonEnd ?? null, episodes: p.episodes ? { start: p.episodes.start, end: p.episodes.end } : null, resolution: p.resolution ?? null, source: p.source ?? null, videoCodec: p.videoCodec ?? null, hdr: [...p.hdr], releaseGroup: p.releaseGroup?.name ?? null, audio: [...p.audio], subtitles: p.subtitles ? [...p.subtitles] : null, audioLangs: p.audioLangs ? [...p.audioLangs] : null, bitDepth: p.bitDepth ?? null, fps: p.fps ?? null, cut: p.cut ?? null, }, }; } /** 固定并发上限的 map(保持输入顺序) */ export async function mapLimit(items: readonly T[], limit: number, fn: (t: T) => Promise): Promise { const out = new Array(items.length); let next = 0; const worker = async () => { while (next < items.length) { const i = next++; out[i] = await fn(items[i]!); } }; await Promise.all(Array.from({ length: Math.min(limit, items.length) }, worker)); return out; } // ── 筛选 / 排序 / 分面 ── const list = (s: string | undefined) => (s ?? "") .split(",") .map((x) => x.trim()) .filter(Boolean); export function filterReleases(items: BaseRelease[], q: ReleaseSearchQuery): BaseRelease[] { const sites = new Set(list(q.siteFilter).length ? list(q.siteFilter) : list(q.sites)); const res = new Set(list(q.resolution).map((x) => x.toLowerCase())); const src = new Set(list(q.source).map((x) => x.toLowerCase())); return items.filter((r) => { if (sites.size && !sites.has(r.siteId)) return false; if (res.size && !res.has((r.parsed.resolution ?? "unknown").toLowerCase())) return false; if (src.size && !src.has((r.parsed.source ?? "unknown").toLowerCase())) return false; if (q.minSize !== undefined && (r.sizeBytes === null || r.sizeBytes < q.minSize)) return false; if (q.maxSize !== undefined && (r.sizeBytes === null || r.sizeBytes > q.maxSize)) return false; if (q.free === "true" && !r.free) return false; if (q.free === "false" && r.free) return false; if (q.hr === "true" && r.hr !== true) return false; if (q.hr === "false" && r.hr === true) return false; return true; }); } /** 搜索结果 → 快速筛选口径的输入 */ export const searchFacetInput = (r: BaseRelease): FacetInput => ({ siteId: r.siteId, siteName: r.siteName, resolution: r.parsed.resolution, source: r.parsed.source, videoCodec: r.parsed.videoCodec, hdr: r.parsed.hdr, releaseGroup: r.parsed.releaseGroup, season: r.parsed.season, seasonEnd: r.parsed.seasonEnd ?? null, episodes: r.parsed.episodes, promotion: { type: r.promotion.type, downloadFactor: r.promotion.downloadFactor, uploadFactor: r.promotion.uploadFactor, expiresAt: r.promotion.expiresAt, }, hr: r.hr, audio: r.parsed.audio ?? [], subtitles: r.parsed.subtitles ?? null, audioLangs: r.parsed.audioLangs ?? null, bitDepth: r.parsed.bitDepth ?? null, fps: r.parsed.fps ?? null, cut: r.parsed.cut ?? null, year: r.parsed.year, category: r.category.name, sizeBytes: r.sizeBytes, seeders: r.seeders, }); /** 快速筛选(SEARCH-3):先过旧筛选参数,再按 qf 多选筛选并统计分面 */ export function quickFilter(items: BaseRelease[], q: ReleaseSearchQuery, now: number) { return applyFacets(filterReleases(items, q), searchFacetInput, parseFacetParam(q.qf), { now }); } const time = (s: string | null) => (s ? Date.parse(s) : null); /** 空值恒排最后;default = 做种数降序,同分发布时间新→旧 */ export function sortReleases( items: BaseRelease[], sort: ReleaseSearchQuery["sort"], order: ReleaseSearchQuery["order"], ): BaseRelease[] { const dir = order === "asc" ? 1 : -1; const cmpNum = (a: number | null, b: number | null, d: number) => a === null && b === null ? 0 : a === null ? 1 : b === null ? -1 : (a - b) * d; const bySeedThenTime = (a: BaseRelease, b: BaseRelease, d: number) => cmpNum(a.seeders, b.seeders, d) || cmpNum(time(a.uploadedAt), time(b.uploadedAt), -1); const cmp = (a: BaseRelease, b: BaseRelease): number => { switch (sort) { case "size": return cmpNum(a.sizeBytes, b.sizeBytes, dir) || bySeedThenTime(a, b, -1); case "uploadedAt": return cmpNum(time(a.uploadedAt), time(b.uploadedAt), dir) || bySeedThenTime(a, b, -1); case "site": return a.siteName.localeCompare(b.siteName) * (dir === 1 ? 1 : -1) || bySeedThenTime(a, b, -1); case "seeders": return bySeedThenTime(a, b, dir); default: return bySeedThenTime(a, b, dir); } }; return [...items].sort((a, b) => cmp(a, b) || a.key.localeCompare(b.key)); } export function facets(items: BaseRelease[]): { sites: FacetCount[]; resolutions: FacetCount[]; sources: FacetCount[]; } { const count = (pick: (r: BaseRelease) => [string, string]) => { const m = new Map(); for (const r of items) { const [value, label] = pick(r); const f = m.get(value) ?? { value, label, count: 0 }; f.count++; m.set(value, f); } return [...m.values()].sort((a, b) => b.count - a.count || a.label.localeCompare(b.label)); }; return { sites: count((r) => [r.siteId, r.siteName]), resolutions: count((r) => [r.parsed.resolution ?? "unknown", r.parsed.resolution ?? "未知"]), sources: count((r) => [r.parsed.source ?? "unknown", r.parsed.source ?? "未知"]), }; } // ── 作品关联 + 库存叠加(每次请求现算)── export type InventoryLoader = (entityIds: string[]) => Promise>; export async function decorate( rt: SourcesRuntime, items: BaseRelease[], loadInventory: InventoryLoader, ): Promise { if (items.length === 0) return []; const { db } = rt.ctx; const srIds = items.map((i) => i.sourceReleaseId).filter((x): x is string => !!x); const resolved = new Map(); if (srIds.length) { const rows = await db .selectFrom("source_release") .select(["id", "media_entity_id"]) .where("id", "in", srIds) .where("media_entity_id", "is not", null) .execute(); for (const r of rows) resolved.set(r.id, r.media_entity_id!); } // 未识别的:解析出的中/英文标题归一化后精确命中别名,且只命中一个作品(年份相差 ≤1)时作为未确认关联 const titleKeys = (r: BaseRelease) => [r.parsed.titleCn, r.parsed.titleEn] .filter((t): t is string => !!t) .map(normalizeTitle) .filter(Boolean); const pending = items.filter((i) => !(i.sourceReleaseId && resolved.has(i.sourceReleaseId))); const norms = [...new Set(pending.flatMap(titleKeys))]; const aliasHits = new Map(); if (norms.length) { const rows = await sql<{ normalized: string; id: string; year: number | null }>` select distinct a.normalized, e.id, e.year from title_alias a join media_entity e on e.id = a.media_entity_id where e.status <> 'merged' and a.normalized = any(${norms})`.execute(db); for (const r of rows.rows) { const l = aliasHits.get(r.normalized) ?? []; l.push({ id: r.id, year: r.year }); aliasHits.set(r.normalized, l); } } const links = new Map(); for (const it of items) { const rid = it.sourceReleaseId ? resolved.get(it.sourceReleaseId) : undefined; if (rid) { links.set(it.key, { id: rid, match: "resolved" }); continue; } const cands = new Set(); for (const k of titleKeys(it)) { for (const h of aliasHits.get(k) ?? []) { if (it.parsed.year && h.year && Math.abs(it.parsed.year - h.year) > 1) continue; cands.add(h.id); } } if (cands.size === 1) links.set(it.key, { id: [...cands][0]!, match: "alias" }); } const entityIds = [...new Set([...links.values()].map((l) => l.id))]; const entities = new Map(); if (entityIds.length) { for (const e of await db .selectFrom("media_entity") .select(["id", "title", "year", "kind"]) .where("id", "in", entityIds) .execute()) entities.set(e.id, e); } const inv = entityIds.length ? await loadInventory(entityIds) : new Map(); // 规则判定:失败只记警告,不影响结果列表 const verdicts = await ruleVerdicts( db, items.map((it) => { const link = links.get(it.key); const e = link ? entities.get(link.id) : undefined; return { key: it.key, sourceReleaseId: it.sourceReleaseId, entity: e ? { id: e.id, kind: e.kind } : null }; }), ).catch((err: Error) => { rt.ctx.logger.warn({ err: { message: err.message } }, "搜索结果规则判定失败"); return new Map(); }); return items.map((it) => { const link = links.get(it.key); const e = link ? entities.get(link.id) : undefined; const entry = link ? inv.get(link.id) : undefined; return { ...it, mediaEntityId: e ? e.id : null, entityMatch: e ? link!.match : null, entity: e ? { id: e.id, title: e.title, year: e.year, kind: e.kind as never } : null, inventory: entry ? overlay(entry, it.sourceReleaseId) : null, ruleVerdict: verdicts.get(it.key) ?? null, }; }); } function overlay(entry: InventoryEntry, sourceReleaseId: string | null): SearchReleaseInventory { const state: SearchInventoryState = entry.downloading.length ? "downloading" : entry.gaps.length ? "gap" : entry.versions.length ? "owned" : entry.subscribed ? "subscribed" : "none"; return { ...entry, candidateReasons: sourceReleaseId ? entry.candidateReasons.filter((c) => c.sourceReleaseId === sourceReleaseId) : [], state, }; }