geo/docs/ai-brand-monitoring-tech-design-v4.md

AI 品牌曝光监测系统技术方案 V4（旧版设计）

旧版设计说明：本文描述的是基于浏览器插件采集、installation_token、/api/plugin/monitoring/* 的旧监控方案。自 2026-04-20 起，当前实现已切换到 desktop_clients / desktop client 架构。本文保留用于工作记录与方案追溯，不再作为当前实现依据。

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

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 适配器接口

// 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 为例）：

// 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 适配器注册表

// 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 采集任务调度器（插件侧）

// 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 分钟间隔）

// 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 前端辅助通道（用户主动触发）

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

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

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

4.7.1 任务生成

// 每日凌晨由 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 任务领取（分布式锁）

// 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 层只做校验和投递，不阻塞插件等待解析完成：

// 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（核心消费者）

// 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（增量聚合）

// 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 配置

# 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 客户端初始化

// 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 平台的跨域规则：

[
  {
    "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 表扩展

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 表

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 连接池修订

# 每个 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 采集：

// 降级判断（每日聚合后执行）
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 反检测策略

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)	处理失败的消息暂存，不丢数据