kol-insight/CLAUDE.md

CLAUDE.md

This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.

交互规范

• 开始任务时说：主人,开始了
• 完成任务时说：主人,我干完了.您看看

项目概述

KOL Insight 是一个 KOL（关键意见领袖）数据查询与分析工具，用于批量查询达人视频数据并计算预估成本指标。

技术栈:

• 前端: Next.js 14.x (App Router) + React + TypeScript + Tailwind CSS
• 后端: Python FastAPI 0.104+ + SQLAlchemy 2.0+ (异步 ORM) + asyncpg
• 数据库: PostgreSQL 14.x+
• 部署: Docker + Uvicorn (ASGI 服务器)

常用命令

前端开发

# 安装依赖
pnpm install

# 开发模式（热重载）
pnpm dev

# 构建生产版本
pnpm build

# 生产模式运行
pnpm start

# 代码检查
pnpm lint

# 类型检查
pnpm type-check  # 如果配置了此脚本

后端开发

# 安装依赖
pip install -r requirements.txt
# 或使用 Poetry
poetry install

# 开发模式运行（热重载）
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

# 生产模式运行
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4

# 运行测试（TDD 必须）
pytest

# 运行测试并生成覆盖率报告
pytest --cov=app --cov-report=html

# 运行特定测试文件
pytest tests/test_query_service.py

# 运行特定测试函数
pytest tests/test_query_service.py::test_query_by_star_id

数据库操作

# 连接数据库（使用 .env 中的连接字符串）
psql "postgresql://user:password@host:5432/yuntu_kol"

# 创建迁移
alembic revision --autogenerate -m "description"

# 执行迁移
alembic upgrade head

# 回滚迁移
alembic downgrade -1

Docker 部署

# 构建并启动所有服务（前端、后端、数据库）
docker-compose up -d

# 查看日志
docker-compose logs -f

# 停止所有服务
docker-compose down

# 重新构建并启动
docker-compose up -d --build

架构设计

前后端分离架构

┌─────────────────────┐
│   Next.js 前端      │  端口: 3000
│   (纯前端渲染)      │
└──────────┬──────────┘
           │ HTTP API 调用
           ▼
┌─────────────────────┐
│   FastAPI 后端      │  端口: 8000
│   (异步 API)        │
└──────────┬──────────┘
           │ asyncpg
           ▼
┌─────────────────────┐
│   PostgreSQL        │  端口: 5432
└─────────────────────┘

前后端分离的关键点:

• 前端通过 HTTP 调用后端 API（需要配置 CORS）
• 前端环境变量: NEXT_PUBLIC_API_URL 指向后端地址
• 后端环境变量: CORS_ORIGINS 配置允许的前端域名
• 独立部署：前端可部署到 Vercel，后端部署到 Docker

核心模块

1. 查询模块 (backend/app/services/query_service.py)

   • 支持三种查询方式：星图ID（star_id）、达人ID（star_unique_id）、昵称（star_nickname）
   • 星图ID 和达人ID 使用精准匹配（WHERE IN）
   • 昵称使用模糊匹配（WHERE LIKE '%昵称%'）
   • 单次查询限制最大 1000 条

2. 计算模块 (backend/app/services/calculator.py)

   • 预估自然CPM: (estimated_video_cost / natural_play_cnt) * 1000
   • 预估自然看后搜人数: (natural_play_cnt / total_play_cnt) * after_view_search_uv
   • 预估自然看后搜人数成本: estimated_video_cost / 预估自然看后搜人数
   • 除零检查：分母为 0 时返回 None
   • 结果保留 2 位小数：使用 round(value, 2)

3. 品牌API集成模块 (backend/app/services/brand_api.py)

   • 批量并发调用：从查询结果提取唯一 brand_id，批量调用品牌API
   • 并发控制：使用 asyncio.Semaphore 限制最大 10 个并发请求
   • 超时设置：单个请求超时 3 秒
   • 降级策略：API 调用失败时显示原始 brand_id
   • 技术实现：使用 httpx.AsyncClient + asyncio.gather

4. 导出模块 (backend/app/services/export_service.py)

   • 支持 Excel (使用 openpyxl 或 xlsxwriter) 和 CSV 格式
   • 使用中文列名作为表头
   • 限制单次导出最大 1000 条

数据流向

用户输入查询条件 (前端)
    ↓
POST /api/v1/query (后端)
    ↓
查询数据库 (SQLAlchemy 异步)
    ↓
提取唯一 brand_id → 批量调用品牌API (httpx 异步，并发10)
    ↓
