# AI 品牌曝光监测系统技术方案 V4

## 1. 文档信息

| 项目 | 内容 |
| --- | --- |
| 文档名称 | AI 品牌曝光监测系统技术方案 V4（插件采集 + 配额约束） |
| 文档版本 | V4.0 |
| 文档状态 | 待评审 |
| 创建日期 | 2026-04-08 |
| 基线文档 | `docs/ai-brand-monitoring-tech-design-v3.md` |
| 适用范围 | 数据追踪模块全部页面 |
| 关联文档 | `docs/geo-platform-prd-v1.md` (PRD 8.4 数据追踪) |
| 关联文档 | `docs/ai-brand-monitoring-collection-feasibility-v1.md` (采集可行性) |
| 容量目标 | 5 万用户（B2B 共享租户，~3,000 租户） |

## 2. V4 修订范围

V4 在 V3 基础上做两项根本性变更：

1. **采集方式**：从服务端 AI API 调用改为 **浏览器插件采集**，消除 AI API 成本。
2. **配额约束**：引入租户级品牌/关键词/问题配额，控制采集规模。

### 2.1 与 V3 的关系

| V3 章节 | V4 处理 | 原因 |
| --- | --- | --- |
| 2.1 数据口径分层 | 保留，`run_mode` 值调整为 `plugin_*` | 口径逻辑不变，采集来源变了 |
| 2.2 Schema Delta | 保留 + 扩展 | 新增配额表、任务分配字段 |
| 3. 读流量与缓存 | **完全保留** | 读侧与采集方式无关 |
| 4. 采集流水线 | **全部替换** | 插件采集替代服务端 Collector |
| 5. 部署拓扑 | 精简 | 去掉 Collector Pod、Queue Redis |
| 6. 降级与容灾 | 保留读侧 + 新增插件降级 | 新增采集不足的降级策略 |
| 7. 压测验收 | 保留读侧 + 新增插件压测 | 新增插件并发场景 |
| 8. 实施计划 | 重写 | 工期和顺序均有变化 |

V3 其他章节（页面需求、前端组件、API 响应结构、指标定义）保持不变。

---

## 3. 用户配额体系

### 3.1 配额规则

| 配额项 | 普通租户（Free） | 高级租户（Pro） | 企业租户（Enterprise） |
| --- | --- | --- | --- |
| 最大品牌数 | 1 | 3 | 自定义 |
| 每品牌关键词数 | 8 | 8 | 自定义 |
| 每关键词问题数 | 5 | 5 | 自定义 |
| **每品牌最大问题数** | **40** | **40** | **自定义** |
| 采集频率 | 每 3 天 | 每天 | 每天 + 按需触发 |
| AI 平台数 | 3（首批） | 6（全部） | 6 + 自定义 |
| 搜索增强采集 | 不支持 | 支持 | 支持 |

### 3.2 配额表 Schema

```sql
CREATE TABLE tenant_monitoring_quotas (
    tenant_id                  UUID PRIMARY KEY REFERENCES tenants(id),
    max_brands                 INT NOT NULL DEFAULT 1,
    max_keywords_per_brand     INT NOT NULL DEFAULT 8,
    max_questions_per_keyword  INT NOT NULL DEFAULT 5,
    collect_frequency          VARCHAR(20) NOT NULL DEFAULT 'every_3_days',
    enabled_platforms          JSONB NOT NULL DEFAULT '["deepseek","qwen","doubao"]',
    search_enhanced            BOOLEAN NOT NULL DEFAULT false,
    plan_tier                  VARCHAR(20) NOT NULL DEFAULT 'free',
    created_at                 TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at                 TIMESTAMPTZ NOT NULL DEFAULT now()
);

COMMENT ON COLUMN tenant_monitoring_quotas.collect_frequency IS 'daily / every_3_days / weekly';
COMMENT ON COLUMN tenant_monitoring_quotas.plan_tier IS 'free / pro / enterprise';
```

### 3.3 采集规模估算

| 参数 | 估值 |
| --- | --- |
| 总用户 | 50,000 |
| 租户数 | ~3,000（平均 15-20 人/租户） |
| 高级租户（10%） | 300，品牌数 900 |
| 普通租户（90%） | 2,700，品牌数 2,700 |
| 品牌实例总数（最大） | 3,600 |
| 监测采用率（30%） | 活跃监测品牌 ~1,080 |
| 实际问题数/品牌（70% 使用率） | ~28 |

每日采集任务量（中等场景）：

```
标准模式:
  1,080 品牌 × 28 问题 × 6 平台 = 181,440 任务/天

考虑采集频率分层（Free 每 3 天）:
  高级 (30%): 324 品牌 × 28q × 6 平台 = 54,432/天
  普通 (70%): 756 品牌 × 28q × 6 平台 ÷ 3 = 42,336/天
  合计 ≈ 96,768 任务/天（均摊）
```

---

## 4. 插件采集架构

### 4.1 现有插件能力复用

当前 GEO Publisher 插件（`apps/browser-extension/`）已具备完整基础设施：

| 现有能力 | 文件位置 | 监测复用方式 |
| --- | --- | --- |
| 平台适配器模式 | `src/adapters/` (13 个) | 新增 AI 平台 adapter |
| 登录状态检测 | 各 adapter 的 `detect()` | 检测 AI 平台登录态 |
| 后台 Service Worker | `entrypoints/background.ts` | 定时触发采集任务 |
| 页面 ↔ 插件通信 | `entrypoints/content.ts` (postMessage) | 前端可触发手动采集（辅助通道） |
| 后端回调 | `src/runtime.ts` (HTTP POST) | 采集结果回传后端 |
| 后端直连 | `installation_token` + `api_base_url` | **插件后台直接拉取任务 + 回传结果** |
| Chrome Storage | `src/storage.ts` | 存储采集队列和进度 |
| DNR 跨域规则 | `public/rules.json` (18 条) | 新增 AI 平台规则 |
| 安装认证 | `installation_token` + `installation_id` | 复用，校验采集权限 |

### 4.2 整体采集流程

