Add opinion monitor quickstart

.gitignore

@@ -81,3 +81,7 @@ odp-rs/target/
 .gstack/
 .repowise/
 .understand-anything/
+.codex/
+.sentrux/agent-sessions/
+.sentrux/baseline.json
+artifacts/

README.md

@@ -101,6 +101,52 @@ docker compose up -d   # 启动中心 + agent-1
 
 ---
 
+### 舆情监控实战闭环
+
+当前实战链路已经投到真实运行面，而不是只停留在配置说明：
+
+1. **多账号 / 多节点采集** — 通过「节点管理」和站点绑定，把 `opencli` 采集路由到指定 WS agent；验收脚本会证明 `chrome_endpoint` 和 `node_url` 都落在绑定节点。
+2. **AI 摘要与打标** — `collect → normalize → store → ai → notify` 流水线会把模型输出写入 `ai_enrichment`，监控台读取真实记录展示摘要、标签和情绪分布。
+3. **飞书推送** — 飞书模板可以直接引用 `{{summary}}`、`{{tags}}`、`{{sentiment}}`，把 AI 处理后的内容推到群机器人。
+4. **可视化验收** — 「监控台」的舆情监控卡片读取 `/api/v1/dashboard/opinion-monitor`，展示最近热点、AI 处理量、Feishu sent/failed 证据和来源贡献。
+
+一键生成实战配置：
+
+```bash
+curl -X POST http://localhost:8000/api/v1/presets/opinion-monitor/apply \
+  -H "Content-Type: application/json" \
+  -d '{
+    "source_prefix": "实战舆情",
+    "feishu_webhook_url": "https://open.feishu.cn/open-apis/bot/v2/hook/xxx"
+  }'
+```
+
+这会创建两条默认 `aibase news` 多账号采集源、对应定时计划，以及一个飞书规则。
+如果暂时不填 `feishu_webhook_url`，飞书规则会以 disabled 状态创建，不会伪造推送成功。
+
+关键验收命令：
+
+```powershell
+scripts\acceptance\fleet-acceptance.ps1 `
+  -Site aibase `
+  -Command news `
+  -Limit 1 `
+  -CenterPort 8032 `
+  -AgentPort 19824 `
+  -FreshDb
+```
+
+如果本机已有旧 API/agent 进程占用端口，可以换一组固定端口，例如
+`-CenterPort 8035 -AgentPort 19828`。
+
+最终应输出：
+
+```text
+ACCEPTANCE: PASS
+```
+
+---
+
 ## 快速开始
 
 ### 方式零：前后端本地开发（推荐）

backend/agent_server.py

@@ -91,15 +91,17 @@ def _resolve_bin(mode: str) -> str:  # noqa: ARG001
 _CENTRAL_API_URL = os.environ.get("CENTRAL_API_URL", "").rstrip("/")
 _AGENT_ADVERTISE_URL = os.environ.get("AGENT_ADVERTISE_URL", "")
 _AGENT_MODE = os.environ.get("AGENT_MODE", "cdp")
-# Deployment/startup type reported to center: "docker" (running in container) | "shell" (native process)
+# Deployment/startup type reported to center:
+# "docker" (container) | "shell" (native process).
 _AGENT_DEPLOY_TYPE = os.environ.get("AGENT_DEPLOY_TYPE", "docker")
 # True when the image was built with INSTALL_CHROME=true (Chrome bundled inside container).
 # False → Chrome runs on the host; localhost must be remapped to host.docker.internal.
 _AGENT_HAS_CHROME = os.environ.get("AGENT_HAS_CHROME", "false").lower() == "true"
 _AGENT_LABEL = os.environ.get("AGENT_LABEL", socket.gethostname())
 # Registration mode:
 #   http — LAN mode: agent POSTs its URL to center, center calls back via HTTP (default)
