# info_collection **Repository Path**: lmrybc49/info_collection ## Basic Information - **Project Name**: info_collection - **Description**: No description available - **Primary Language**: Unknown - **License**: Not specified - **Default Branch**: master - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 0 - **Forks**: 0 - **Created**: 2026-07-07 - **Last Updated**: 2026-07-08 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # 情报搜集工作台 > Intelligence Collection Workbench — 公开信息采集与基础研判层 ## 项目简介 情报搜集工作台承载"信息采集与基础研判层"。它面向运营人员，帮助用户围绕明确专题或持续运行的热点观察任务，完成**公开信息采集、证据留存、AI 初析、人工确认、报告生成和后续分流**。 ### 核心能力 | 能力 | 说明 | |------|------| | 专题采集 | 创建专题任务，围绕明确议题/对象/平台/时间范围采集公开信息 | | 热点观察 | 低成本持续监测，发现候选专题、异常升温和新增线索 | | 证据留存 | 报告结论可追溯到来源 URL、平台内容 ID、采集时间和证据片段 | | AI 初析 | 基于已入库证据，LLM 自动提取摘要、关键发现和实体 | | 人工确认 | 采集结果、AI 结论和报告内容均支持人工确认/驳回/标记补采 | | 报告生成 | 支持日报、周报、专题报告和风险线索报告（Markdown 输出） | | 分流交接 | 人工判断专题是否进入专题图谱增强研判或行动编排 | ### 技术栈 - **Web 框架**: FastAPI (异步) - **任务调度**: Celery + Redis - **数据库**: PostgreSQL 16 (SQLAlchemy 2.x 异步) - **配置管理**: Pydantic Settings - **通用采集**: HTTPX + Playwright + BeautifulSoup4 - **搜索引擎**: SearXNG (聚合搜索, 自部署) - **热搜发现**: ourongxing/newsnow - **平台适配**: PlatformAdapter 插件式架构 - **AI 初析**: LLM + JSON Schema (OpenAI 兼容) - **报告**: Jinja2 模板 → Markdown - **部署**: Docker Compose (6 服务: API, Worker, Beat, PostgreSQL, Redis, SearXNG) --- ## 快速开始 ### 前置条件 - Python 3.11+ - Docker & Docker Compose (推荐) - 或本地安装: PostgreSQL 16, Redis 7 ### 使用 Docker Compose 启动（推荐） ```bash # 1. 克隆项目 cd ownProject # 2. 复制环境变量模板 cp .env.example .env # 3. 编辑 .env，填入你的 LLM API Key 等配置 vim .env # 4. 启动所有服务 docker-compose up -d # 5. 验证启动 curl http://localhost:8000/health ``` 启动后访问： - **API 文档 (Swagger)**: http://localhost:8000/docs - **API 文档 (ReDoc)**: http://localhost:8000/redoc ### 本地开发启动 ```bash # 1. 安装依赖 pip install -r requirements.txt playwright install --with-deps chromium # 2. 配置环境变量 cp .env.example .env # 编辑 .env，将 INTEL_DB_HOST 指向你的本地 PostgreSQL # 3. 初始化数据库 python scripts/init_db.py # 4. 启动 API 服务 uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 # 5. (可选) 填充演示数据 python scripts/seed_demo.py ``` --- ## 项目架构 ``` ownProject/ ├── app/ # 主应用 │ ├── main.py # FastAPI 应用工厂 (lifespan, middleware, router) │ ├── config.py # Pydantic Settings 配置 (env 前缀: INTEL_) │ ├── database.py # async SQLAlchemy 引擎 + 会话工厂 │ ├── celery_app.py # Celery 实例 (Redis broker, 3 queues) │ │ │ ├── api/ # ---- 接口层 ---- │ │ ├── deps.py # 依赖注入 (get_db session) │ │ └── v1/ # API v1 │ │ ├── router.py # 路由聚合 (9 模块) │ │ ├── tasks.py # 采集任务 CRUD │ │ ├── sources.py # 采集入口管理 │ │ ├── collections.py # 采集执行触发 │ │ ├── evidence.py # 证据查看与确认 │ │ ├── leads.py # 线索管理与转换 │ │ ├── analysis.py # AI 初析触发与确认 │ │ ├── reports.py # 报告生成与发布 │ │ ├── handoffs.py # 分流交接记录 │ │ └── dashboard.py # 工作台首页统计 │ │ │ ├── models/ # ---- ORM 模型 (8 实体) ---- │ │ ├── base.py # DeclarativeBase + TimestampMixin │ │ ├── collection_task.py # 采集任务 (topic/watch) │ │ ├── collection_source.py # 采集入口 (account/keyword/url) │ │ ├── collection_run.py # 采集执行记录 │ │ ├── evidence_item.py # 证据材料 (含 url_hash 去重) │ │ ├── lead.py # 线索 (未确认的候选) │ │ ├── analysis_result.py # AI 初析结果 │ │ ├── report.py # 报告 (Markdown) │ │ └── handoff_record.py # 分流交接记录 │ │ │ ├── schemas/ # ---- Pydantic 请求/响应 ---- │ │ ├── common.py # PaginationParams, PageResponse, ErrorResponse │ │ ├── task.py # TaskCreate/Update/Response/Filter │ │ ├── source.py # SourceCreate/Update/Response │ │ ├── evidence.py # EvidenceResponse/Confirm/BatchAction │ │ ├── lead.py # LeadResponse/Confirm/Convert │ │ ├── analysis.py # AnalysisTrigger/Response/Confirm │ │ ├── report.py # ReportGenerate/Response/Update │ │ ├── handoff.py # HandoffCreate/Response │ │ └── dashboard.py # DashboardStats, PendingQueue │ │ │ ├── repositories/ # ---- 数据访问层 ---- │ │ ├── base.py # BaseRepository[T] 泛型基类 │ │ ├── task_repo.py # TaskRepository │ │ ├── source_repo.py # SourceRepository │ │ ├── run_repo.py # RunRepository │ │ ├── evidence_repo.py # EvidenceRepository │ │ ├── lead_repo.py # LeadRepository │ │ ├── analysis_repo.py # AnalysisRepository │ │ ├── report_repo.py # ReportRepository │ │ └── handoff_repo.py # HandoffRepository │ │ │ ├── services/ # ---- 业务逻辑层 ---- │ │ ├── task_service.py # 任务生命周期管理 │ │ ├── source_service.py # 采集入口管理 │ │ ├── collection_service.py # 采集编排 (分发到 Celery Worker) │ │ ├── evidence_service.py # 证据去重、落库、确认 │ │ ├── lead_service.py # 线索生成、确认、转专题 │ │ ├── analysis_service.py # AI 初析编排 (LLM 调用) │ │ ├── report_service.py # 报告组装、模板渲染 │ │ └── handoff_service.py # 分流记录管理 │ │ │ ├── collection/ # ---- 采集引擎 ---- │ │ ├── base.py # CollectionContext + BaseCollector │ │ ├── registry.py # CollectorRegistry (单例, 域名→适配器映射) │ │ ├── web_collector.py # WebCollector (HTTPX + Playwright 编排) │ │ ├── url_resolver.py # URL→平台识别 │ │ ├── dedup.py # DedupService (URL hash + origin_id) │ │ └── adapters/ # 平台适配器 (插件式) │ │ ├── base.py # PlatformAdapter ABC + CollectedItem │ │ └── generic.py # GenericWebAdapter (通用网页回退) │ │ │ ├── analysis/ # ---- AI 分析引擎 ---- │ │ ├── llm_client.py # LLMClient (OpenAI 兼容 + Tenacity 重试) │ │ ├── prompt_templates.py # 中文 Prompt 模板 (摘要/线索/实体) │ │ ├── schema_models.py # JSON Schema (Pydantic AnalysisOutput 等) │ │ ├── summarizer.py # Summarizer (证据→摘要+发现+实体) │ │ ├── entity_extractor.py # EntityExtractor (文本→结构化实体) │ │ └── lead_detector.py # LeadDetector (热点数据→候选线索) │ │ │ ├── reporting/ # ---- 报告生成 ---- │ │ ├── report_builder.py # ReportBuilder (Jinja2 渲染) │ │ ├── markdown_writer.py # MarkdownWriter (格式化工具) │ │ └── templates/ # Jinja2 模板 │ │ ├── topic.md.j2 # 专题报告模板 │ │ ├── daily.md.j2 # 日报模板 │ │ ├── weekly.md.j2 # 周报模板 │ │ └── risk_lead.md.j2 # 风险线索报告模板 │ │ │ ├── scheduler/ # ---- Celery 调度 ---- │ │ ├── tasks.py # 异步任务 (采集/分析/报告/发现) │ │ ├── beats.py # Beat 定时调度配置 │ │ └── triggers.py # 手动触发封装 │ │ │ ├── discovery/ # ---- 热点发现 ---- │ │ ├── searxng_client.py # SearXNG 聚合搜索客户端 │ │ ├── newsnow_client.py # NewsNow 热搜客户端 (12 平台) │ │ └── lead_generator.py # LeadGenerator (发现→线索) │ │ │ └── middleware/ # ---- 中间件 ---- │ ├── request_logging.py # 请求日志 (方法/路径/耗时/状态码) │ └── error_handler.py # 全局异常处理 (统一 JSON 错误响应) │ ├── docker/ # Docker 配置 │ ├── Dockerfile # 多阶段构建 (Playwright Chromium) │ ├── celery-worker-entrypoint.sh # Worker 启动脚本 │ └── celery-beat-entrypoint.sh # Beat 启动脚本 │ ├── scripts/ # 运维脚本 │ ├── init_db.py # 数据库初始化 │ └── seed_demo.py # 演示数据填充 │ ├── tests/ # 测试套件 │ ├── conftest.py # Pytest fixtures (async engine, test client) │ ├── test_api/ # API 集成测试 │ └── test_services/ # 业务逻辑单元测试 │ ├── alembic/ # 数据库迁移 ├── logs/ # 日志输出 (运行时) ├── reports/ # 报告输出 (运行时) ├── evidence_files/ # 证据附件存储 (运行时) │ ├── docker-compose.yml # 6 服务编排 ├── pyproject.toml # 项目配置 (ruff, mypy, pytest) ├── requirements.txt # Python 依赖 ├── .env.example # 环境变量模板 └── README.md # 本文档 ``` --- ## 分层架构设计 ``` ┌─────────────────────────────────────────┐ │ api/ (FastAPI路由) │ ← 接口层: 参数校验、HTTP 映射 ├─────────────────────────────────────────┤ │ services/ (业务逻辑) │ ← 服务层: 编排、事务管理 ├─────────────────────────────────────────┤ │ repositories/ (数据访问) │ ← 仓储层: SQL 封装 ├─────────────────────────────────────────┤ │ models/ (ORM模型) │ ← 数据层: 表映射 └─────────────────────────────────────────┘ ``` **依赖方向**: 上层 → 下层，**禁止反向依赖**。同层之间可横向调用（如 Service 调用多个 Repository）。 ### 核心设计决策 1. **Repository + Service 分层**: 数据访问与业务逻辑彻底解耦，Service 不直接写 SQL 2. **PlatformAdapter ABC**: 新增平台只需继承基类、实现 3 个方法，无需修改核心采集流程 3. **UUID 对外标识**: 所有 API 使用 `uuid` 作为资源标识符，防止 ID 枚举攻击 4. **三级容错**: 每个采集适配器支持 API → 浏览器自动化 → 直接 HTTP → 优雅降级 5. **Celery 任务队列**: 采集 (collection)、分析 (analysis)、默认 (default) 三队列隔离 ### 代码规范（遵循 BettaFish 标准） | 元素 | 规范 | |------|------| | 模块文档 | `"""中文标题\n负责具体职责"""` | | 类文档 | `"""类的一行中文描述"""` | | 方法文档 | 中文描述 + `Args:` + `Returns:` 章节 | | 内联注释 | 中文, `# ` 前缀, 描述下一段代码 | | 变量命名 | `snake_case` (英文) | | 类命名 | `PascalCase` (英文) | | 常量命名 | `UPPER_SNAKE_CASE` (英文) | | 类型标注 | 所有函数参数和返回值必须标注 | | 日志 | `from loguru import logger`, 中文消息, 可选 emoji | | 导入 | `from __future__ import annotations` 放在文件第一行 | | 异常处理 | 多层回退, 单次失败不阻断整体流程 | --- ## API 概览 | 模块 | 方法 | 路径 | 说明 | |------|------|------|------| | 采集任务 | `POST` | `/api/v1/tasks` | 创建任务 | | 采集任务 | `GET` | `/api/v1/tasks` | 任务列表 (支持筛选) | | 采集任务 | `GET` | `/api/v1/tasks/{uuid}` | 任务详情 | | 采集任务 | `PATCH` | `/api/v1/tasks/{uuid}` | 更新任务 | | 采集任务 | `POST` | `/api/v1/tasks/{uuid}/activate` | 激活任务 | | 采集入口 | `POST` | `/api/v1/tasks/{uuid}/sources` | 添加采集入口 | | 采集入口 | `GET` | `/api/v1/tasks/{uuid}/sources` | 入口列表 | | 采集执行 | `POST` | `/api/v1/tasks/{uuid}/collect` | 手动触发采集 | | 证据材料 | `GET` | `/api/v1/tasks/{uuid}/evidence` | 证据列表 | | 证据材料 | `POST` | `/api/v1/evidence/{uuid}/confirm` | 确认证据 | | 线索管理 | `GET` | `/api/v1/tasks/{uuid}/leads` | 线索列表 | | 线索管理 | `POST` | `/api/v1/leads/{uuid}/convert` | 线索转专题 | | AI 初析 | `POST` | `/api/v1/tasks/{uuid}/analyze` | 触发 AI 分析 | | 报告管理 | `POST` | `/api/v1/tasks/{uuid}/reports` | 生成报告 | | 报告管理 | `GET` | `/api/v1/reports/{uuid}` | 查看报告 (Markdown) | | 分流交接 | `POST` | `/api/v1/handoffs` | 创建交接记录 | | 工作台 | `GET` | `/api/v1/dashboard/stats` | 首页统计 | 完整 API 文档运行后访问: http://localhost:8000/docs --- ## 配置说明 所有配置通过 `.env` 文件或环境变量管理，前缀为 `INTEL_`。 ### 必填配置 | 环境变量 | 说明 | 示例 | |----------|------|------| | `INTEL_LLM_API_KEY` | LLM API 密钥 | `sk-xxxx` | | `INTEL_LLM_BASE_URL` | LLM API 地址 | `https://api.openai.com/v1` | | `INTEL_LLM_MODEL_NAME` | LLM 模型名 | `gpt-4o` | | `INTEL_DB_HOST` | 数据库主机 | `localhost` | | `INTEL_DB_PASSWORD` | 数据库密码 | `your_password` | ### 可选配置 | 环境变量 | 说明 | 默认值 | |----------|------|--------| | `INTEL_SEARXNG_BASE_URL` | SearXNG 地址 | `http://localhost:8080` | | `INTEL_COLLECTION_RATE_LIMIT_RPM` | 每分钟最大请求数 | `10` | | `INTEL_LLM_TEMPERATURE` | LLM 温度 | `0.3` | --- ## 开发指南 ### 添加新平台适配器 1. 在 `app/collection/adapters/` 下创建新文件（如 `weibo.py`） 2. 继承 `PlatformAdapter` 基类，实现所有抽象方法 3. 在 `app/collection/adapters/__init__.py` 中注册 ```python # weibo.py class WeiboAdapter(PlatformAdapter): @property def platform_name(self) -> str: return "weibo" @property def domain_patterns(self) -> list[str]: return ["weibo.com", "weibo.cn"] async def collect_by_account(self, account_url, source_config): # 实现微博账号采集逻辑 ... async def collect_by_keyword(self, keyword, source_config): # 实现微博关键词搜索逻辑 ... async def collect_by_url(self, content_url, source_config): # 实现微博单条内容采集逻辑 ... ``` ### 运行测试 ```bash pytest tests/ -v ``` ### 代码检查 ```bash ruff check app/ mypy app/ ``` --- ## 部署架构 ``` ┌──────────────────┐ │ Nginx / LB │ └────────┬─────────┘ │ :80/:443 ┌────────▼─────────┐ │ FastAPI API │ (api:8000) │ 主服务 │ └────────┬─────────┘ │ 分发任务 ┌────────────┼────────────┐ │ │ │ ┌────────▼──────┐ ┌──▼────────┐ ┌─▼──────────┐ │ Celery Worker │ │Celery Beat│ │ SearXNG │ │ (采集/分析) │ │(定时调度) │ │ (8080) │ └───────┬───────┘ └───────────┘ └────────────┘ │ ┌────────▼────────┐ ┌──────────────┐ │ PostgreSQL 16 │ │ Redis 7 │ │ (数据存储) │ │ (消息队列) │ └─────────────────┘ └──────────────┘ ``` --- ## 项目状态 - **版本**: 0.1.0（最小可试运行版本） - **开发周期**: 2026-07-07 开始 - **路线图**: 参见项目 PRD `docs/1-info-collect.html` ## 参考项目 本项目架构设计参考了 [BettaFish（微舆）](https://github.com/666ghj/BettaFish) 的以下方面： - 多引擎分层架构 (Insight/Media/Query/Report/Forum) - 中文代码规范 (文档注释 + 英文标识符) - PlatformAdapter 插件式设计模式 - 多层容错和优雅降级策略 - Celery 三队列任务隔离 详见 `docs/bettafish-analysis-report.md`。