```
┌──────────────────────────────────────────────────────────────────┐
│                          后端调度器                                │
│  1. 根据 tenant_monitoring_quotas 和 collect_frequency            │
│     生成当日采集任务（品牌 × 去重问题 × 平台）                      │
│  2. 写入 monitoring_collect_tasks 表                              │
│  3. 状态机: pending → assigned → completed / failed / expired     │
└───────────────────────────────┬──────────────────────────────────┘
                                │
             插件 Background SW 定时心跳（每 15 分钟）
                                ▼
┌──────────────────────────────────────────────────────────────────┐
│           浏览器插件 Background Service Worker                     │
│                                                                   │
│  主通道：插件后台直接拉取（不依赖前端页面）                           │
│  ──────────────────────────────────────────                        │
│  4. 心跳触发 → GET /api/plugin/monitoring/tasks                   │
│     (Header: X-Geo-Installation-Token)                            │
│     → 后端根据 installation → tenant 映射，返回待采集任务           │
│  5. POST /api/plugin/monitoring/tasks/{id}/claim → 领取任务        │
│  6. 按 AI 平台分发到对应 adapter                                   │
│  7. 检测用户是否登录该 AI 平台（Cookie 检测）                        │
│  8. 已登录 → 调用平台 Web API 创建对话 → 输入问题 → 等待回答        │
│     未登录 → 标记 skip                                             │
│  9. 提取回答文本 + 引用信息（如有）                                  │
│  10. POST /api/callback/plugin/monitor → 结果回传后端              │
│                                                                   │
│  辅助通道：前端 postMessage（用户主动触发时）                        │
│  ──────────────────────────────────────────                        │
│  Admin Web 点击"立即采集" → postMessage → 插件立即执行              │
│  (仅作为加速手段，不是必须路径)                                      │
└───────────────────────────────┬──────────────────────────────────┘
                                │
                POST /api/callback/plugin/monitor
                                ▼
┌──────────────────────────────────────────────────────────────────┐
│                        后端采集服务                                │
│  11. 校验 installation_token + 任务归属                            │
│  12. 解析回答：品牌提及、位置、情感、引用                             │
│  13. 写入 question_monitor_runs + question_monitor_parse_results  │
│  14. 原始回答存 MinIO（对象存储）                                   │
│  15. 标记任务 completed                                           │
└──────────────────────────────────────────────────────────────────┘

                    ↓ 每日 6:30 聚合

┌──────────────────────────────────────────────────────────────────┐
│                     Aggregator Job（不变）                         │
│  16. 按 business_date 聚合已完成的采集数据                          │
│  17. 写入 10 张 monitoring_*_daily 汇总表                          │
│  18. 预热热门品牌缓存                                              │
└──────────────────────────────────────────────────────────────────┘
```

**关键设计**：插件通过 Background SW 的定时心跳（每 15 分钟）**直接调用后端 API** 拉取和领取任务，不依赖 Admin Web 前端页面是否打开。前端 postMessage 仅作为用户主动触发"立即采集"的辅助通道。

后端需要新增 **插件专用 API**（不走 JWT 认证，走 installation_token 认证）：

```
GET  /api/plugin/monitoring/tasks          → 返回当前插件所属租户的 pending 任务
POST /api/plugin/monitoring/tasks/{id}/claim → 领取任务（分布式锁）
POST /api/callback/plugin/monitor          → 回传采集结果（已有回调模式）
POST /api/callback/plugin/monitor/batch-skip → 批量跳过未登录平台的任务
```

### 4.3 AI 平台适配器

#### 4.3.1 适配器接口

```typescript
// apps/browser-extension/src/adapters/ai/types.ts

export interface AIMonitorAdapter {
  /** 平台标识 */
  platformId: AIPlatformId;

  /** 检测用户是否已登录该 AI 平台 */
  detect(): Promise<AIMonitorPlatformState>;

  /** 向 AI 平台提问并获取回答 */
  ask(question: string, options?: AskOptions): Promise<AIMonitorResult>;
}

export type AIPlatformId = 'deepseek' | 'qwen' | 'doubao' | 'kimi' | 'ernie' | 'hunyuan';

export interface AIMonitorPlatformState {
  platformId: AIPlatformId;
  connected: boolean;
  userId?: string;
  nickname?: string;
  message?: string;
}

export interface AskOptions {
  /** 是否启用联网搜索 */
  searchEnabled?: boolean;
  /** 任务超时（毫秒） */
  timeout?: number;
}

export interface AIMonitorResult {
  success: boolean;
  question: string;
  answer: string;
  citations: Citation[];
  searchResults: SearchResult[];
  responseTime: number;
  model: string;
  timestamp: string;
  rawHtml?: string;
}

export interface Citation {
  url: string;
  title: string;
  domain: string;
  snippet?: string;
}

export interface SearchResult {
  url: string;
  title: string;
  snippet: string;
}
```

#### 4.3.2 各平台采集方式

| AI 平台 | Web 端 URL | 采集方式 | 登录检测 | 引用支持 |
| --- | --- | --- | --- | --- |
| DeepSeek | chat.deepseek.com | 调用 Web 端 chat API（SSE 流） → 拼接完整回答 | Cookie 检测 | 无原生引用 |
| 千问 | tongyi.aliyun.com | 调用 Web 端对话 API → 等待回答完成 | Cookie 检测 | 联网搜索模式支持 |
| 豆包 | doubao.com | 调用 Web 端对话 API → 等待回答 | Cookie 检测 | 联网搜索支持 |
| Kimi | kimi.moonshot.cn | 调用 Web 端对话 API → 等待回答 | Cookie 检测 | 联网搜索支持 |
| 文心一言 | yiyan.baidu.com | 调用 Web 端对话 API → 等待回答 | 百度账号 Cookie | 支持 |
| 混元 | hunyuan.tencent.com | 调用 Web 端对话 API → 等待回答 | Cookie 检测 | 搜索增强支持 |

**核心实现模式**（以 DeepSeek 为例）：

```typescript
// apps/browser-extension/src/adapters/ai/deepseek.ts

import type { AIMonitorAdapter, AIMonitorPlatformState, AIMonitorResult, AskOptions } from './types';

const DEEPSEEK_CHAT_URL = 'https://chat.deepseek.com';
const DEEPSEEK_API_BASE = 'https://chat.deepseek.com/api/v0';

export const deepseekAdapter: AIMonitorAdapter = {
  platformId: 'deepseek',

  async detect(): Promise<AIMonitorPlatformState> {
    try {
      // 通过 Cookie 判断登录状态，类似现有 zhihu/bilibili adapter
      const cookies = await browser.cookies.getAll({ domain: '.deepseek.com' });
      const hasSession = cookies.some(c => c.name === 'ds_session' || c.name === 'token');

      if (!hasSession) {
        return { platformId: 'deepseek', connected: false, message: '未登录' };
      }

      // 验证 token 有效性
      const resp = await fetch(`${DEEPSEEK_API_BASE}/user/current`, {
        credentials: 'include',
      });

      if (!resp.ok) {
        return { platformId: 'deepseek', connected: false, message: '登录已过期' };
      }

      const data = await resp.json();
      return {
        platformId: 'deepseek',
        connected: true,
        userId: data.data?.id,
        nickname: data.data?.nickname,
      };
    } catch {
      return { platformId: 'deepseek', connected: false, message: '检测失败' };
    }
  },

  async ask(question: string, options?: AskOptions): Promise<AIMonitorResult> {
    const startTime = Date.now();
    const timeout = options?.timeout ?? 60_000;

    try {
      // 1. 创建新对话
      const createResp = await fetch(`${DEEPSEEK_API_BASE}/chat/create`, {
        method: 'POST',
        credentials: 'include',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({}),
      });
      const { data: chat } = await createResp.json();

      // 2. 发送问题并监听 SSE 流
      const answer = await streamChat(chat.id, question, timeout);

      return {
        success: true,
        question,
        answer: answer.text,
        citations: [],  // DeepSeek 标准模式无引用
        searchResults: [],
        responseTime: Date.now() - startTime,
        model: answer.model ?? 'deepseek-chat',
        timestamp: new Date().toISOString(),
      };
    } catch (err: any) {
      return {
        success: false,
        question,
        answer: '',
        citations: [],
        searchResults: [],
        responseTime: Date.now() - startTime,
        model: '',
        timestamp: new Date().toISOString(),
      };
    }
  },
};

async function streamChat(chatId: string, question: string, timeout: number) {
  // SSE 流式读取，拼接完整回答
  // 具体实现需根据 DeepSeek Web 端实际 API 格式调整
  const controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), timeout);

  try {
    const resp = await fetch(`${DEEPSEEK_API_BASE}/chat/completion`, {
      method: 'POST',
      credentials: 'include',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        chat_id: chatId,
        prompt: question,
        stream: true,
      }),
      signal: controller.signal,
    });

    const reader = resp.body!.getReader();
    const decoder = new TextDecoder();
    let fullText = '';
    let model = '';

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      const chunk = decoder.decode(value, { stream: true });
      // 解析 SSE data 行
      for (const line of chunk.split('\n')) {
        if (line.startsWith('data: ')) {
          try {
            const json = JSON.parse(line.slice(6));
            if (json.choices?.[0]?.delta?.content) {
              fullText += json.choices[0].delta.content;
            }
            if (json.model) model = json.model;
          } catch { /* 跳过非 JSON 行 */ }
        }
      }
    }
    return { text: fullText, model };
  } finally {
    clearTimeout(timer);
  }
}
```

