Kuaixun 快讯系统架构规范

一、系统概览

1.1 快讯 vs 数据管道：本质区别

二、当前架构深度分析

2.1 模块全景

kuaixun-service/
├── app/                       # FastAPI 后端（7 路由 + DAL）
│   ├── main.py                # 入口，端口 8100
│   ├── dal.py                 # 数据访问层（571 行 ⚠ 过大）
│   ├── database.py            # SQLAlchemy SessionLocal
│   ├── dependencies.py        # 鉴权 + WebSocket 管理器
│   └── routers/               # 6 个路由模块
│       ├── public.py          # /api/v1/news/live | summary | heat | tags
│       ├── external.py        # /api/v1/news | /search（需 X-API-Key）
│       ├── internal.py        # POST /batch | /score | /push
│       ├── admin.py           # API Key 管理
│       ├── collector.py       # 采集器 start/stop/restart
│       └── websocket.py       # ws://.../live 实时推送
├── common/                    # 公共基础设施
│   ├── config.py              # load_env()
│   └── security.py            # CORS + 限流 + 安全头
├── extend/                    # 扩展模块（核心管线）
│   ├── pipeline.py            # Pipeline 主循环（194 行）
│   ├── fetchers/              # 12 个数据源采集器
│   │   └── manager.py         # FetcherManager 调度器
│   ├── dedup/                 # 三层去重引擎
│   ├── notifier/              # 6 个推送渠道
│   ├── translator.py          # AI 翻译
│   ├── scorer.py              # AI 评分 0-5
│   ├── tagger.py              # AI 打标签
│   ├── summarizer.py          # AI 摘要
│   ├── content_enricher.py    # 原文抓取 + 背景搜索
│   └── bridge.py              # MySQL → TrendRadar 桥
└── tests/                     # 10 个测试文件

2.2 发现的关键问题

三、架构风格选择：事件驱动管道

3.1 为什么不用 Medallion 三层

快讯系统处理的是实时文本流，不是批量结构化数据。Medallion 三层的 Bronze→Silver→Gold 适合"采集后等一天再清洗"，但快讯需要采集后秒级推送。

3.2 事件驱动管道模型

flowchart LR
    subgraph Collect["采集层"]
        F[12 Fetchers] -- async gather --> Q1[(事件队列)]
    end
    subgraph Process["处理层"]
        Q1 --> D[去重
3 层引擎]
        D --> T[翻译]
        T --> S[AI 评分]
        S --> G[打标签]
        G --> M[AI 摘要]
        M --> E[内容增强]
    end
    subgraph Deliver["消费层"]
        E --> DB[(MySQL)]
        E --> WS[WebSocket]
        E --> PUSH[6 渠道推送]
    end
    style Collect fill:#eff6ff,stroke:#2563eb,stroke-width:2px
    style Process fill:#f5f3ff,stroke:#7c3aed,stroke-width:2px
    style Deliver fill:#f0fdf4,stroke:#059669,stroke-width:2px
  

3.3 与数据管道的共性——共享分层原则

虽然处理模型不同，但分层原则完全一致：采集层不关心处理，处理层不关心消费，依赖单向流动。

flowchart LR
    subgraph allow["依赖方向（采集层 → 处理层 → 消费层）"]
        C[采集层
fetchers/]
        P[处理层
dedup/ + AI 模块]
        D[消费层
API + WebSocket + Push]
    end
    C --> P --> D
    style allow fill:#f0fdf4,stroke:#059669,stroke-width:2px
  

四、依赖规则

当前合规：dedup/translator/scorer 等均不依赖 fetchers。

当前合规。

当前合规：DAL 只有 SQL 查询，无业务逻辑引用。

当前：pipeline.py 直接 import app.dal / app.database / extend.* 全栈引用。建议：定义 NewsProcessor Protocol，DAL 作为注入的 Repository。

当前违反：kuaixun-service 无 pyproject.toml，无 src/ 层。

五、核心组件设计契约

5.1 Fetcher（采集器）

5.2 DedupEngine（去重引擎）

5.3 AI 增强模块

5.4 DAL（数据访问层）

5.5 Notifier（推送渠道）

插件式架构已很合理（Base → 6 实现 + Manager），无需变动。新增渠道只需继承 PushSender。

六、数据流设计

sequenceDiagram
    participant Timer as Timer(30s)
    participant FM as FetcherManager
    participant Q as asyncio.Queue
    participant Pipe as Pipeline
    participant Dedup as DedupEngine
    participant AI as AI Modules
    participant DB as MySQL
    participant WS as WebSocket

    Timer->>FM: 触发采集
    FM->>FM: asyncio.gather(12 fetchers)
    FM->>Q: 放入原始 items
    
    loop Pipeline 循环
        Q->>Pipe: 取出 items
        Pipe->>Dedup: is_duplicate(item)
        Dedup-->>Pipe: True/False
        Pipe->>AI: translate → score → tag → summarize → enrich
        Pipe->>DB: batch_insert
        Pipe->>WS: broadcast(items)
    end
  

6.1 阶段解耦改进