-#   ws   — NAT/reverse-channel mode: agent opens WS to center, registration via WS handshake (Phase 2)
+#   ws   — NAT/reverse-channel mode: agent opens WS to center, then
+#          registers through the WS handshake.
 #   off  — disable auto-registration entirely
 _AGENT_REGISTER = os.environ.get("AGENT_REGISTER", "http").lower()
 # opencli subprocess execution timeout in seconds
@@ -191,7 +193,12 @@ async def _register_with_center(advertise_url: str) -> None:
             return
         except Exception as exc:
             wait = attempt * 3
-            logger.warning("Registration attempt %d failed: %s — retrying in %ds", attempt, exc, wait)
+            logger.warning(
+                "Registration attempt %d failed: %s — retrying in %ds",
+                attempt,
+                exc,
+                wait,
+            )
             await asyncio.sleep(wait)
     logger.error("Could not register with center after 5 attempts")
 
@@ -281,7 +288,11 @@ async def _send_result(result: dict) -> None:
                     "event": event,
                 }))
             except Exception as exc:
-                logger.error("WS: failed to send agent_event for request_id=%s: %s", request_id, exc)
+                logger.error(
+                    "WS: failed to send agent_event for request_id=%s: %s",
+                    request_id,
+                    exc,
+                )
         if terminal_event is None:
             # Contract violation (adapter yielded nothing) — still must resolve
             # the center's pending future rather than hang it until timeout.
@@ -293,7 +304,11 @@ async def _send_result(result: dict) -> None:
             }
         await _send_result(terminal_event)
     except RuntimeInvocationError as exc:
-        logger.exception("WS agent_task request_id=%s: adapter invocation error: %s", request_id, exc)
+        logger.exception(
+            "WS agent_task request_id=%s: adapter invocation error: %s",
+            request_id,
+            exc,
+        )
         await _send_result({
             "type": "error",
             "task_id": request_id,
@@ -409,11 +424,19 @@ async def lifespan(app: FastAPI):
                     _CENTRAL_API_URL or "", _AGENT_REGISTER)
     elif _AGENT_REGISTER == "http":
         advertise_url = _detect_advertise_url()
-        logger.info("LAN registration: advertise_url=%s → center=%s", advertise_url, _CENTRAL_API_URL)
+        logger.info(
+            "LAN registration: advertise_url=%s → center=%s",
+            advertise_url,
+            _CENTRAL_API_URL,
+        )
         asyncio.get_event_loop().create_task(_register_with_center(advertise_url))
     elif _AGENT_REGISTER == "ws":
         advertise_url = _detect_advertise_url()
-        logger.info("WS registration: advertise_url=%s → center=%s", advertise_url, _CENTRAL_API_URL)
+        logger.info(
+            "WS registration: advertise_url=%s → center=%s",
+            advertise_url,
+            _CENTRAL_API_URL,
+        )
         _ws_task = asyncio.get_event_loop().create_task(_register_via_ws(advertise_url))
     yield
     if _ws_task and not _ws_task.done():
@@ -468,7 +491,11 @@ async def _cleanup_cdp_tabs(cdp_endpoint: str, pre_existing_ids: set[str]) -> No
                 if tab.get("type") == "page" and tab_id not in pre_existing_ids:
                     try:
                         await client.get(f"{cdp_endpoint}/json/close/{tab_id}")
-                        logger.info("cleanup: closed new tab %s url=%s", tab_id, tab.get("url", "")[:80])
+                        logger.info(
+                            "cleanup: closed new tab %s url=%s",
+                            tab_id,
+                            tab.get("url", "")[:80],
+                        )
                         remaining_pages -= 1
                     except Exception:
                         pass
@@ -560,7 +587,7 @@ async def collect(req: CollectRequest) -> dict:
         )
         stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=_OPENCLI_TIMEOUT)
         rc = proc.returncode
-    except asyncio.TimeoutError:
+    except TimeoutError:
         logger.error("timeout | cmd=%s", " ".join(cmd))
         if proc:
             proc.kill()