#### 4.3.3 适配器注册表

```typescript
// apps/browser-extension/src/adapters/ai/index.ts

import { deepseekAdapter } from './deepseek';
import { qwenAdapter } from './qwen';
import { doubaoAdapter } from './doubao';
import { kimiAdapter } from './kimi';
import { ernieAdapter } from './ernie';
import { hunyuanAdapter } from './hunyuan';
import type { AIMonitorAdapter, AIPlatformId } from './types';

export const aiAdapters: Record<AIPlatformId, AIMonitorAdapter> = {
  deepseek: deepseekAdapter,
  qwen: qwenAdapter,
  doubao: doubaoAdapter,
  kimi: kimiAdapter,
  ernie: ernieAdapter,
  hunyuan: hunyuanAdapter,
};

export function getAIAdapter(platformId: AIPlatformId): AIMonitorAdapter | undefined {
  return aiAdapters[platformId];
}
```

### 4.4 采集任务调度器（插件侧）

```typescript
// apps/browser-extension/src/monitor/scheduler.ts

import { getAIAdapter } from '../adapters/ai';
import type { AIMonitorResult, AIPlatformId } from '../adapters/ai/types';
import { getState } from '../storage';

/** 反检测配置 */
const ANTI_DETECT = {
  maxQueriesPerPlatformPerHour: 3,
  minIntervalSec: 30,
  maxIntervalSec: 120,
  maxDailyQueries: 50,
};

interface CollectTask {
  id: number;
  question_text: string;
  platform_id: AIPlatformId;
  brand_id: number;
  question_id: number;
  search_enabled: boolean;
}

/** 执行一批采集任务 */
export async function executeCollectTasks(tasks: CollectTask[]): Promise<void> {
  const state = await getState();
  if (!state.installation_token || !state.api_base_url) return;

  // 按平台分组，串行执行（避免同时操作多个平台）
  const byPlatform = groupBy(tasks, t => t.platform_id);

  for (const [platformId, platformTasks] of Object.entries(byPlatform)) {
    const adapter = getAIAdapter(platformId as AIPlatformId);
    if (!adapter) continue;

    // 检测登录状态
    const detection = await adapter.detect();
    if (!detection.connected) {
      // 标记该平台全部任务为 skip
      await reportSkipped(state, platformTasks, '未登录');
      continue;
    }

    // 逐条执行，加入随机延迟
    for (const task of platformTasks) {
      // 检查日限额
      if (await getDailyCount() >= ANTI_DETECT.maxDailyQueries) break;
      // 检查平台小时限额
      if (await getPlatformHourlyCount(platformId) >= ANTI_DETECT.maxQueriesPerPlatformPerHour) break;

      const result = await adapter.ask(task.question_text, {
        searchEnabled: task.search_enabled,
        timeout: 60_000,
      });

      // 回传后端
      await reportResult(state, task, result);
      await incrementCounters(platformId);

      // 随机延迟
      const delay = randomBetween(ANTI_DETECT.minIntervalSec, ANTI_DETECT.maxIntervalSec);
      await sleep(delay * 1000);
    }
  }
}

async function reportResult(
  state: ExtensionStorageState,
  task: CollectTask,
  result: AIMonitorResult,
) {
  await fetch(`${state.api_base_url}/api/callback/plugin/monitor`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'X-Geo-Installation-Token': state.installation_token!,
      'X-Geo-Installation-Id': String(state.plugin_installation_id),
    },
    body: JSON.stringify({
      task_id: task.id,
      platform_id: task.platform_id,
      brand_id: task.brand_id,
      question_id: task.question_id,
      success: result.success,
      answer: result.answer,
      citations: result.citations,
      search_results: result.searchResults,
      response_time: result.responseTime,
      model: result.model,
      timestamp: result.timestamp,
      raw_html: result.rawHtml,
      client_version: browser.runtime.getManifest().version,
    }),
  });
}

async function reportSkipped(state: ExtensionStorageState, tasks: CollectTask[], reason: string) {
  await fetch(`${state.api_base_url}/api/callback/plugin/monitor/batch-skip`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'X-Geo-Installation-Token': state.installation_token!,
      'X-Geo-Installation-Id': String(state.plugin_installation_id),
    },
    body: JSON.stringify({
      task_ids: tasks.map(t => t.id),
      reason,
    }),
  });
}

function randomBetween(min: number, max: number): number {
  return Math.floor(Math.random() * (max - min + 1)) + min;
}

function sleep(ms: number): Promise<void> {
  return new Promise(resolve => setTimeout(resolve, ms));
}

function groupBy<T>(arr: T[], key: (item: T) => string): Record<string, T[]> {
  return arr.reduce((acc, item) => {
    const k = key(item);
    (acc[k] ??= []).push(item);
    return acc;
  }, {} as Record<string, T[]>);
}
```

### 4.5 采集执行策略

采用**插件自主 + 前端辅助**模式：

| 模式 | 触发方式 | 单次任务数 | 频率 | 依赖前端 |
| --- | --- | --- | --- | --- |
| **心跳自主采集（主通道）** | Background SW 定时心跳 | 3-5 个任务 | 每 15 分钟 | 否 |
| **用户主动采集（辅助）** | Dashboard 点击"立即采集" → postMessage | 批量 pending 任务 | 用户触发 | 是 |

心跳自主采集保证：**只要用户安装了插件且浏览器在运行，采集就在持续进行**，无需用户打开 Admin Web。

每日限额：单用户最多 50 个采集任务。

### 4.6 Background SW 扩展

#### 4.6.1 新增采集心跳（15 分钟间隔）