填充品牌名称 → 计算预估指标
    ↓
返回完整数据 (JSON)
    ↓
前端展示结果表格
    ↓
用户点击导出 → GET /api/v1/export → 下载 Excel/CSV

数据库设计

KolVideo 模型

class KolVideo(Base):
    __tablename__ = "kol_videos"

    # 主键
    item_id = Column(String, primary_key=True)

    # 查询字段（必须建立索引）
    star_id = Column(String, nullable=False, index=True)           # 星图ID
    star_unique_id = Column(String, nullable=False, index=True)    # 达人unique_id
    star_nickname = Column(String, nullable=False, index=True)     # 达人昵称

    # 基础信息
    title = Column(String, nullable=True)
    viral_type = Column(String, nullable=True)
    video_url = Column(String, nullable=True)
    publish_time = Column(DateTime, nullable=True)

    # 曝光指标（用于计算）
    natural_play_cnt = Column(Integer, default=0)      # 自然播放量
    heated_play_cnt = Column(Integer, default=0)       # 加热播放量
    total_play_cnt = Column(Integer, default=0)        # 总播放量

    # 互动指标
    total_interact = Column(Integer, default=0)
    like_cnt = Column(Integer, default=0)
    share_cnt = Column(Integer, default=0)
    comment_cnt = Column(Integer, default=0)

    # 效果指标（用于计算）
    new_a3_rate = Column(Float, nullable=True)
    after_view_search_uv = Column(Integer, default=0)  # 看后搜人数
    return_search_cnt = Column(Integer, default=0)

    # 商业信息
    industry_id = Column(String, nullable=True)
    industry_name = Column(String, nullable=True)
    brand_id = Column(String, nullable=True)           # 用于调用品牌API
    estimated_video_cost = Column(Float, default=0)    # 预估视频成本（用于计算）

索引策略:

• idx_star_id: 星图ID 精准查询
• idx_star_unique_id: 达人ID 精准查询
• idx_star_nickname: 昵称模糊查询（需要支持 LIKE）

API 设计

POST /api/v1/query

批量查询 KOL 视频数据。

请求体:

{
    "type": "star_id" | "unique_id" | "nickname",  # 查询方式
    "values": ["id1", "id2", ...]                  # 批量ID或单个昵称
}

响应体:

{
    "success": true,
    "data": [
        {
            "item_id": "...",
            "title": "...",
            # ... 26个字段
            "brand_name": "...",              # 后端填充
            "estimated_natural_cpm": 12.34,   # 后端计算
            # ...
        }
    ],
    "total": 100
}

GET /api/v1/export

导出查询结果。

查询参数:

• format: xlsx 或 csv

响应: 文件下载（Content-Disposition: attachment）

开发规范

TDD（测试驱动开发）要求

这个项目强制使用 TDD，必须先写测试再写实现代码。

1. 单元测试覆盖率要求: 100%（所有分支覆盖）
2. 集成测试要求: 使用真实数据库连接（.env 中的连接字符串）
3. 测试框架: pytest + pytest-cov
4. 测试文件位置: backend/tests/

TDD 流程:

1. 编写测试用例 (tests/test_*.py)
2. 运行测试（应该失败）
3. 编写最小实现代码
4. 运行测试（应该通过）
5. 重构代码（保持测试通过）
6. 生成覆盖率报告验证 100% 覆盖

测试示例:

# tests/test_calculator.py
def test_calculate_natural_cpm():
    # 正常情况
    result = calculate_natural_cpm(1000, 50000)
    assert result == 20.0

    # 除零情况
    result = calculate_natural_cpm(1000, 0)
    assert result is None

# tests/test_query_service.py
@pytest.mark.asyncio
async def test_query_by_star_id(db_session):
    # 使用真实数据库连接
    result = await query_videos(
        db_session,
        query_type="star_id",
        values=["test_id_1", "test_id_2"]
    )
    assert len(result) > 0

异步编程规范

后端必须使用异步编程以提升性能。

1. 数据库操作: 使用 SQLAlchemy 异步 API

   from sqlalchemy.ext.asyncio import AsyncSession
   
   async def query_videos(session: AsyncSession, ...):
       stmt = select(KolVideo).where(...)
       result = await session.execute(stmt)
       return result.scalars().all()
   

2. 外部API调用: 使用 httpx.AsyncClient

   async with httpx.AsyncClient(timeout=3.0) as client:
       response = await client.get(url)
   

3. 并发控制: 使用 asyncio.Semaphore

   semaphore = asyncio.Semaphore(10)  # 限制10并发
   async with semaphore:
       # 执行异步任务
   

