华为小米荣耀集体调价，我用 Python 写了套可长期运行的手机比价监控
早上刷到一条消息，华为、小米、荣耀多款手机最高涨了 1000 块，机构说这股压力要拖到明年。看完我关心的不是涨多少，而是通告里的涨幅有没有落到电商页面的实时标价上。官方调价、渠道补贴、以旧换新、限时券叠在一起，购物车里的成交价常常和新闻对不上。要验证，只能把几个平台的标价按天抓下来自己比。我花了一下午把这套监控写成能长期跑的脚本。下面把工程上踩过的点和对应的写法拆开讲，重点不在能跑，在于能稳定跑一个月不掉链子。 先定义数据契约很多人先写请求再想存什么，抓回来一堆字段再筛。我反过来，先用 dataclass 把一条价格记录的形状定死，后续落库、写文件、画图都顺着这个结构，不用来回改。 ``` from dataclasses import dataclass from datetime import datetime from zoneinfo import ZoneInfo CST = ZoneInfo("Asia/Shanghai") @dataclass class PhonePrice: model: str platform: str sku: str price: float currency: str captured_at: datetime # 带时区的感知时间 ``` currency 和 sku 是我后来补上的。不同平台价格单位不统一，加上 sku 才能精确定位到具体商品，而不是只靠型号名去猜。captured_at 用 zoneinfo 带时区，机器时区一变数据就乱，这个坑我踩过。价格走独立接口，不在 HTML 里电商页面是前后端分离，标价由浏览器另发请求从价接口取回。京东的价接口大概在 p.3.cn，拼上 sku id 返回 JSON。直接打接口比解析整页快得多，也绕开了页面里大量无关 DOM。 ``` import requests def fetch_raw(sku: str, session: requests.Session) -> dict: url = f"https://p.3.cn/prices/mgets?skuIds=J_{sku}" resp = session.get(url, timeout=(3, 8), headers={ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://item.jd.com/", }) resp.raise_for_status() return resp.json()[0] ``` timeout 我写成元组 (connect, read)，把建连和读取分开控。空 UA 有的接口直接甩空响应，加 Referer 更接近真实浏览器轨迹。raise_for_status 让 4xx、5xx 立刻抛错，不会静默拿到半截数据继续往下走。单 IP 撑不住，代理层要独立可替换这才是真正的工程难点。本机一个 IP 高频敲同一接口，当天下午基本 403，严重的返回空数据还不报错。平台风控比前年紧，单 IP 硬刚当天出局。我把代理抽成一个独立组件，不让抓取逻辑依赖具体供应商。长期用的是亿牛云代理，企业级动态 IP，调用它的 API 拿可用 IP，按量计费，IP 池够大，掉线换一个就行。对接时把代理抽象成可迭代的源，后面换供应商只改这一处。 ``` from typing import Iterator class YiniuProxySource: """亿牛云动态代理源，按需拉取并在本地缓存一批 IP。""" def __init__(self, api: str, pool_size: int = 5) -> None: self._api = api self._pool: list[str] = [] self._pool_size = pool_size def _refill(self) -> None: resp = requests.get(self._api, timeout=5).json() # 亿牛云返回 ip 与 port 字段，按实际响应适配 self._pool = [f"{r['ip']}:{r['port']}" for r in resp["data"]] def __iter__(self) -> Iterator[str]: return self def __next__(self) -> str: if not self._pool: self._refill() return f"http://{self._pool.pop(0)}" ``` 本地攒一个小池，用完再拉，避免每条请求都打一次供应商 API。这样代理获取本身不会成为瓶颈，也不会因为频繁调用把配额耗光。网络会抖，重试和日志要分两层重试我放在两个层面。连接层用 urllib3 的 Retry 挂到 Session 上，处理瞬时抖动；业务层用 tenacity 处理代理失效、JSON 解析失败这类需要换代理的重试。 ``` import logging from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from requests.exceptions import ProxyError, Timeout, HTTPError logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s" ) logger = logging.getLogger("price_crawler") def build_session(proxies: dict) -> requests.Session: s = requests.Session() adapter = requests.adapters.HTTPAdapter( max_retries=requests.packages.urllib3.util.retry.Retry( total=2, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503] ) ) s.mount("https://", adapter) s.proxies.update(proxies) return s @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1), retry=retry_if_exception_type((ProxyError, Timeout, HTTPError)), reraise=True, ) def fetch_price(sku: str, proxies: dict) -> float: session = build_session(proxies) raw = fetch_raw(sku, session) price = float(raw["p"]) logger.info("sku=%s price=%.2f proxy=%s", sku, price, proxies["http"]) return price ``` 指数退避比固定间隔稳。429 是限流信号，我把它塞进 status_forcelist，让连接层先退避再试。业务层只重试代理和网络类异常，价格解析出的业务错误不该被重试掩盖，该报错就报错。拿到价格先校验，再落库抓回来的数不能直接信。价接口偶尔返回 0、返回字符串占位、返回和上次差出几倍的数。落库前做一道校验，脏数据挡在外面。 ``` def validate(price: float, last: float | None) -> bool: if price 100000: return False if last is not None and abs(price - last) / last > 0.5: return False # 单日波动超五成，大概率是接口异常 return True ``` 落库我换成 SQLite，靠 (sku, captured_at) 做幂等，重跑不会插重复行。CSV 适合一次性分析，长期监控还是库省心。 ``` import sqlite3 from contextlib import closing def save(conn: sqlite3.Connection, row: PhonePrice) -> None: with closing(conn): conn.execute( "INSERT OR IGNORE INTO prices(" "model,platform,sku,price,currency,captured_at) " "VALUES(?,?,?,?,?,?)", (row.model, row.platform, row.sku, row.price, row.currency, row.captured_at.astimezone(CST).isoformat()), ) conn.commit() ``` 控制频率，别把对面打挂多 sku 循环时我在每次请求间加随机间隔，模拟人手节奏。想快可以用 ThreadPoolExecutor 开几个 worker，但并发数要压住，否则代理 IP 被你自己的请求耗光，风控也来得更快。 ``` import random, time def run_once(conn: sqlite3.Connection, source: YiniuProxySource) -> None: last: dict[str, float] = {} for model, platform, sku in WATCH_LIST: try: price = fetch_price(sku, {"http": next(source), "https": next(source)}) except Exception as exc: logger.error("跳过 sku=%s 原因=%s", sku, exc) continue if not validate(price, last.get(sku)): logger.warning("sku=%s 校验未过 price=%.2f", sku, price) continue save(conn, PhonePrice(model, platform, sku, price, "CNY", datetime.now(CST))) last[sku] = price time.sleep(random.uniform(1.5, 3.5)) ``` 数据到手能看什么SQLite 攒够一周，按 sku 分组拉出来画折线，谁在哪一刻动的价清清楚楚。新闻说最高涨 1000，你能在图上区分是全线普调还是只动了顶配。有时候通告喊涨，券后价反而更低，这种反差只有自己抓了才知道。定时跑我交给系统计划任务，每天早九晚八各扫一次。比人肉盯着稳，人会有忘的时候，脚本不会。整套下来，能跑通半天就能写完，能稳定跑一个月才是硬功夫。爬虫逻辑不值钱，代理质量和重试、校验、限频这些工程细节才决定你拿到的是干净数据还是一堆 403。这也是我长期用亿牛云的原因，IP 池和稳定性过关，剩下的都是自己的事。只想看个大概，手动查两次也行。要做长期价格档案，就让脚本替你守着。