```typescript
// entrypoints/background.ts

// 原有 60 分钟心跳保持不变（平台检测 + 状态同步）
// 新增 15 分钟采集心跳
browser.alarms.create('monitor-heartbeat', { periodInMinutes: 15 });

browser.alarms.onAlarm.addListener(async (alarm) => {
  if (alarm.name === 'heartbeat') {
    // 原有逻辑：刷新内容平台登录状态
    await refreshPlatformState();
  }

  if (alarm.name === 'monitor-heartbeat') {
    // 新增：插件后台直接拉取并执行采集任务
    await pullAndExecuteMonitorTasks();
  }
});

async function pullAndExecuteMonitorTasks() {
  const state = await getState();
  if (!state.installation_token || !state.api_base_url) return;

  try {
    // 1. 直接调后端 API 拉取待采集任务（不经过前端）
    const resp = await fetch(`${state.api_base_url}/api/plugin/monitoring/tasks`, {
      headers: {
        'X-Geo-Installation-Token': state.installation_token,
        'X-Geo-Installation-Id': String(state.plugin_installation_id),
      },
    });
    if (!resp.ok) return;

    const { data: tasks } = await resp.json();
    if (!tasks?.length) return;

    // 2. 执行采集（串行，带反检测延迟）
    await executeCollectTasks(tasks);
  } catch {
    // 静默失败，下次心跳再试
  }
}
```

#### 4.6.2 前端辅助通道（用户主动触发）

```typescript
// 新增 action 处理（前端 postMessage 触发）
case 'collectMonitorData':
  // 用户点击"立即采集"时，前端传入任务列表
  return await handleCollectMonitorData(payload);

case 'detectAIPlatforms':
  // 检测 AI 平台登录状态（供前端展示）
  return await handleDetectAIPlatforms();
```

### 4.7 分布式任务调度（后端侧）

#### 4.7.1 任务生成

```go
// 每日凌晨由 Aggregator 或独立 CronJob 生成
func (s *MonitoringScheduler) GenerateDailyTasks(ctx context.Context, businessDate time.Time) error {
    // 1. 获取所有启用监测的租户
    tenants, _ := s.repo.ListMonitoringTenants(ctx)

    for _, tenant := range tenants {
        quota, _ := s.repo.GetMonitoringQuota(ctx, tenant.ID)

        // 检查采集频率
        if !shouldCollectToday(quota.CollectFrequency, businessDate) {
            continue
        }

        // 2. 获取租户的品牌和问题
        brands, _ := s.repo.ListBrandsWithQuestions(ctx, tenant.ID, quota.MaxBrands)

        for _, brand := range brands {
            for _, question := range brand.Questions {
                for _, platformID := range quota.EnabledPlatforms {
                    // 3. 生成任务（幂等，按唯一键去重）
                    s.repo.UpsertCollectTask(ctx, MonitorCollectTask{
                        TenantID:      tenant.ID,
                        BrandID:       brand.ID,
                        QuestionID:    question.ID,
                        PlatformID:    platformID,
                        BusinessDate:  businessDate,
                        RunMode:       "plugin_standard",
                        Status:        "pending",
                        SearchEnabled: quota.SearchEnhanced && platformSupportsSearch(platformID),
                    })
                }
            }
        }
    }
    return nil
}
```

#### 4.7.2 任务领取（分布式锁）

```go
// GET /api/tenant/monitoring/collect-tasks
func (s *MonitoringService) GetPendingTasks(ctx context.Context, tenantID uuid.UUID, limit int) ([]CollectTask, error) {
    // 只返回 status=pending 的任务
    return s.repo.ListPendingCollectTasks(ctx, tenantID, limit)
}

// POST /api/tenant/monitoring/collect-tasks/{id}/claim
func (s *MonitoringService) ClaimTask(ctx context.Context, taskID int64, installationID string) error {
    // Redis 分布式锁，TTL 10 分钟
    lockKey := fmt.Sprintf("mon:task_lock:%d", taskID)
    acquired, _ := s.cache.SetNX(ctx, lockKey, installationID, 10*time.Minute)
    if !acquired {
        return ErrTaskAlreadyClaimed
    }

    // 更新任务状态
    return s.repo.UpdateCollectTaskStatus(ctx, taskID, "assigned", installationID)
}
```

#### 4.7.3 结果接收（HTTP → RabbitMQ 异步处理）

插件回传的采集结果通过 RabbitMQ 异步处理，API 层只做校验和投递，**不阻塞插件等待解析完成**：

```go
// POST /api/callback/plugin/monitor
func (h *MonitorCallbackHandler) HandleMonitorResult(c *gin.Context) {
    var req MonitorResultRequest
    if err := c.ShouldBindJSON(&req); err != nil {
        c.JSON(400, gin.H{"error": "invalid request"})
        return
    }

    // 校验 installation_token
    installationToken := c.GetHeader("X-Geo-Installation-Token")
    if !h.service.ValidateInstallation(c, installationToken) {
        c.JSON(401, gin.H{"error": "unauthorized"})
        return
    }

    // 仅做两件事：标记任务 assigned → received，投递到 MQ
    // 解析、入库、MinIO 存储全部由 Consumer 异步完成
    if err := h.service.MarkTaskReceived(c, req.TaskID); err != nil {
        c.JSON(409, gin.H{"error": "task not found or already completed"})
        return
    }

    // 投递到 RabbitMQ
    if err := h.mq.Publish(c, "monitor.result", req); err != nil {
        // MQ 投递失败 → 回退任务状态，插件下次心跳会重试
        h.service.MarkTaskPending(c, req.TaskID)
        c.JSON(500, gin.H{"error": "queue unavailable"})
        return
    }

    c.JSON(200, gin.H{"ok": true})
}
```

### 4.8 RabbitMQ 异步处理架构

#### 4.8.1 队列拓扑

```
                    插件 HTTP 回传
                         │
                         ▼
┌──────────────────────────────────────────────────┐
│              tenant-api (HTTP 层)                  │
│  校验 token → 标记 received → publish to MQ       │
└───────────────────────┬──────────────────────────┘
                        │
                        ▼
┌──────────────────────────────────────────────────┐
│               RabbitMQ Broker                     │
│                                                   │
│  Exchange: monitor.direct (direct)                │
│  ┌─────────────────────────────────────────────┐ │
│  │ Queue: monitor.result.parse                  │ │
│  │   routing_key: monitor.result                │ │
│  │   consumer: MonitorParseWorker × 3           │ │
│  │   作用: 解析回答 + 入库 + MinIO 存储          │ │
│  └─────────────────────────────────────────────┘ │
│  ┌─────────────────────────────────────────────┐ │
│  │ Queue: monitor.task.generate                 │ │
│  │   routing_key: monitor.task.generate         │ │
│  │   consumer: TaskGenerateWorker × 1           │ │
│  │   作用: 凌晨批量生成采集任务                   │ │
│  └─────────────────────────────────────────────┘ │
│  ┌─────────────────────────────────────────────┐ │
│  │ Queue: monitor.aggregate.trigger             │ │
│  │   routing_key: monitor.aggregate             │ │
│  │   consumer: AggregateWorker × 1              │ │
│  │   作用: 按品牌触发增量聚合                     │ │
│  └─────────────────────────────────────────────┘ │
│  ┌─────────────────────────────────────────────┐ │
│  │ Queue: monitor.result.dlq (死信队列)          │ │
│  │   routing_key: monitor.result.dlq            │ │
│  │   作用: 处理失败的消息暂存，人工排查            │ │
│  └─────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────┘
```

