实战：把采集结果做成 API 与看板

采集到的数据如果只躺在数据库里，价值就只属于写脚本的人。把它包装成「一个能查的接口 + 一块能看的看板」，数据就能被团队复用：运营自己筛、分析师自己导、其他系统直接调。本章用 FastAPI 做最小可用接口，并补上缓存、部署与数据质量校验。

从表到服务只有三件事：查询接口（GET /items 支持筛选与分页，决定用 offset 还是游标分页、要不要鉴权）、看板（趋势图与明细表，决定静态导出还是实时查询）、更新节奏（定时任务频率与接口缓存 TTL 的配合）。经验规则是：接口只做查询，不做采集——采集与查询混在一个进程里，一次慢查询就能把采集节奏拖垮。

FastAPI 最小接口

# api.py —— Python 3.10+，依赖 fastapi、uvicorn、pydantic
import sqlite3
from typing import Optional
from fastapi import Depends, FastAPI, Header, HTTPException, Query
from pydantic import BaseModel, Field
app = FastAPI(title="采集数据查询接口", version="1.0")
DB_PATH = "price.db"
API_KEY = "change-me-in-production"          # 生产环境从环境变量或配置中心读取

class Item(BaseModel):
    """对外暴露的字段白名单：不要直接把数据库行原样返回"""
    sku_id: str = Field(..., description="SKU 标识")
    collect_date: str = Field(..., description="采集日期 YYYY-MM-DD")
    price: Optional[float] = Field(None, description="券后到手价，缺货为 null")
    in_stock: int = Field(1, description="1 在售 0 缺货")

def get_db():
    """每个请求一个连接，用完即关；高并发场景换成连接池"""
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
    finally:
        conn.close()

def verify_key(x_api_key: str = Header(default="")):
    if x_api_key != API_KEY:
        raise HTTPException(status_code=401, detail="无效的 API Key")

@app.get("/items", response_model=list[Item], dependencies=[Depends(verify_key)])
def list_items(
    sku_id: Optional[str] = None,
    start: Optional[str] = Query(None, description="起始日期，含"),
    end: Optional[str] = Query(None, description="结束日期，含"),
    page: int = Query(1, ge=1),
    size: int = Query(20, ge=1, le=200),     # 限制上限，防止 size=100000 拖垮服务
    db: sqlite3.Connection = Depends(get_db),
):
    """按 SKU 与日期区间筛选并分页；值全部走占位符，杜绝注入"""
    where, params = [], []
    if sku_id:
        where.append("sku_id = ?"); params.append(sku_id)
    if start:
        where.append("collect_date >= ?"); params.append(start)
    if end:
        where.append("collect_date <= ?"); params.append(end)
    clause = ("WHERE " + " AND ".join(where)) if where else ""
    rows = db.execute(
        f"SELECT sku_id, collect_date, price, in_stock FROM price_history {clause} "
        f"ORDER BY collect_date DESC, sku_id LIMIT ? OFFSET ?",
        (*params, size, (page - 1) * size)).fetchall()
    return [dict(row) for row in rows]

@app.get("/stats/sku/{sku_id}")
def sku_stats(sku_id: str, db: sqlite3.Connection = Depends(get_db)):
    """聚合视图：近 30 天最低价、最高价与采样点数，看板直接消费"""
    row = db.execute(
        "SELECT MIN(price) AS low, MAX(price) AS high, COUNT(*) AS samples "
        "FROM price_history WHERE sku_id = ? AND price IS NOT NULL "
        "AND collect_date >= date('now', '-30 day')", (sku_id,)).fetchone()
    if not row or row["samples"] == 0:
        raise HTTPException(status_code=404, detail="该 SKU 近 30 天无数据")
    return dict(row)

pip install fastapi uvicorn pydantic
uvicorn api:app --host 127.0.0.1 --port 8000 --workers 2
curl -s -H 'x-api-key: change-me-in-production' \
  'http://127.0.0.1:8000/items?sku_id=demo-1001&page=1&size=10'

