feat: acquisition hardening — opencli fetch() migration, catalog-driven demand matching, dedup proof

backend/channels/opencli_channel.py

@@ -14,7 +14,13 @@
 
 import yaml
 
-from backend.channels.base import AbstractChannel, Capabilities, ChannelResult
+from backend.channels.base import (
+    AbstractChannel,
+    Capabilities,
+    ChannelResult,
+    FetchContext,
+    FetchResult,
+)
 from backend.channels.registry import register_channel
 from backend.opencli_runtime import configured_opencli_bin, resolve_opencli_bin
 
@@ -610,12 +616,27 @@ async def _collect_with_opencli_subprocess(
 
 @register_channel
 class OpenCLIChannel(AbstractChannel):
-    """Collect data by running the opencli CLI tool."""
+    """Collect data by running the opencli CLI tool.
+
+    Migrated onto the thick ``fetch()`` contract (see ``fetch()`` below) via the
+    same narrow-override pattern as ``BrowserActChannel`` (backend.channels.
+    browser_act_channel): ``collect()`` stays the single source of truth for
+    site/command routing, and ``fetch()`` only exists to flip channel_runner's
+    ``channel_migrated`` check.
+    """
 
     channel_type = "opencli"
     # Drives a real Chrome from the shared pool → must run on the node holding the
     # live session; the pipeline resolves a site-keyed browser binding for it.
-    capabilities = Capabilities(session_affinity=True)
+    # incremental/paginated stay False: opencli's site/command catalog is an
+    # external binary discovered at runtime via `--help` (see _get_named_options /
+    # _command_requires_browser) — there is no cursor or page-token contract for
+    # it anywhere in this codebase to drive a runner-owned pagination loop against.
+    # default_rate is spelled out (rather than left to the dataclass default) to
+    # document it's a deliberate choice, not an oversight: same 60/min every other
+    # browser-driving channel (BrowserActChannel, SkillChannel) accepts, since
+    # there's no empirical number specific to opencli to justify a different one.
+    capabilities = Capabilities(session_affinity=True, default_rate="60/min")
 
     async def collect(
         self, config: dict[str, Any], parameters: dict[str, Any]
@@ -787,6 +808,38 @@ async def collect(
 
             return result
 
+    async def fetch(self, ctx: FetchContext) -> FetchResult:
+        """Thick-contract entry point, overridden narrowly (same pattern as
+        ``BrowserActChannel.fetch()``, backend.channels.browser_act_channel:454) so
+        that ``type(chan).fetch is not AbstractChannel.fetch`` and
+        ``channel_runner.run_channel`` (backend.pipeline.channel_runner) treats
+        opencli as migrated: it builds a ``RateLimitedClient`` from
+        ``capabilities.default_rate`` for the run instead of skipping it, and any
+        failure's ``error_type`` (already set by ``_collect_with_opencli_subprocess``
+        for TimeoutError/FileNotFoundError/OSError/JSON-parse errors — see that
+        function) reaches ``error_taxonomy.effective_error_type`` directly instead
+        of relying on the inherited default doing the identical thing implicitly.
+
+        ``collect()`` remains the single source of truth for site/command routing
+        (direct subprocess vs. LAN-agent HTTP vs. WS-agent dispatch, browser-pool
+        acquisition, CDP tab snapshot/cleanup) — this override delegates straight
+        to the inherited default, because unlike ``BrowserActChannel`` there is no
+        ``ctx.source_id``-driven credential lookup to thread through: opencli has
+        no per-source encrypted-credential path (``capabilities.auth_kind`` stays
+        "none", so ``run_channel`` resolves ``AuthContext`` without a DB hit).
+
+        ``ctx.http`` is deliberately NOT threaded into ``collect()``: opencli's
+        transport is a local subprocess (direct/cdp/bridge modes) or a LAN-agent
+        HTTP/WS dispatch to an internal node (``_collect_via_agent`` /
+        ``_collect_via_ws_agent``) authenticated with the fleet's own bearer
+        token — neither is the public-API/SSRF-guarded shape ``ctx.http``'s
+        rate-limited client is built for. Same accepted trade-off
+        ``BrowserActChannel.fetch()`` documents: the ``RateLimitedClient`` the
+        runner builds but this channel never reads is one Python object for the
+        run's duration, not an open socket.
+        """
+        return await AbstractChannel.fetch(self, ctx)
+
     async def validate_config(self, config: dict[str, Any]) -> list[str]:
         errors: list[str] = []
         if not config.get("site"):

backend/workflow/demand_assembler.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import logging
 import re
 from typing import Any
 
@@ -14,8 +15,11 @@
     WorkflowProjectEdge,
     WorkflowProjectNode,
 )
+from backend.workflow.opencli_adapter_nodes import list_opencli_adapter_nodes
 from backend.workflow.patcher import preview_workflow_patch
 
+logger = logging.getLogger(__name__)
+
 
 def draft_workflow_demand(body: WorkflowDemandDraftRequest) -> WorkflowPatchResponse:
     """Translate a user collection need into reviewable native-node patches.
@@ -416,6 +420,19 @@ def _data_operators_for_need(text: str) -> list[dict[str, Any]]:
 
 
 def _source_slots_for_need(text: str) -> list[dict[str, Any]]:
+    """Resolve a collection need to native OpenCLI source slots.
+
+    Consults the OpenCLI adapter catalog first (~205 site adapters) so any
+    catalog-known site can be matched, not just the two hardcoded below.
+    Catalog access is best-effort and never raises: if it can't be reached
+    (binary missing, subprocess/decode failure, anything at all) this falls
+    back to the legacy keyword floor, which is byte-for-byte what this
+    function did before catalog matching existed.
+    """
+    return _catalog_slots_for_need(text) or _legacy_keyword_slots_for_need(text)
+
+
+def _legacy_keyword_slots_for_need(text: str) -> list[dict[str, Any]]:
     normalized = text.lower()
     slots: list[dict[str, Any]] = []
     keyword = _keyword_from_need(text)
@@ -459,6 +476,98 @@ def _keyword_from_need(text: str) -> str:
     return value or "热门"
 
 
+# --- OpenCLI adapter catalog matching --------------------------------------
+#
+# Chinese aliases for catalog sites that are commonly typed in Chinese rather
+# than by their OpenCLI site slug. An alias for a site that isn't present in
+# the loaded catalog is harmless -- it simply never matches anything.
+_CATALOG_SITE_ALIASES: dict[str, tuple[str, ...]] = {
+    "xiaohongshu": ("小红书", "xiaohongshu", "xhs"),
+    "bilibili": ("哔哩", "bilibili", "b站", "bili"),
+}
+
+_CATALOG_SLOT_CAP = 3
+_CATALOG_TOKEN_PATTERN = re.compile(r"[a-zA-Z一-鿿]+")
+
+
+def _catalog_slots_for_need(text: str) -> list[dict[str, Any]]:
+    """Match a need against the OpenCLI adapter catalog.
+
+    Deterministic, no LLM. Scores each read-access catalog site against the
+    need text by (a) exact site id/name token, (b) Chinese alias substring,
+    (c) description/domain/strategy keyword -- in that priority -- and keeps
+    only the best-scoring tier per site. Never raises: any catalog access
+    failure (missing binary, subprocess error, decode error, ...) is caught
+    and treated as "no catalog available", returning an empty list so the
+    caller can fall back to the legacy keyword floor.
+    """
+    try:
+        nodes = list_opencli_adapter_nodes(include_write=False).nodes
+    except Exception:
+        logger.debug(
+            "opencli adapter catalog unavailable for demand matching", exc_info=True
+        )
+        return []
+
+    normalized = text.strip().lower()
+    if not normalized or not nodes:
+        return []
+
+    keyword = _keyword_from_need(text)
+    best: dict[str, tuple[int, int, str]] = {}
+    for index, node in enumerate(nodes):
+        tier = _catalog_match_tier(normalized, node)
+        if tier is None:
+            continue
+        current = best.get(node.site)
+        if current is None or tier < current[0]:
+            best[node.site] = (tier, index, node.command)
+
+    ordered_sites = sorted(best.items(), key=lambda item: (item[1][0], item[1][1]))
+    slots: list[dict[str, Any]] = []
+    for site, (_tier, _index, command) in ordered_sites[:_CATALOG_SLOT_CAP]:
+        slots.append(
+            {
+                "id": _catalog_slot_id(site),
+                "label": f"{site.title()} {command.title()}".strip(),
+                "sourceGroup": "opencli",
+                "site": site,
+                "command": command,
+                "args": {"keyword": keyword},
+            }
+        )
+    return slots
+
+
+def _catalog_match_tier(normalized: str, node: Any) -> int | None:
+    # Command names (node.command) repeat across unrelated sites, so a
+    # command-only hit must never claim the exact-site tier.
+    site = node.site.lower()
+    if len(site) >= 2 and site in normalized:
+        return 0
+    for alias in _CATALOG_SITE_ALIASES.get(site, ()):
+        if alias.lower() in normalized:
+            return 1
+    if any(token in normalized for token in _catalog_description_tokens(node)):
+        return 2
+    return None
+
+
+def _catalog_description_tokens(node: Any) -> list[str]:
+    haystack = " ".join(
+        value for value in (node.description, node.domain, node.strategy) if value
+    )
+    return [
+        token.lower()
+        for token in _CATALOG_TOKEN_PATTERN.findall(haystack)
+        if len(token) >= 2
+    ]
+
+
+def _catalog_slot_id(site: str) -> str:
+    return re.sub(r"[^a-z0-9]+", "-", site.strip().lower()).strip("-") or "source"
+
+
 def _unique_node_id(project: WorkflowProject, base: str) -> str:
     return _unique_id({node.id for node in project.nodes}, base)
 

docs/schema.md

@@ -17,12 +17,21 @@ SQLAlchemy 模型：[`backend/models/record.py`](../backend/models/record.py)
 | `normalized_data` | JSON | **下游应消费的标准字段**，详见下表 |
 | `ai_enrichment` | JSON / NULL | AI processor 写回的 enrichment；schema 由 processor + prompt 决定，无强制契约 |
 | `content_hash` | TEXT (sha256) | `(source_id, content_hash)` 唯一约束用于去重 |
+| `identity_key` | TEXT / NULL | 渠道 `identity()` 提供的稳定原生 id（RSS entry id、tweet id...）；NULL 表示渠道未实现 `identity()`。补充去重键，非 `content_hash` 的替代——同一 identity 命中且内容变化时就地更新该行，而不是插入新行 |
 | `status` | TEXT | `raw` → `normalized` → `ai_processed` → `notified`；失败为 `error` |
 | `error_message` | TEXT / NULL | `status='error'` 时填错误描述 |
 | `created_at` | DATETIME | 入库时间 |
 | `updated_at` | DATETIME | 末次更新（含 AI 回填） |
 
-唯一约束：`uq_source_content (source_id, content_hash)`。
+唯一约束：`uq_source_content (source_id, content_hash)`。非唯一索引：`ix_collected_records_source_identity (source_id, identity_key)`。
+
+### 去重保证
+
+按数据源去重是**入库层的强制保证**，不是可选的工作流节点：任何 source（不论通过 `POST /sources` 直接创建，还是工作流编排）在写入这张表时都会经过同一条路径——`backend/pipeline/normalizer.py` 对每条 item 无条件计算 `content_hash`，`backend/pipeline/storer.py` 在写入前按 `source_id` 查重（含同一批次内部去重），并由上面的唯一约束在 DB 层兜底并发竞争。跳过的条数经由 `SinkResult.duplicates` 一路传导到 `SourceMeasurement.duplicates`/`duplicate_rate`（见 [`CONTROL_THEORY_ARCHITECTURE.md`](CONTROL_THEORY_ARCHITECTURE.md)）。
+
+工作流图上的 `text.deduplicate` 算子（`backend/workflow/demand_assembler.py`，仅在需求文本命中质量类关键词时才挂载）是完全独立的另一层：面向已入库内容的 AI 语料整理（exact/SimHash 近似去重可配置字段），不是入库去重的前提条件——没有它，入库去重依然生效。
+
+写入路径由 `write_strategy` 决定（`backend/pipeline/sinks/strategy.py`）：默认 `legacy` 和 `odp_shadow`/`odp_dual_required`/`odp_primary`（`DualSink`，legacy leg 为准）都会经过上述去重路径；只有 `odp_only`（不落这张表）把去重交给独立的 Rust ODP ingest 服务，语义由该服务自行保证，不在本文档范围内。
 
 ## `normalized_data` JSON 内的标准字段