#### 4.8.2 Parse Worker（核心消费者）

```go
// server/internal/monitoring/worker/parse_worker.go

type MonitorParseWorker struct {
    repo     *repository.MonitoringRepository
    parser   *MonitorAnswerParser
    storage  objectstorage.Client
    mq       *rabbitmq.Client
}

func (w *MonitorParseWorker) Handle(ctx context.Context, msg MonitorResultMessage) error {
    // 1. 存原始回答到 MinIO
    objectKey := fmt.Sprintf("monitor/%s/%d/%d.json",
        msg.Timestamp[:10], msg.TaskID, time.Now().UnixMilli())
    if err := w.storage.PutJSON(ctx, objectKey, msg); err != nil {
        return fmt.Errorf("minio put failed: %w", err)
    }

    // 2. 写 question_monitor_runs
    run, err := w.repo.CreateMonitorRun(ctx, CreateMonitorRunInput{
        TaskID:       msg.TaskID,
        PlatformID:   msg.PlatformID,
        BrandID:      msg.BrandID,
        QuestionID:   msg.QuestionID,
        RunMode:      "plugin_standard",
        RawAnswer:    msg.Answer,
        ResponseTime: msg.ResponseTime,
        Model:        msg.Model,
        ObjectKey:    objectKey,
    })
    if err != nil {
        return fmt.Errorf("create run failed: %w", err)
    }

    // 3. 解析回答（品牌提及、位置、情感、引用）
    parseResults := w.parser.Parse(ctx, run, msg.Answer, msg.Citations)
    if err := w.repo.SaveParseResults(ctx, run.ID, parseResults); err != nil {
        return fmt.Errorf("save parse results failed: %w", err)
    }

    // 4. 标记任务 completed
    if err := w.repo.CompleteTask(ctx, msg.TaskID); err != nil {
        return fmt.Errorf("complete task failed: %w", err)
    }

    // 5. 检查该品牌当日任务是否全部完成 → 触发增量聚合
    allDone, _ := w.repo.IsBrandDailyTasksComplete(ctx, msg.BrandID, msg.BusinessDate)
    if allDone {
        w.mq.Publish(ctx, "monitor.aggregate", AggregateMessage{
            BrandID:      msg.BrandID,
            TenantID:     msg.TenantID,
            BusinessDate: msg.BusinessDate,
        })
    }

    return nil
}
```

#### 4.8.3 Aggregate Worker（增量聚合）

```go
// server/internal/monitoring/worker/aggregate_worker.go

type AggregateWorker struct {
    repo  *repository.MonitoringRepository
    cache cache.Cache
}

func (w *AggregateWorker) Handle(ctx context.Context, msg AggregateMessage) error {
    // 按品牌 + business_date 增量聚合（替代 V3 的全量定时聚合）
    // 好处：品牌任务完成即聚合，不必等到每日 6:30
    if err := w.repo.AggregateBrandDaily(ctx, msg.TenantID, msg.BrandID, msg.BusinessDate); err != nil {
        return err
    }

    // 预热该品牌缓存
    w.cache.WarmBrandComposite(ctx, msg.TenantID, msg.BrandID)
    return nil
}
```

#### 4.8.4 RabbitMQ 配置

```yaml
# server/configs/config.yaml 新增
rabbitmq:
  url: "amqp://guest:guest@localhost:5672/"
  exchange: "monitor.direct"
  queues:
    result_parse:
      name: "monitor.result.parse"
      routing_key: "monitor.result"
      concurrency: 3          # 3 个并发消费者
      prefetch: 5             # 每个消费者预取 5 条
      retry_max: 3            # 最大重试 3 次
      retry_delay: "5s"       # 重试间隔
      dlq: "monitor.result.dlq"
    task_generate:
      name: "monitor.task.generate"
      routing_key: "monitor.task.generate"
      concurrency: 1
    aggregate:
      name: "monitor.aggregate.trigger"
      routing_key: "monitor.aggregate"
      concurrency: 1
      prefetch: 1
```

#### 4.8.5 Go 客户端初始化

```go
// server/internal/shared/mq/rabbitmq.go

import amqp "github.com/rabbitmq/amqp091-go"

type Client struct {
    conn    *amqp.Connection
    channel *amqp.Channel
    exchange string
}

func NewClient(url, exchange string) (*Client, error) {
    conn, err := amqp.Dial(url)
    if err != nil {
        return nil, err
    }
    ch, err := conn.Channel()
    if err != nil {
        return nil, err
    }

    // 声明 exchange
    err = ch.ExchangeDeclare(exchange, "direct", true, false, false, false, nil)
    if err != nil {
        return nil, err
    }

    return &Client{conn: conn, channel: ch, exchange: exchange}, nil
}

func (c *Client) Publish(ctx context.Context, routingKey string, body any) error {
    data, _ := json.Marshal(body)
    return c.channel.PublishWithContext(ctx, c.exchange, routingKey, false, false, amqp.Publishing{
        DeliveryMode: amqp.Persistent,
        ContentType:  "application/json",
        Body:         data,
        MessageId:    uuid.NewString(),
        Timestamp:    time.Now(),
    })
}

func (c *Client) Consume(queueName string, prefetch int, handler func(amqp.Delivery)) error {
    c.channel.Qos(prefetch, 0, false)
    msgs, err := c.channel.Consume(queueName, "", false, false, false, false, nil)
    if err != nil {
        return err
    }
    for msg := range msgs {
        handler(msg)
    }
    return nil
}
```

#### 4.8.6 异步化带来的架构优势

| 对比项 | 同步处理（V4 原方案） | RabbitMQ 异步 |
| --- | --- | --- |
| 插件回调延迟 | ~500ms（含解析+入库+MinIO） | ~10ms（仅校验+投递） |
| API Pod 负载 | 高（CPU 密集的解析在 API 进程） | 低（解析转移到 Worker） |
| 解析失败影响 | 插件收到 500，需重试整个采集 | 消息重试，不影响插件 |
| 聚合时机 | 每日 6:30 全量聚合 | **品牌级增量聚合**（任务完成即触发） |
| 数据可见延迟 | T+1（次日早上） | **准实时**（品牌任务完成后分钟级可见） |
| 峰值吸收 | API 直接承压 | MQ 削峰填谷 |
| 故障隔离 | 解析崩溃 → API 不可用 | Worker 崩溃 → 消息堆积 → 恢复后自动消化 |

### 4.8 DNR 规则扩展

在 `apps/browser-extension/public/rules.json` 中新增 AI 平台的跨域规则：