前端实现规范

前端采用"粗略实现"策略：重点在功能可用，样式可简化。

1. 组件化开发: 使用 React 组件（QueryForm, ResultTable, ExportButton）
2. API 调用封装: 在 lib/api.ts 中统一封装
3. 环境变量: 使用 NEXT_PUBLIC_API_URL 配置后端地址
4. 状态管理: 页面状态包括：默认态、输入态、查询中、结果态、空结果态、错误态

错误处理规范

1. 后端:

   • 所有 API 路由使用 try-except 包裹
   • 数据库连接失败、品牌API超时等场景有降级处理
   • 错误信息记录到日志

2. 前端:

   • 网络错误显示用户友好提示
   • 空结果显示引导文案

性能要求

• 查询响应时间 ≤ 3 秒（100 条数据）
• 页面加载时间 ≤ 2 秒
• 导出响应时间 ≤ 5 秒（1000 条数据）
• 品牌 API 并发限制: 10 个请求，单请求超时 3 秒

环境变量配置

前端 (.env.local)

NEXT_PUBLIC_API_URL=http://localhost:8000/api/v1

后端 (.env)

DATABASE_URL=postgresql://user:password@host:5432/yuntu_kol
CORS_ORIGINS=http://localhost:3000,https://your-frontend-domain.com
BRAND_API_BASE_URL=https://api.internal.intelligrow.cn

目录结构

kol-insight/
├── frontend/                    # 前端项目（Next.js）
│   ├── src/
│   │   ├── app/                 # App Router 路由
│   │   ├── components/          # React 组件
│   │   ├── lib/                 # API 调用、工具函数
│   │   └── types/               # TypeScript 类型定义
│   └── package.json
│
├── backend/                     # 后端项目（FastAPI）
│   ├── app/
│   │   ├── main.py              # FastAPI 应用入口
│   │   ├── config.py            # 配置管理
│   │   ├── database.py          # 数据库连接
│   │   ├── models/              # SQLAlchemy 模型
│   │   ├── schemas/             # Pydantic 请求/响应模型
│   │   ├── api/v1/              # API 路由
│   │   └── services/            # 业务逻辑
│   ├── tests/                   # 测试文件（TDD 必须）
│   └── requirements.txt
│
├── doc/                         # 项目文档
│   ├── PRD.md                   # 产品需求文档
│   ├── FeatureSummary.md        # 功能摘要
│   ├── UIDesign.md              # UI 设计
│   ├── DevelopmentPlan.md       # 开发计划
│   └── tasks.md                 # 任务列表
│
├── docker-compose.yml           # Docker 编排
└── README.md

关键技术决策

为什么使用 FastAPI？

• 原生支持异步编程（async/await）
• 自动生成 OpenAPI 文档（Swagger UI）
• Pydantic 类型验证，类型安全
• 性能优异（基于 Starlette 和 Pydantic）

为什么使用前后端分离？

• 前端可独立部署到 Vercel 等平台
• 后端可独立扩展和优化
• 职责分离：前端专注 UI，后端专注业务逻辑
• 便于团队协作（前端/后端可并行开发）

为什么强制 TDD + 100% 覆盖率？

• 保证代码质量和可维护性
• 避免回归问题
• 文档化代码行为（测试即文档）
• 便于重构（测试作为安全网）

为什么品牌API在后端调用？

• 避免前端暴露内部API地址
• 统一错误处理和降级逻辑
• 利用后端异步能力优化并发性能
• 减少前端复杂度

常见问题

Q: 如何运行单个测试？

pytest tests/test_calculator.py::test_calculate_natural_cpm -v

Q: 如何查看测试覆盖率？

pytest --cov=app --cov-report=html
# 打开 htmlcov/index.html 查看详细报告

Q: 前端如何调用后端API？

在 lib/api.ts 中封装：

const API_BASE_URL = process.env.NEXT_PUBLIC_API_URL || 'http://localhost:8000/api/v1';

export async function queryVideos(request: QueryRequest): Promise<QueryResponse> {
  const response = await fetch(`${API_BASE_URL}/query`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(request),
  });
  return response.json();
}

Q: 如何调试品牌API批量调用？

1. 检查后端日志（应该记录API调用状态）
2. 使用断点调试 services/brand_api.py
3. 验证并发控制和超时设置是否生效

Q: 数据库索引如何验证？

-- 连接数据库后执行
\d kol_videos
-- 应该看到 idx_star_id, idx_star_unique_id, idx_star_nickname