Bright Data 商品价格监控：自动抓取、SQLite 记录与降价提醒

用 Python 做一个商品价格监控：定时抓取商品页面，解析价格写入 SQLite，价格低于你设定的目标价或创下历史新低时提醒你。抓取层接入 Bright Data Web Unlocker API，遇到反爬页面时由它负责重试和解封。

先看清成品

成品是一个命令行工具：python watch.py 检查 products.json 里的全部商品，python watch.py --history 查看历史价格。本文代码已实际运行：3 个测试通过；未配置 Bright Data 时直接抓取练习站 books.toscrape.com，成功解析出两本书的价格（51.77、53.74，练习站的虚构价格）并写入数据库。本文环境没有 Bright Data 账号，经过 Web Unlocker API 的请求没有实际发送，请求格式按官方文档编写。

注意：采集前先读目标网站的服务条款和 robots.txt，只采集公开、非个人的数据，控制频率，不绕过登录和付费墙，不采集个人信息。本文示例使用专门供练习的网站 toscrape.com。用于商业用途前请咨询法律意见，并遵守所在地关于数据和个人信息保护的法律。

做完之后你会得到什么？

需要准备哪些环境？

第一步：怎样在 Bright Data 准备 Web Unlocker？

1. 创建区域（zone）

登录控制台，添加一个 Web Unlocker API 区域，名称例如 price_unlocker。

2. 复制 API 密钥

区域的 Overview 标签页里有 API 密钥和请求示例。官方建议给 API 密钥设置过期时间，密钥只显示一次，要立即保存。

3. 先用 curl 测一次

curl -H "Content-Type: application/json" -H "Authorization: Bearer 你的API密钥" \
  -d '{"zone":"price_unlocker","url":"https://books.toscrape.com/","format":"raw"}' \
  https://api.brightdata.com/request

返回网页 HTML 就说明区域可用。官方文档列出的常见错误：401 是密钥不对，400 是缺少 zone、url、format。

第二步：代码怎么写？

1. 项目结构

price-watch/
├── fetch.py          # 抓取层
├── watch.py          # 解析、存储、提醒
├── products.json     # 监控清单
├── test_watch.py
└── prices.db         # 运行后生成

2. fetch.py

"""统一的抓取函数：配置了 Bright Data 就走 Web Unlocker API，否则直接请求（只用于练习站点）。"""
import os

import requests

API = "https://api.brightdata.com/request"


def fetch(url: str, timeout: int = 60) -> str:
    key, zone = os.environ.get("BRIGHTDATA_API_KEY"), os.environ.get("BRIGHTDATA_ZONE")
    if key and zone:
        r = requests.post(
            API,
            headers={"Authorization": f"Bearer {key}"},
            json={"zone": zone, "url": url, "format": "raw"},
            timeout=timeout,
        )
    else:
        r = requests.get(url, timeout=timeout, headers={"User-Agent": "price-watch-tutorial/1.0"})
    r.raise_for_status()
    r.encoding = r.apparent_encoding if r.encoding in (None, "ISO-8859-1") else r.encoding
    return r.text

3. watch.py

"""价格监控：抓取商品页 -> 解析价格 -> 写入 SQLite -> 低于目标价时提醒。
用法：python watch.py            # 检查 products.json 里的全部商品
      python watch.py --history  # 查看历史记录
"""
import argparse
import json
import re
import sqlite3
import sys
import time
from datetime import datetime
from pathlib import Path

from bs4 import BeautifulSoup

from fetch import fetch

DB = Path(__file__).with_name("prices.db")


def parse_price(html: str, selector: str) -> tuple[str, float]:
    soup = BeautifulSoup(html, "html.parser")
    title = (soup.select_one("h1") or soup.title).get_text(strip=True)
    node = soup.select_one(selector)
    if node is None:
        raise ValueError(f"页面里找不到价格元素 {selector}，可能页面结构变了或被拦截")
    m = re.search(r"\d+(?:[.,]\d{1,2})?", node.get_text().replace(",", ""))
    if not m:
        raise ValueError(f"价格元素里没有数字：{node.get_text(strip=True)!r}")
    return title, float(m.group())


def init(db):
    db.execute("CREATE TABLE IF NOT EXISTS price (url TEXT, title TEXT, price REAL, checked_at TEXT)")


def save(db, url, title, price):
    db.execute("INSERT INTO price VALUES (?, ?, ?, ?)", (url, title, price, datetime.now().isoformat(timespec="seconds")))
    db.commit()