```json
[
  {
    "id": 101,
    "action": {
      "type": "modifyHeaders",
      "requestHeaders": [
        {"header": "Referer", "operation": "set", "value": "https://chat.deepseek.com"},
        {"header": "origin", "operation": "set", "value": "https://chat.deepseek.com"}
      ]
    },
    "condition": {
      "urlFilter": "https://chat.deepseek.com/*",
      "resourceTypes": ["xmlhttprequest"]
    }
  },
  {
    "id": 102,
    "action": {
      "type": "modifyHeaders",
      "requestHeaders": [
        {"header": "Referer", "operation": "set", "value": "https://tongyi.aliyun.com"},
        {"header": "origin", "operation": "set", "value": "https://tongyi.aliyun.com"}
      ]
    },
    "condition": {
      "urlFilter": "https://tongyi.aliyun.com/*",
      "resourceTypes": ["xmlhttprequest"]
    }
  },
  {
    "id": 103,
    "action": {
      "type": "modifyHeaders",
      "requestHeaders": [
        {"header": "Referer", "operation": "set", "value": "https://www.doubao.com"},
        {"header": "origin", "operation": "set", "value": "https://www.doubao.com"}
      ]
    },
    "condition": {
      "urlFilter": "https://www.doubao.com/*",
      "resourceTypes": ["xmlhttprequest"]
    }
  },
  {
    "id": 104,
    "action": {
      "type": "modifyHeaders",
      "requestHeaders": [
        {"header": "Referer", "operation": "set", "value": "https://kimi.moonshot.cn"},
        {"header": "origin", "operation": "set", "value": "https://kimi.moonshot.cn"}
      ]
    },
    "condition": {
      "urlFilter": "https://kimi.moonshot.cn/*",
      "resourceTypes": ["xmlhttprequest"]
    }
  },
  {
    "id": 105,
    "action": {
      "type": "modifyHeaders",
      "requestHeaders": [
        {"header": "Referer", "operation": "set", "value": "https://yiyan.baidu.com"},
        {"header": "origin", "operation": "set", "value": "https://yiyan.baidu.com"}
      ]
    },
    "condition": {
      "urlFilter": "https://yiyan.baidu.com/*",
      "resourceTypes": ["xmlhttprequest"]
    }
  },
  {
    "id": 106,
    "action": {
      "type": "modifyHeaders",
      "requestHeaders": [
        {"header": "Referer", "operation": "set", "value": "https://hunyuan.tencent.com"},
        {"header": "origin", "operation": "set", "value": "https://hunyuan.tencent.com"}
      ]
    },
    "condition": {
      "urlFilter": "https://hunyuan.tencent.com/*",
      "resourceTypes": ["xmlhttprequest"]
    }
  }
]
```

---

## 5. 数据口径修订（覆盖 V3 2.1）

### 5.1 `run_mode` 调整

| `run_mode` | 含义 | 进入主统计 | 默认消费页面 |
| --- | --- | --- | --- |
| `plugin_standard` | 插件在 AI Web 端标准对话采集 | 是 | Dashboard、平台占比、竞争对手、业务主题、AI 对话问题、品牌印象 |
| `plugin_search` | 插件在 AI Web 端开启联网搜索后采集 | 是，单独统计 | AI 引用排名 |
| `api_standard` | 后端 API 直接调用（降级/验证用） | 是 | 同 plugin_standard |

插件采集的回答更接近真实用户体验。采集可行性方案已确认 DeepSeek API 与 Web 版本不同，插件方案反而是优势。

### 5.2 Schema Delta 补充

在 V3 Schema Delta 基础上，新增/修改：

#### 5.2.1 `monitoring_collect_tasks` 表扩展

```sql
CREATE TABLE monitoring_collect_tasks (
    id                  BIGSERIAL PRIMARY KEY,
    tenant_id           UUID NOT NULL REFERENCES tenants(id),
    brand_id            INT NOT NULL,
    question_id         INT NOT NULL,
    question_version_id INT NOT NULL,
    ai_platform_id      VARCHAR(30) NOT NULL,
    run_mode            VARCHAR(30) NOT NULL DEFAULT 'plugin_standard',
    business_date       DATE NOT NULL,
    search_enabled      BOOLEAN NOT NULL DEFAULT false,
    status              VARCHAR(20) NOT NULL DEFAULT 'pending',
    assigned_to         VARCHAR(100),          -- 插件 installation_id
    assigned_at         TIMESTAMPTZ,
    completed_at        TIMESTAMPTZ,
    error_message       TEXT,
    retry_count         INT NOT NULL DEFAULT 0,
    priority            INT NOT NULL DEFAULT 0,
    created_at          TIMESTAMPTZ NOT NULL DEFAULT now(),

    CONSTRAINT uk_collect_task_idempotent
        UNIQUE (tenant_id, brand_id, question_id, ai_platform_id, run_mode, business_date)
);

CREATE INDEX idx_collect_tasks_pending
    ON monitoring_collect_tasks(tenant_id, status, business_date)
    WHERE status = 'pending';

COMMENT ON COLUMN monitoring_collect_tasks.status IS 'pending / assigned / completed / failed / expired / skipped';
COMMENT ON COLUMN monitoring_collect_tasks.assigned_to IS '领取该任务的插件 installation_id';
```

#### 5.2.2 `ai_platforms` 表

```sql
CREATE TABLE ai_platforms (
    id                       VARCHAR(30) PRIMARY KEY,
    name                     VARCHAR(100) NOT NULL,
    web_url                  VARCHAR(200) NOT NULL,
    supports_standard        BOOLEAN NOT NULL DEFAULT true,
    supports_search          BOOLEAN NOT NULL DEFAULT false,
    supports_citation        BOOLEAN NOT NULL DEFAULT false,
    plugin_adapter_available BOOLEAN NOT NULL DEFAULT false,
    display_order            INT NOT NULL DEFAULT 0,
    enabled                  BOOLEAN NOT NULL DEFAULT true,
    created_at               TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- 初始数据
INSERT INTO ai_platforms (id, name, web_url, supports_search, supports_citation, plugin_adapter_available, display_order) VALUES
('deepseek', 'DeepSeek', 'https://chat.deepseek.com', false, false, true, 1),
('qwen', '通义千问', 'https://tongyi.aliyun.com', true, true, true, 2),
('doubao', '豆包', 'https://www.doubao.com', true, true, true, 3),
('kimi', 'Kimi', 'https://kimi.moonshot.cn', true, true, false, 4),
('ernie', '文心一言', 'https://yiyan.baidu.com', true, true, false, 5),
('hunyuan', '混元', 'https://hunyuan.tencent.com', true, true, false, 6);
```

---

## 6. 读流量模型与缓存架构（保留 V3 不变）

**完全保留 V3 第 3 章的设计**，包括：

- 3.1 页面 Fan-out 矩阵
- 3.2 QPS 模型（13,500 设计容量 vs 实际 ~900 QPS，绰绰有余）
- 3.3 BFF 聚合层（Dashboard composite 接口）
- 3.4 缓存防雪崩（singleflight + TTL 抖动 + stale-while-revalidate + L1）
- 3.5 缓存 Key 修订（`scope` 改为对应 `plugin_standard`/`plugin_search`）
- 3.6 缓存失效策略（渐进式失效 + 预热）
- 3.7 TTL 矩阵

---

## 7. 部署拓扑与资源（修订 V3 第 5 章）

### 7.1 部署架构

