docs(plan): on-demand site detail page parsing (HHClub details.php)

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

apps/server/src/modules/search/releases.ts

@@ -1,3 +1,4 @@
+import { randomUUID } from "node:crypto";
 import type { SiteRelease as AdapterRelease } from "@umt/adapters";
 import { SiteError } from "@umt/adapters";
 import type {
@@ -15,8 +16,9 @@ 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 进识别流水线)。
+// 跨站资源搜索:选中的已启用站点逐站调适配器 search(每站共享限流器 + 单站超时 + 站点级并发上限);
+// 按(站点, 查询词)并发合并、短缓存 —— 换站点选择时已查过的站点直接复用;结果 upsert 为资源
+// (source_release,新资源照常发 release.observed 进识别流水线)。
 // 作品关联与库存叠加在每次请求时现算(不进缓存),识别完成后刷新即可看到。
 
 /** 缓存里的资源条目(不含作品关联与库存叠加) */
@@ -37,46 +39,96 @@ export interface FanoutOptions {
 
 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; fanout: Fanout }>();
-  private readonly inflight = new Map<string, Promise<Fanout>>();
+  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,
   ) {}
 
-  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 });
+  /** 参与搜索的候选站点:启用且未删除(按名称排序) */
+  async enabledSites(): Promise<SiteRow[]> {
+    return this.rt()
+      .ctx.db.selectFrom("site")
+      .selectAll()
+      .where("enabled", "=", true)
+      .where("deleted_at", "is", null)
+      .orderBy("name")
+      .execute();
+  }
+
+  /**
+   * 在给定站点上搜一组查询词:每站每词一次请求(同站多个词的结果按站点资源 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 fresh: string[] = [];
+    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;
         }
-        return fanout;
-      })
-      .finally(() => this.inflight.delete(key));
-    this.inflight.set(key, p);
-    return { fanout: await p, cached: false, coalesced: false };
+        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;
+        this.cache.set(key, { expires: this.opts.now() + this.opts.ttlMs, result: r });
+        fresh.push(key);
+        parts.push(r);
+      }
+      return mergeSite(site, parts);
+    });
+    // 本轮新查的站点全部失败:撤掉这些缓存
+    const freshResults = fresh.map((k) => this.cache.get(k)?.result).filter((r): r is SiteResult => !!r);
+    if (freshResults.length && freshResults.every((r) => r.coverage.status !== "ok"))
+      for (const k of fresh) this.cache.delete(k);
+    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 { fanout } of this.cache.values()) {
-      const r = fanout.items.find((i) => i.siteId === siteId && i.siteReleaseId === siteReleaseId);
+    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 };
     }
@@ -87,32 +139,11 @@ export class ReleaseFanout {
   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);
+    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 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 }> {
+  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,
@@ -122,22 +153,27 @@ export class ReleaseFanout {
       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 { items: [], coverage: cov("skipped", 0, (e as Error).message) };
+      return done([], cov("skipped", 0, (e as Error).message));
     }
-    if (!adapter.capabilities.search) return { items: [], coverage: cov("skipped", 0, "站点不支持搜索") };
+    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 { items: [], coverage: cov("timeout", 0, timeoutMsg) };
+      if (signal.aborted) return done([], cov("timeout", 0, timeoutMsg));
       const msg = e instanceof SiteError ? `${e.kind}: ${e.message}` : (e as Error).message || "搜索失败";
-      return { items: [], coverage: cov("failed", 0, msg) };
+      return done([], cov("failed", 0, msg));
     }
     // 同站同 ID 去重(翻页边界可能重复)
     const seen = new Set<string>();
@@ -152,13 +188,70 @@ export class ReleaseFanout {
       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 { items, coverage: cov("ok", items.length, null) };
+    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 {
@@ -226,7 +319,7 @@ const list = (s: string | undefined) =>
     .filter(Boolean);
 
 export function filterReleases(items: BaseRelease[], q: ReleaseSearchQuery): BaseRelease[] {
-  const sites = new Set(list(q.sites));
+  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) => {