def lowest(db, url):
    row = db.execute("SELECT min(price) FROM price WHERE url = ?", (url,)).fetchone()
    return row[0]


def check(products, db, delay=3.0):
    init(db)
    alerts = []
    for i, p in enumerate(products):
        if i:
            time.sleep(delay)  # 控制频率，对目标网站友好
        try:
            title, price = parse_price(fetch(p["url"]), p["selector"])
        except Exception as e:  # 单个商品失败不影响其他
            print(f"✗ {p['url']}：{e}")
            continue
        before = lowest(db, p["url"])
        save(db, p["url"], title, price)
        flag = ""
        if price <= p["target"]:
            flag = f"  ← 低于目标价 {p['target']}"
            alerts.append((title, price, p["url"]))
        elif before is not None and price < before:
            flag = f"  ← 历史新低（之前最低 {before}）"
        print(f"✓ {title}：{price}{flag}")
    return alerts


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--history", action="store_true")
    a = ap.parse_args()
    db = sqlite3.connect(DB)
    init(db)
    if a.history:
        for row in db.execute("SELECT checked_at, title, price FROM price ORDER BY checked_at"):
            print(*row, sep="  ")
        return 0
    products = json.loads(Path(__file__).with_name("products.json").read_text(encoding="utf-8"))
    alerts = check(products, db)
    if alerts:
        print("\n需要关注：")
        for t, pr, u in alerts:
            print(f"- {t} 现价 {pr}：{u}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

4. products.json

[
  {
    "url": "https://books.toscrape.com/catalogue/a-light-in-the-attic_1000/index.html",
    "selector": "p.price_color",
    "target": 50
  },
  {
    "url": "https://books.toscrape.com/catalogue/tipping-the-velvet_999/index.html",
    "selector": "p.price_color",
    "target": 40
  }
]

用浏览器打开商品页，右键价格选择“检查”，就能找到价格元素的选择器。

第三步：怎样运行和测试？

1. 安装

python3 -m venv .venv && source .venv/bin/activate
pip install requests beautifulsoup4 pytest

2. 测试

import sqlite3

import pytest

from watch import check, parse_price

HTML = '<html><title>t</title><h1>示例商品</h1><p class="price">¥1,299.00</p></html>'


def test_parse_price_with_currency_and_comma():
    assert parse_price(HTML, "p.price") == ("示例商品", 1299.0)


def test_missing_selector_raises():
    with pytest.raises(ValueError, match="找不到价格元素"):
        parse_price(HTML, ".nope")


def test_check_alerts_and_history(monkeypatch):
    prices = iter(["¥120", "¥99"])
    monkeypatch.setattr("watch.fetch", lambda url: f'<h1>耳机</h1><p class="price">{next(prices)}</p>')
    db = sqlite3.connect(":memory:")
    item = [{"url": "u", "selector": "p.price", "target": 100}]
    assert check(item, db, delay=0) == []
    assert check(item, db, delay=0) == [("耳机", 99.0, "u")]
    assert db.execute("select count(*) from price").fetchone()[0] == 2

python -m pytest -q

本文实际输出 3 passed。

3. 直连练习站运行

python watch.py
python watch.py --history

本文实际输出：

✓ A Light in the Attic：51.77
✓ Tipping the Velvet：53.74
2026-10-21T13:35:43  A Light in the Attic  51.77
2026-10-21T13:35:47  Tipping the Velvet  53.74

4. 改用 Web Unlocker

export BRIGHTDATA_API_KEY=你的API密钥
export BRIGHTDATA_ZONE=price_unlocker
python watch.py

5. 每天自动检查

crontab -e

0 9 * * * cd /home/me/price-watch && .venv/bin/python watch.py >> watch.log 2>&1

出问题时该怎么调？

1. 报 no such table: price

本文第一版在建表前就查询了最低价，第一次运行时报错。修复方法是在检查前先调用 init(db) 建表，现在的代码已经修好。

2. 找不到价格元素

3. 价格解析成错误数字

正则只取第一个数字。页面上同时有原价和折扣价时，把选择器写得更精确。

常见问题怎么处理？

项目资源

建议先不接 Bright Data，直接用练习站跑通 watch.py 和 --history，理解解析和存储流程。

然后再为一个你有权采集的真实页面配置选择器，确认直连被拦截时，才接入 Web Unlocker。