```
                         ┌─────────────────────────────┐
                         │      CDN (静态资源)           │
                         └──────────────┬──────────────┘
                                        │
                         ┌──────────────▼──────────────┐
                         │   Nginx / SLB                │
                         │   限流: 每 IP 200 req/s       │
                         └──────────┬───────────────────┘
                                    │
              ┌─────────────────────┼─────────────────────┐
              ▼                     ▼                     ▼
       ┌────────────┐       ┌────────────┐       ┌────────────┐
       │ tenant-api  │       │ tenant-api  │       │ (HPA 弹性)  │
       │   Pod #1    │       │   Pod #2    │       │   Pod #3    │
       │ L1 cache    │       │ L1 cache    │       │             │
       └──┬────┬─────┘       └──┬────┬─────┘       └──┬────┬─────┘
          │    │                 │    │                 │    │
  查询读取 │    │ 投递消息        │    │                 │    │
          │    │                 │    │                 │    │
       ┌──▼────┴─────────────────▼────┴─────────────────▼────┴──┐
       │                Cache Redis (主从)                        │
       │   用途: API 缓存 + JWT 会话 + 任务分布式锁              │
       └────────────────────────┬────────────────────────────────┘
                                │
       ┌────────────────────────┼────────────────────────┬───────────────┐
       │                        │                        │               │
       ▼                        ▼                        ▼               ▼
┌─────────────┐         ┌─────────────┐         ┌─────────────┐  ┌──────────────┐
│ PG Primary  │────────▶│ PG Replica  │         │    MinIO     │  │  RabbitMQ    │
│  (读写)     │  流复制  │  (只读)     │         │  (原始回答)   │  │  (异步队列)   │
└─────────────┘         └─────────────┘         └─────────────┘  └──────┬───────┘
       ▲                        ▲                                        │
       │                        │                                        │
       │                        │                              消费消息   │
       │                        │                                        ▼
┌──────┴────────────────────────┴────────────────────────────────────────────┐
│              Worker 进程（与 API 独立部署）                                  │
│                                                                            │
│  ┌─────────────────────────────────────────────────────────┐              │
│  │ Monitor Worker (单进程，多 goroutine)                     │              │
│  │  • ParseWorker × 3     ← consume: monitor.result.parse  │              │
│  │  • AggregateWorker × 1 ← consume: monitor.aggregate     │              │
│  │  • TaskGenWorker × 1   ← consume: monitor.task.generate │              │
│  │  • 写 PG Primary / 读 PG Replica / 写 MinIO              │              │
│  └─────────────────────────────────────────────────────────┘              │
╚═══════════════════════════════════════════════════════════════════════════╝

            ┌─────────────────────────────────────┐
            │       浏览器插件 × 30,000（用户端）     │
            │  替代 V3 的 Collector Pod × 3         │
            │  采集结果通过 HTTP 回传 API             │
            │  API 投递到 RabbitMQ 异步处理           │
            └─────────────────────────────────────┘
```

**对比 V3 的变更**：
- 去掉 Collector Pod × 3（插件替代）
- 去掉 Queue Redis Sentinel → 替换为 **RabbitMQ**（功能更强：死信队列、重试、多队列路由）
- 新增 **Monitor Worker** 进程（消费 MQ 消息，执行解析/入库/聚合）
- V3 的 Aggregator 独立进程合并到 Monitor Worker 中（AggregateWorker 消费者）

### 7.2 资源规格

| 组件 | 实例数 | 规格 | 月成本（阿里云） |
| --- | --- | --- | --- |
| **tenant-api** | 2（HPA 2-4） | 4C8G | ¥2,400 |
| **Cache Redis** | 1 主 1 从 | 4G 内存 | ¥1,500 |
| **RabbitMQ** | 1（镜像队列可选主从） | 2C4G | ¥800 |
| **PG Primary** | 1 | 4C16G RDS | ¥2,500 |
| **PG Replica** | 1 | 2C8G 只读 | ¥1,000 |
| **MinIO / OSS** | 1 | 50G | ¥200 |
| **Monitor Worker** | 1 | 2C4G | ¥600 |
| **SLB + CDN** | - | - | ¥500 |
| **合计** | | | **¥9,500/月** |

### 7.3 连接池修订

```yaml
# 每个 API 实例
database:
  max_open_conns: 20

# 2 API 实例 → PG 总连接: 40
# 1 Monitor Worker → PG 连接: 15
# 合计: 55 连接
# PostgreSQL: max_connections = 100 (留余量)
```

---

## 8. 降级与容灾（扩展 V3 第 6 章）

### 8.1 采集侧降级

| 故障场景 | 降级措施 | 用户感知 |
| --- | --- | --- |
| 某 AI 平台改版导致 adapter 失效 | 该平台任务自动跳过，其他平台正常 | 该平台数据缺失，标记"暂不可用" |
| 插件安装率不足 / 用户不在线 | 未完成任务次日继续；Dashboard 展示已采集数据 | 部分平台数据延迟 1-2 天 |
| 采集完成率 < 70% | 告警通知；可选开启少量 API 采集补充 | 无感知（API 降级透明） |
| AI 平台风控封禁用户 | 降低该平台采集频率；通知用户重新登录 | 该平台数据暂停 |
| 全部 AI 平台不可用 | Dashboard 展示最后一次成功聚合数据 | 显示"数据更新时间" |

### 8.2 API 采集降级通道（可选）

保留 V3 的 API 采集能力作为降级通道。当插件采集完成率连续 3 天 < 70% 时，自动为该租户开启 API 采集：

```go
// 降级判断（每日聚合后执行）
func (s *MonitoringService) CheckCollectionCoverage(ctx context.Context, tenantID uuid.UUID) {
    rate := s.repo.GetCollectionCompletionRate(ctx, tenantID, 3) // 近 3 天
    if rate < 0.70 {
        s.enableAPIFallback(ctx, tenantID)
    }
}
```

### 8.3 读侧降级（保留 V3 不变）

V3 第 6 章的 Cache Redis 故障、PG 故障、Aggregator 异常等降级策略完全保留。

### 8.4 监控告警

| 指标 | 告警阈值 | 说明 |
| --- | --- | --- |
| API P99 延迟 | > 2s 持续 5 分钟 | 同 V3 |
| Cache Redis 命中率 | < 80% | 同 V3 |
| PG 连接池使用率 | > 80% | 同 V3 |
| 日采集完成率 | < 70% | **新增**，按平台分别告警 |
| 单平台 adapter 连续失败 | > 50 次/小时 | **新增**，可能平台改版 |
| 聚合 Job 未在 8:00 前完成 | 超时 | 同 V3 |
| 待采集任务堆积 | > 50,000 pending | **新增**，可能插件采集能力不足 |

---

## 9. 插件方案的风险与反检测

### 9.1 风险矩阵

| 风险 | 严重性 | 概率 | 对策 |
| --- | --- | --- | --- |
| AI 平台 Web 端改版 | 高 | 中 | adapter 独立维护，改版快速热更新；多平台冗余 |
| AI 平台检测自动化行为 | 高 | 中 | 控制频率 ≤ 3 次/平台/小时；模拟真实行为；随机延迟 |
| 用户未登录目标 AI 平台 | 中 | 高 | 跳过未登录平台；引导登录；多用户互补覆盖 |
| 采集数据质量不一致 | 中 | 中 | 同一问题多次采集取中位；异常过滤 |
| 采集延迟（依赖用户在线） | 中 | 中 | T+1 模型兼容延迟；次日继续 |

