实战：电商价格监控

价格监控是需求最明确、也最容易踩线的采集场景：技术上只是「定时抓一个数字」，但「能不能抓、抓完能做什么」决定了整个项目是否成立。本章先讲清边界，再讲 SKU 结构、价格口径、历史表设计与降价提醒。

先讲边界：三类用途，三种结论

用途	典型做法	合规前提
个人自用比价	本地记录自己关注的几件商品	仍需遵守站点条款与 `robots.txt`，控制频率
企业内部分析	比价用于选品、定价参考	需确认平台条款是否允许自动化抓取；优先用开放平台/官方 API 或授权数据服务
对外再分发	做成比价网站、公开发布或售卖	必须获得授权。商品数据通常有明确使用限制，商业再分发前应取得书面许可

三条硬红线：不绕过登录、验证码与风控；不采集需要登录或付费才能看到的价格；不做高频轮询式抓取。 价格数据还涉及《著作权法》下的商品图文、平台服务条款，以及商业宣传与竞争合规要求（可能涉及《反不正当竞争法》）；涉及个人信息处理须符合《个人信息保护法》《数据安全法》。工程上的第一选择永远是：能走官方开放平台或授权数据源，就不要自己抓。

价格落点：商品 → 规格 → SKU

一个商品页面是一棵「商品 → 规格 → SKU」的树：商品 ID（如 100012043978）是页面级标识，规格（颜色、容量、套餐）决定组合维度，SKU ID 才是价格与库存的真实载体，也是监控的最小单位。

价格形态	页面表现	处理方式
券前价（划线价）	`￥1,299`	单独存 `list_price`，不用于比价
券后到手价	`券后￥1,099`	存为 `price`，比价主口径
会员/新人价	需登录才可见	不采集，越过登录即越界
存储层的铁律：每个价格行必须带「口径」和「采集时间」，否则「昨天 1099、今天 1299」可能只是券前券后混着比；无货时写 `in_stock=0`，绝不写成价格 0。

历史价格表设计