当前 Pipeline.fetch_and_insert 中 8 步全部在一个 async 方法中串行调用。改进为 Queue 驱动的独立阶段：

# extend/pipeline.py — 改进后的结构
class Pipeline:
    def __init__(self):
        self.input_queue = asyncio.Queue(maxsize=1000)
        self._stages = [
            Stage("dedup", self._run_dedup, input_queue=self.input_queue),
            Stage("translate", self._run_translate),
            Stage("score", self._run_score),
            Stage("store", self._run_store),
        ]
    
    async def run_forever(self, interval=30):
        """采集 → 放入队列，各 Stage 独立消费"""
        asyncio.create_task(self._run_collector(interval))
        await asyncio.gather(*(s.run() for s in self._stages))

七、去重策略三层设计

7.1 当前状态

7.2 启用方案

在 Pipeline 的 fetch_and_insert 中，将 check_existing_ids 替换为 DedupEngine.is_duplicate()：

# 改进前（仅 L1 DB 查重）
existing = dal.check_existing_ids(table, ids)
new_items = [i for i in items if i["source_news_id"] not in existing]

# 改进后（三层查重）
dedup = DedupEngine()
new_items = []
for item in items:
    if not dedup.is_duplicate(item["source_code"], item["source_news_id"], item["title"]):
        new_items.append(item)

7.3 去重配置

七-B、news_unified 整合库设计

7B.1 架构对照

7B.2 数据流

flowchart TB
    subgraph Bronze["🥉 Bronze: 12 分表"]
        direction LR
        CLS[财联社] --- SINA[新浪] --- WS[华尔街见闻] --- EM[东方财富]
        J10[金十] --- GLH[格隆汇] --- K36[36氪] --- GN[GNews]
    end
    subgraph Worker["四层去重 Worker"]
        L1[L1: content_hash 精确]
        L2[L2: SimHash 指纹]
        L3[L3: Jaccard 语义]
        L4[L4: DB 查重兜底]
    end
    subgraph Silver["🥈 Silver: news_unified"]
        NU[(news_unified)]
        NU_DETAIL["source_list: JSON
同一事件多源合并
primary_source + primary_url"]
    end
    subgraph Gold["🥇 Gold: 消费层"]
        API[REST API]
        WS[WebSocket]
        PUSH[6渠道推送]
    end
    Bronze --> Worker --> NU --> API & WS & PUSH
    style Bronze fill:#eff6ff,stroke:#2563eb,stroke-width:2px
    style Silver fill:#f5f3ff,stroke:#7c3aed,stroke-width:2px
    style Gold fill:#f0fdf4,stroke:#059669,stroke-width:2px
  

7B.3 news_unified 表结构

CREATE TABLE news_unified (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    content_hash CHAR(32) NOT NULL,          -- L1 精确匹配键 MD5(title + 前200字)
    simhash BIGINT UNSIGNED DEFAULT 0,       -- L2 SimHash 64-bit 指纹
    title VARCHAR(512) NOT NULL,
    content TEXT,
    summary VARCHAR(256),                     -- AI 摘要（选择最高质量来源）
    source_list JSON NOT NULL,                -- ["cls","sina","wallstreetcn"] 多源合并
    source_count INT DEFAULT 1,               -- 来源数量（越多越重要）
    primary_source VARCHAR(32),               -- 主来源（最先采集到的）
    primary_url VARCHAR(1024),                -- 主来源原文链接
    tags JSON,                                -- AI 标签（跨源合并去重）
    importance INT DEFAULT 0,                 -- AI 评分 0-5（取最高分）
    published_at DATETIME,
    first_seen_at DATETIME,                   -- 首次采集时间
    last_seen_at DATETIME,                    -- 最近一次采集时间
    UNIQUE KEY uk_hash (content_hash),
    KEY idx_simhash (simhash),
    KEY idx_published (published_at),
    KEY idx_importance (importance)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

7B.4 四层去重算法

7B.5 窗口策略

去重时只对比 最近 24 小时的 news_unified 记录。3 天前的事件不会和今天重复，全量对比无意义且浪费性能。

7B.6 新增模块

kuaixun-service/
├── src/kuaixun_service/
│   ├── dedup.py              # 四层去重引擎（升级 DedupEngine）
│   └── unified_writer.py     # 新增：去重 → 写入 news_unified
└── sql/
    └── news_unified.sql      # 建表 DDL

八、AI 增强模块规范

8.1 统一配置管理

# extend/config.py — 统一配置
@dataclass
class AIConfig:
    provider: str = os.getenv("AI_PROVIDER", "ollama")
    model: str = os.getenv("AI_MODEL", "qwen2.5:7b")
    api_url: str = os.getenv("AI_API_URL", "http://localhost:11434")
    score_enabled: bool = os.getenv("AI_SCORE_ENABLED", "true") == "true"
    tag_enabled: bool = os.getenv("AI_TAGGING_ENABLED", "true") == "true"
    summary_enabled: bool = os.getenv("AI_SUMMARY_ENABLED", "true") == "true"

ai_config = AIConfig()  # 单例