华为小米荣耀集体调价，我用 Python 写了套可长期运行的手机比价监控

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  # 带时区的感知时间

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]

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)}"

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

def validate(price: float, last: float | None) -> bool:
 if price <= 0 or price > 100000:
 return False
 if last is not None and abs(price - last) / last > 0.5:
 return False # 单日波动超五成，大概率是接口异常
 return True

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()

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))