-- 唯一键 = (sku_id, collect_date)：同一 SKU 同一天只留一条，补跑时可安全重写
CREATE TABLE price_history (
  id           BIGINT AUTO_INCREMENT PRIMARY KEY,
  sku_id       VARCHAR(64)   NOT NULL,
  collect_date DATE          NOT NULL COMMENT '采集日期，用于按天去重',
  price        DECIMAL(10,2) NULL COMMENT '券后到手价',
  list_price   DECIMAL(10,2) NULL COMMENT '券前/划线价',
  in_stock     TINYINT       NOT NULL DEFAULT 1 COMMENT '1 在售 0 缺货/下架',
  collect_time DATETIME      NOT NULL,
  UNIQUE KEY uk_sku_date (sku_id, collect_date)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

可运行的采集与提醒脚本

# price_monitor.py —— Python 3.10+，依赖 requests、beautifulsoup4
import json, re, sqlite3, time
from datetime import datetime
import requests
from bs4 import BeautifulSoup
SKUS = [{"sku_id": "demo-1001", "url": "https://example.com/item/1001"}]
INTERVAL = 3.0                      # 同一站点请求间隔，控制在保守区间
DDL = """CREATE TABLE IF NOT EXISTS price_history (
  id INTEGER PRIMARY KEY AUTOINCREMENT, sku_id TEXT NOT NULL, collect_date TEXT NOT NULL,
  price REAL, list_price REAL, in_stock INTEGER NOT NULL DEFAULT 1, collect_time TEXT NOT NULL,
  UNIQUE (sku_id, collect_date));"""
session = requests.Session()
session.headers.update({"User-Agent": "MyResearchBot/1.0 (+contact@example.com)"})
def parse_price(text: str) -> float | None:
    """「券后 ￥1,099.00 起」→ 数值；解析不到返回 None，绝不猜成 0"""
    cleaned = re.sub(r"[^\d.]", "", (text or "").replace(",", ""))
    try:
        return round(float(cleaned), 2) if cleaned else None
    except ValueError:
        return None
def crawl_one(sku: dict) -> dict | None:
    """抓取并解析单个商品页；失败返回 None，不把异常写进历史表"""
    try:
        resp = session.get(sku["url"], timeout=10)
    except requests.RequestException as exc:
        print(f"请求失败 {sku['url']}：{exc}")
        return None
    if resp.status_code != 200:
        print(f"[{resp.status_code}] {sku['url']}")
        return None
    soup = BeautifulSoup(resp.text, "html.parser")
    final, listed = soup.select_one("span.final-price"), soup.select_one("del.list-price")
    now = datetime.now()
    return {"sku_id": sku["sku_id"], "collect_date": now.strftime("%Y-%m-%d"),
            "price": parse_price(final.get_text(strip=True) if final else ""),
            "list_price": parse_price(listed.get_text(strip=True) if listed else ""),
            "in_stock": 0 if "缺货" in soup.get_text()[:2000] else 1,
            "collect_time": now.strftime("%Y-%m-%d %H:%M:%S")}
def upsert(conn: sqlite3.Connection, row: dict) -> None:
    """幂等写入：同一 SKU 同一天覆盖，重复运行不产生重复行"""
    conn.execute(
        "INSERT INTO price_history(sku_id, collect_date, price, list_price, in_stock, collect_time) "
        "VALUES(:sku_id, :collect_date, :price, :list_price, :in_stock, :collect_time) "
        "ON CONFLICT(sku_id, collect_date) DO UPDATE SET price=excluded.price, "
        "list_price=excluded.list_price, in_stock=excluded.in_stock", row)
    conn.commit()
def check_drop(conn: sqlite3.Connection, sku_id: str) -> str | None:
    """对比昨日价与近 7 天最低价；实用时要加最小降幅阈值与静默期"""
    rows = conn.execute(
        "SELECT price, in_stock FROM price_history WHERE sku_id=? AND price IS NOT NULL "
        "ORDER BY collect_date DESC LIMIT 8", (sku_id,)).fetchall()
    if len(rows) < 2:
        return None
    (today, in_stock), (yesterday, _) = rows[0], rows[1]
    if in_stock == 0:
        return f"{sku_id} 当前缺货，最近价格 {today} 元"
    if yesterday and today < yesterday:
        drop = round((yesterday - today) / yesterday * 100, 1)
        return f"{sku_id} 降价 {drop}%：{yesterday} → {today} 元"
    return None
if __name__ == "__main__":
    conn = sqlite3.connect("price.db")
    conn.executescript(DDL)
    alerts = []
    for sku in SKUS:
        row = crawl_one(sku)
        time.sleep(INTERVAL)
        if row is None:             # 抓取失败不写库，避免假数据污染趋势
            continue
        upsert(conn, row)
        if message := check_drop(conn, sku["sku_id"]):
            alerts.append(message)
    print(json.dumps({"alerts": alerts}, ensure_ascii=False))
    conn.close()

频率与并发建议

商品规模	采集频率与并发	说明
< 50 个 SKU	每天 1~2 次，单线程	典型自用规模，绝不值得高频轮询
50 ~ 5000 个 SKU	每天 1 次夜间分批，并发 1~4	需代理或降低单机频率，先确认条款允许

价格监控的价值来自趋势而非实时性：多数决策每天一次数据就够，把频率从「每分钟」降到「每天」，风险与成本会下降一个量级；量级再大（> 5000 个 SKU）就该改用官方开放平台或授权数据源。

常见坑

现象	原因	处理
价格忽高忽低	券前价与券后价混存	分列存 `price` 与 `list_price`，比价只用 `price`
缺货被记为 0 元	无价格时写入了 0	无价格写 NULL，缺货写 `in_stock=0`
降价提醒反复触发	与昨日比且小幅波动	加最小降幅阈值（如 ≥ 3%）与静默期

小结：价格监控的第一性问题是合规——自用比价、内部分析、对外再分发门槛完全不同，商业再分发必须取得授权，能走官方开放平台就不要自抓；技术侧把监控粒度定在 SKU，价格必须区分券前券后并记录采集时间，历史表用 (sku_id, collect_date) 唯一键保证幂等；提醒逻辑要加最小降幅与静默期，宁可少提醒也不要天天报假警。

笔记加载中…

标签：#Python3 #爬虫

返回知识库