### 9.2 反检测策略

```typescript
const ANTI_DETECT_CONFIG = {
  // 频率控制
  maxQueriesPerPlatformPerHour: 3,
  maxDailyQueries: 50,

  // 时间随机化
  minIntervalSec: 30,
  maxIntervalSec: 120,

  // 行为模拟
  typeDelay: { min: 50, max: 150 },      // 逐字输入延迟 ms
  postAnswerDelay: { min: 2000, max: 5000 }, // 回答后停留 ms

  // 请求特征
  randomizeUserAgent: false,  // 使用浏览器真实 UA
  addRandomHeaders: false,    // 不添加额外头，避免特征
};
```

---

## 10. 压测验收标准（扩展 V3 第 7 章）

### 10.1 读侧压测（保留 V3 不变）

V3 第 7 章的 S1-S5 场景、SLA 指标完全保留。

### 10.2 新增：插件采集压测

| 场景 | 方法 | 目标 |
| --- | --- | --- |
| S7: 1000 并发插件实例同时回传结果 | 模拟 1000 个 POST /api/callback/plugin/monitor 并发 | P99 < 500ms，0 丢失 |
| S8: 任务领取竞争 | 500 个并发 claim 同一任务 | 只有 1 个成功，其余 409 |
| S9: 高频任务生成 | 生成 100K 采集任务 | 生成耗时 < 30s |

### 10.3 数据质量验证

| 验证项 | 方法 | 目标 |
| --- | --- | --- |
| 回答完整性 | 同一问题 3 次采集，对比回答长度 | 长度标准差 < 30% |
| 品牌提及准确率 | 人工标注 100 条回答，对比解析结果 | 准确率 > 90% |
| 引用提取率 | 有引用的平台，对比插件提取 vs 手动检查 | 召回率 > 85% |

---

## 11. 实施计划

### 11.1 Phase 分期

| Phase | 内容 | 估时 | 依赖 |
| --- | --- | --- | --- |
| A | 插件 AI 适配器框架 + 首批 3 平台（DeepSeek/千问/豆包） | 8 天 | 无 |
| B | 后端：Migration + 配额 + RabbitMQ 基础设施 + 采集回调 API + 插件认证 API | 8 天 | A（接口定义） |
| C | 后端：Monitor Worker（ParseWorker + AggregateWorker）+ 缓存层 + 查询 API | 8 天 | B |
| D | 前端：6 个数据追踪页面 + 采集控制面板 | 7 天 | C（API 就绪） |
| E | 第二批 3 个 AI 适配器（Kimi/文心/混元）+ 联调 | 5 天 | A + D |
| F | 压测 + 数据质量验证 + 验收 | 5 天 | E |
| **合计** | | **41 天** | |

Phase B 细分：
- RabbitMQ 客户端封装 + Exchange/Queue 声明（1 天）
- Migration（配额表 + 任务表 + AI 平台表 + Schema Delta）（2 天）
- 插件认证 API（`/api/plugin/monitoring/tasks`、`/claim`）（2 天）
- 采集回调 API（接收 → 投递 MQ）（1 天）
- TaskGenerateWorker（凌晨批量生成任务）（2 天）

Phase C 细分：
- ParseWorker（MQ 消费 → 解析 → 入库 → MinIO）（3 天）
- AggregateWorker（品牌级增量聚合）（2 天）
- 缓存层（singleflight + stale-while-revalidate + L1）（1 天）
- 查询 API（6 个页面 + composite）（2 天）

双人并行方案（前后端各 1 人）：

```
Week 1-2:  后端 Phase B (MQ + Migration + API) |  插件 Phase A (适配器)
Week 3-4:  后端 Phase C (Worker + 缓存 + 查询)  |  前端 Phase D (页面)
Week 5:    联调 Phase E                         |  第二批适配器
Week 6:    压测 Phase F
总计: 28 天
```

### 11.2 关键里程碑

| 时间点 | 里程碑 | 验收标准 |
| --- | --- | --- |
| Day 8 | 首个 AI 适配器可用 | DeepSeek 登录检测 + 提问 + 获取回答 E2E 通过 |
| Day 10 | RabbitMQ 基础设施就绪 | Exchange/Queue 声明完成，Publish/Consume 冒烟测试通过 |
| Day 16 | 后端采集链路通 | 插件回传 → MQ → ParseWorker → PG 入库全链路通过 |
| Day 24 | 查询 API + 增量聚合可用 | 品牌任务完成后分钟级数据可见 |
| Day 31 | 前端页面完成 | 6 个页面 + 采集面板可交互 |
| Day 36 | 全部 6 平台适配器可用 | 全平台检测 + 采集 E2E |
| Day 41 | 验收通过 | 读侧压测 SLA + MQ 吞吐压测 + 数据质量验证 |

---

## 12. V3 → V4 变更汇总

| 变更编号 | V3 方案 | V4 方案 | 原因 |
| --- | --- | --- | --- |
| **C1** | 服务端 AI API 采集（30 worker × 3 Pod） | 浏览器插件采集（30,000 用户端节点） | 消除 AI API 成本 |
| **C2** | Queue Redis Stream 任务队列 | RabbitMQ 异步处理（解析/入库/聚合） + PG 任务表 + Redis 分布式锁 | 插件 HTTP 回传 → MQ 异步解耦，API 不阻塞 |
| **C3** | 100 品牌 × 100 问题 | 配额限制（Free 1 品牌 / Pro 3 品牌，每品牌 40 问题） | 控制采集规模 |
| **C4** | 全品牌每日采集 | Free 每 3 天 / Pro 每天 | 降低采集压力 |
| **C5** | `run_mode`: api_standard / api_search_grounded | `run_mode`: plugin_standard / plugin_search | 采集来源变更 |
| **C6** | 月成本 ~¥42K（含 API ¥22K） | 月成本 ~¥9.5K（无 API 费用） | 成本优化 77% |
| **C7** | 无配额机制 | `tenant_monitoring_quotas` 租户级配额 | 支撑多租户规模化 |
| **C8** | 3 个 Collector Pod + Queue Redis | Collector → 插件替代；Queue Redis → RabbitMQ + Monitor Worker | 采集去中心化，处理异步化 |
| **C9** | 数据一致性高（固定 prompt/温度） | 数据更接近真实用户（Web 端采集） | 采集可行性方案确认 API ≠ Web |
| **新增** | 无 | 反检测策略 | 插件采集必须的安全措施 |
| **新增** | 无 | 采集控制面板（前端） | 用户管理采集进度和平台登录 |
| **新增** | 无 | API 降级通道 | 插件采集不足时的兜底方案 |
| **新增** | 无 | RabbitMQ 异步处理架构 | 解析/入库/聚合与 API 层解耦，插件回调 ~10ms 返回 |
| **新增** | 每日 6:30 全量定时聚合 | 品牌级增量聚合（MQ 触发） | 数据可见延迟从 T+1 降至准实时（分钟级） |
| **新增** | 无 | Monitor Worker 独立进程 | ParseWorker × 3 + AggregateWorker + TaskGenWorker |
| **新增** | 无 | 死信队列 (DLQ) | 处理失败的消息暂存，不丢数据 |