feat(search): global search aggregation backend (W9-G)
- GET /search?q=&scope=all|media|releases|actions&limit=: local works (title_alias
exact/prefix/trgm + name prefix, slot status summary), TMDB candidates (cached,
localEntityId), cross-site releases, page/action jump list; tookMs per group
- cross-site: enabled sites via shared per-site limiter, 15s AbortSignal timeout,
site concurrency cap 4, same-query coalescing, 30s cache; partial + failedSites;
results upserted to source_release; inventory overlay via GET /inventory
- GET /search/releases: site/resolution/source/size/free/H&R filters, sort, cursor,
facets; shares the cache
- POST /search/releases/{siteId}/{siteReleaseId}/plan: forwards in-process to
POST /acquisition-plans (same auth, validation, audit, rejections)
- contracts/search.ts, runbook, tests (partial/timeout, coalesce, cache, overlay,
filters/sort, plan)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
rootandClaude Opus 5.5 committed
1 parent
e0c0c4f51d
commit
a0946bf480
12 files changed
+1869
-1
No files matched your search
@@ -0,0 +1,398 @@
|
||||
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 { normalizeTitle, parseRelease } from "@umt/domain";
|
||||
import type { SourcesRuntime } from "../sources-identity/runtime.ts";
|
||||
import { upsertSiteRelease } from "../sources-identity/service/snapshot.ts";
|
||||
import { raceAbort } from "./media.ts";
|
||||
|
||||
// 跨站资源搜索:已启用站点逐站调适配器 search(每站共享限流器 + 单站超时 + 站点级并发上限);
|
||||
// 同一查询词并发合并、结果短缓存;结果 upsert 为资源(source_release,新资源照常发 release.observed 进识别流水线)。
|
||||
// 作品关联与库存叠加在每次请求时现算(不进缓存),识别完成后刷新即可看到。
|
||||
|
||||
/** 缓存里的资源条目(不含作品关联与库存叠加) */
|
||||
export type BaseRelease = Omit<SearchRelease, "mediaEntityId" | "entityMatch" | "entity" | "inventory">;
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
const cacheKey = (q: string) => q.trim().toLowerCase().replace(/\s+/g, " ");
|
||||
|
||||
export class ReleaseFanout {
|
||||
private readonly cache = new Map<string, { expires: number; fanout: Fanout }>();
|
||||
private readonly inflight = new Map<string, Promise<Fanout>>();
|
||||
|
||||
constructor(
|
||||
private readonly rt: () => SourcesRuntime,
|
||||
private readonly opts: FanoutOptions,
|
||||
) {}
|
||||
|
||||
async search(q: string, opts: { refresh?: boolean } = {}): Promise<FanoutHit> {
|
||||
const key = cacheKey(q);
|
||||
const hit = this.cache.get(key);
|
||||
if (!opts.refresh && hit && hit.expires > this.opts.now())
|
||||
return { fanout: hit.fanout, cached: true, coalesced: false };
|
||||
const running = this.inflight.get(key);
|
||||
if (running) return { fanout: await running, cached: false, coalesced: true };
|
||||
const p = this.run(q.trim())
|
||||
.then((fanout) => {
|
||||
// 全部站点失败不缓存(下次直接重试)
|
||||
if (fanout.sites.length === 0 || fanout.sites.some((s) => s.status === "ok")) {
|
||||
this.prune();
|
||||
this.cache.set(key, { expires: this.opts.now() + this.opts.ttlMs, fanout });
|
||||
}
|
||||
return fanout;
|
||||
})
|
||||
.finally(() => this.inflight.delete(key));
|
||||
this.inflight.set(key, p);
|
||||
return { fanout: await p, cached: false, coalesced: false };
|
||||
}
|
||||
|
||||
/** 缓存里找一条站点资源(下载时资源行还没落库的兜底) */
|
||||
findCached(siteId: string, siteReleaseId: string): { release: BaseRelease; raw: AdapterRelease } | null {
|
||||
for (const { fanout } of this.cache.values()) {
|
||||
const r = fanout.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<string, AdapterRelease>();
|
||||
|
||||
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 > 200) this.cache.delete(this.cache.keys().next().value as string);
|
||||
if (this.raws.size > 20_000) this.raws.clear();
|
||||
}
|
||||
|
||||
private async run(q: string): Promise<Fanout> {
|
||||
const rt = this.rt();
|
||||
const sites = await rt.ctx.db
|
||||
.selectFrom("site")
|
||||
.selectAll()
|
||||
.where("enabled", "=", true)
|
||||
.where("deleted_at", "is", null)
|
||||
.orderBy("name")
|
||||
.execute();
|
||||
const results = await mapLimit(sites, Math.max(1, this.opts.concurrency), (s) => this.searchSite(rt, s, q));
|
||||
return {
|
||||
items: results.flatMap((r) => r.items),
|
||||
sites: results.map((r) => r.coverage),
|
||||
fetchedAt: new Date(this.opts.now()).toISOString(),
|
||||
};
|
||||
}
|
||||
|
||||
private async searchSite(
|
||||
rt: SourcesRuntime,
|
||||
site: SiteRow,
|
||||
q: string,
|
||||
): Promise<{ items: BaseRelease[]; coverage: SearchSiteCoverage }> {
|
||||
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,
|
||||
});
|
||||
let adapter: ReturnType<SourcesRuntime["siteAdapter"]>;
|
||||
try {
|
||||
adapter = rt.siteAdapter(site);
|
||||
} catch (e) {
|
||||
return { items: [], coverage: cov("skipped", 0, (e as Error).message) };
|
||||
}
|
||||
if (!adapter.capabilities.search) return { items: [], coverage: 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 { items: [], coverage: cov("timeout", 0, timeoutMsg) };
|
||||
const msg = e instanceof SiteError ? `${e.kind}: ${e.message}` : (e as Error).message || "搜索失败";
|
||||
return { items: [], coverage: cov("failed", 0, msg) };
|
||||
}
|
||||
// 同站同 ID 去重(翻页边界可能重复)
|
||||
const seen = new Set<string>();
|
||||
found = found.filter((r) => !seen.has(r.siteReleaseId) && !!seen.add(r.siteReleaseId));
|
||||
const ids = new Map<string, string>();
|
||||
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 }, "搜索结果写入资源表失败");
|
||||
}
|
||||
const items = found.map((r) => {
|
||||
const base = toBase(site, r, ids.get(r.siteReleaseId) ?? null);
|
||||
this.raws.set(base.key, r);
|
||||
return base;
|
||||
});
|
||||
return { items, coverage: cov("ok", items.length, null) };
|
||||
}
|
||||
}
|
||||
|
||||
function toBase(site: Pick<SiteRow, "id" | "name">, 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,
|
||||
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,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/** 固定并发上限的 map(保持输入顺序) */
|
||||
export async function mapLimit<T, R>(items: readonly T[], limit: number, fn: (t: T) => Promise<R>): Promise<R[]> {
|
||||
const out = new Array<R>(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.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;
|
||||
});
|
||||
}
|
||||
|
||||
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<string, FacetCount>();
|
||||
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<Map<string, InventoryEntry>>;
|
||||
|
||||
export async function decorate(
|
||||
rt: SourcesRuntime,
|
||||
items: BaseRelease[],
|
||||
loadInventory: InventoryLoader,
|
||||
): Promise<SearchRelease[]> {
|
||||
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<string, string>();
|
||||
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<string, { id: string; year: number | null }[]>();
|
||||
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<string, { id: string; match: "resolved" | "alias" }>();
|
||||
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<string>();
|
||||
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<string, { id: string; title: string; year: number | null; kind: string }>();
|
||||
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<string, InventoryEntry>();
|
||||
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,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
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,
|
||||
};
|
||||
}
|
||||
Reference in new issue
Block a user