Files
UHD_Media_Tool/apps/server/src/modules/search/releases.ts
T
rootandClaude Opus 5.5 6e71df2769 fix(search): 站点搜索失败、超时、被拦截的结果不再写缓存
之前失败结果也缓存 30 秒,只在一轮全部站点失败时才撤销;某站一次失败后,
后续几次搜索都复用同一个失败。现在只缓存成功结果,失败 / 超时 / 跳过下次直接重试。

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
2026-10-06 15:33:58 +08:00

545 lines
21 KiB
TypeScript

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<SearchRelease, "mediaEntityId" | "entityMatch" | "entity" | "inventory" | "ruleVerdict">;
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<SearchSiteCoverage["status"], number> = { ok: 0, timeout: 1, failed: 2, skipped: 3 };
export class ReleaseFanout {
private readonly cache = new Map<string, { expires: number; result: SiteResult }>();
private readonly inflight = new Map<string, Promise<SiteResult>>();
constructor(
private readonly rt: () => SourcesRuntime,
private readonly opts: FanoutOptions,
) {}
/** 参与搜索的候选站点:启用、未删除且支持搜索(IA-2 按站点类型与定义的获取方式;按名称排序) */
async enabledSites(): Promise<SiteRow[]> {
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<FanoutHit> {
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<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 > 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<SiteResult> {
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<SourcesRuntime["siteAdapter"]>;
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<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 }, "搜索结果写入资源表失败");
}
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<string>();
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<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,
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<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.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<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>();
// 规则判定:失败只记警告,不影响结果列表
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<string, never>();
});
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,
};
}