response_model 是白名单，能把原始 HTML、内部备注挡在接口外；f-string 只拼接 WHERE 结构，值一律走占位符。部署用 systemd 托管：unit 里设 WorkingDirectory、Environment="API_KEY=..."、ExecStart=.../uvicorn api:app --host 127.0.0.1 --port 8000 --workers 2 与 Restart=always，然后 systemctl enable --now crawler-api；采集任务由 APScheduler 或 cron 单独调度，接口只读。

看板：静态导出、Streamlit 还是 Grafana

方案	优点	缺点	适用
定时导出 CSV/Excel	零运维、业务方熟悉	手动打开、无交互	每日/每周报表
Streamlit	十几行代码出交互页（`st.line_chart` + `st.dataframe`）	每次访问都跑查询，不适合高并发	内部小工具、分析师自助
Grafana + 时序库	图表专业、告警一体	需维护 Prometheus/MySQL 数据源	长期运行的运营看板
自建前端	完全定制	成本最高	对外产品

Grafana 的做法是把 MySQL 配成数据源，用 SELECT collect_date AS time, price FROM price_history WHERE sku_id = $sku 这样的查询变量做面板，并配合 Prometheus 的采集指标看「数据是否在按时更新」——两块看板同屏就能一眼看出采集是否正常。

缓存与更新频率

数据特征	更新频率	缓存策略
每日快照（价格、职位）	每天 1 次	接口加 5~10 分钟 TTL 或进程内 `lru_cache`
小时级指标	每小时	缓存 1 分钟，或读时物化到汇总表
实时明细	随采集更新	不缓存，但要用游标分页 + 索引
聚合统计	与明细同频	预计算成汇总表，接口只查汇总

别让接口每次请求都扫明细表做聚合，这是最常见的性能坑：把 MIN/MAX/COUNT 在采集完成后算一次写入 summary 表。

权限、限流与数据质量

鉴权：内部系统用固定 API Key（Header 校验）即可；对外或多角色场景改用 JWT/OAuth2，权限收敛到「能查哪些字段、哪些时间范围」。
限流：入口用 Nginx limit_req 按 IP 限流，或应用内用 slowapi 按 Key 限流；聚合接口设更严的配额，因为它最贵。
数据质量校验：当天数据条数低于近 7 日均值 30% 时返回警告并在看板标红，避免展示「采集已经坏了」的旧数据；判定逻辑应与第 22 章的告警共用同一份实现，口径必须一致。

校验项	规则	动作
条数突降	今日条数 < 近 7 日均值 × 30%	看板标红 + 告警
数据陈旧	最新 `collect_date` 距今 > 2 天	接口返回 `stale=true`
空值率	关键字段空值率 > 20%	标记质量异常，不进对外报表
口径混用	同一 SKU 同一天出现多个价格口径	拒绝返回，先修数据

常见坑

现象	原因	处理
接口返回了内部字段	直接 `SELECT *` 并原样返回	用 `response_model` 定义字段白名单
深分页越来越慢	大 `OFFSET` 扫描	改游标分页（按 `id > last_id`）
看板数据不是最新	缓存 TTL 过长或采集失败未告警	暴露数据时间戳并加质量校验
数据库连接被耗尽	每次请求新建连接且未关闭	用依赖注入的 `finally` 关闭或换连接池
采集慢查询拖垮接口	采集与查询共用库且无隔离	读写分离，或让采集走独立库

合规提示：把数据变成服务会显著放大合规责任——一旦对外提供接口，你就从「自己看数据」变成「对外提供数据」。只采集公开数据并遵守 robots.txt 与站点服务条款，不采集个人隐私信息，不绕过登录、验证码与付费墙；接口应做鉴权、限流与字段白名单，避免被批量拉取；对外提供或再分发前需确认《个人信息保护法》《数据安全法》《著作权法》相关要求并取得授权，优先使用官方 API 与开放数据。

小结：接口只做查询，用 FastAPI 加 pydantic 白名单和参数化 SQL，分页限制 size 上限并做鉴权限流；看板按场景在静态导出、Streamlit 与 Grafana 之间选；把聚合预计算成汇总表，并用「条数突降 + 数据时间戳」两道校验保证看板展示的永远是可信数据。

笔记加载中…

标签：#Python3 #爬虫

返回知识库