用 Python 给 Amazon 做“全身 CT”——可量产、可扩展的商品详情爬虫实战
一、技术选型：为什么选 Python 而不是 Java？
| 维度 | Python | Java |
|---|---|---|
| 开发效率 | 语法简洁，爬虫框架丰富（requests、scrapy、playwright） | 啰嗦，但类型安全 |
| 动态页面 | Playwright 一键拦截 JS，渲染速度秒杀 Selenium | HtmlUnit 坑多，Selenium 太重 |
| 数据科学 | Pandas + Jupyter 即时分析，无缝衔接 ML | 需要额外转数据格式 |
| 运维成本 | 脚本即服务，Serverless（Lambda/云函数）原生支持 | 需要打包 JVM，冷启动大 |
结论：“调研阶段用 Python，上线后如果 QPS 爆表再考虑 Java 重构。”
二、整体架构速览（3 分钟看懂）
┌---------------------------┐
| Amazon 商品详情页 |
└------------┬--------------┘│ 1. 随机 UA + 住宅代理池▼
┌---------------------------┐
| 解析层 (Playwright) |
| 自动等待 / 重试 / 熔断 |
└------------┬--------------┘│ 2. 字段清洗▼
┌---------------------------┐
| 存储层 (CSV/SQLite/S3) |
| 增量 / 版本控制 |
└------------┬--------------┘│ 3. 监控告警▼飞书群 + Grafana三、开发前准备（5 分钟搞定）
-
环境
Python 3.11 + VSCode + 虚拟环境 -
依赖一次性装完
bashpython -m venv venv source venv/bin/activate pip install playwright pandas tqdm loguru fake-useragent aiofiles playwright install chromium # 自动下载浏览器 -
目标字段 & CSS 选择器
字段 选择器 标题 #productTitle价格 .a-price.a-text-price.a-size-medium.apexPriceToPay .a-offscreenRating span.a-icon-alt评论数 #acrCustomerReviewText库存 #availability span图片 #landingImage的data-old-hires五点 #feature-bullets ul li span
四、MVP：120 行代码即可跑通
单文件脚本，支持异步并发 10 个 ASIN，自动重试 429，结果直接写
amazon.csv。
Python
import asyncio, csv, re, random
from pathlib import Path
from playwright.async_api import async_playwright
from loguru import logger
from fake_useragent import UserAgent
import pandas as pdCONCURRENCY = 10
RETRY = 3
TIMEOUT = 35_000
RESULT = "amazon.csv"HEADERS = ["asin","title","price","rating","review_count","availability","img_url","scrape_time"]async def scrape_one(page, asin: str) -> dict:url = f"https://www.amazon.com/dp/{asin}"logger.info("🚀 正在抓取 {}", asin)for attempt in range(1, RETRY+1):try:await page.goto(url, wait_until="domcontentloaded", timeout=TIMEOUT)await page.wait_for_selector("#productTitle", timeout=8000)breakexcept Exception as e:logger.warning("⚠️ {} 第 {} 次失败: {}", asin, attempt, e)await asyncio.sleep(random.uniform(2, 4))else:logger.error("❌ {} 超过最大重试", asin)return {"asin": asin, "title": "N/A"}# 字段提取def txt(sel): return (await page.locator(sel).first.text_content() or "").strip()title = txt("#productTitle")price = txt(".a-price.a-text-price.a-size-medium.apexPriceToPay .a-offscreen") or txt(".a-price .a-offscreen")rating = (await page.locator("span.a-icon-alt").first.get_attribute("innerHTML") or "")rating = re.search(r"([\d.]+)\sout", rating).group(1) if rating else "N/A"review_count = txt("#acrCustomerReviewText").replace(",", "").split()[0]availability = txt("#availability span")img_url = await page.locator("#landingImage").first.get_attribute("data-old-hires") or ""return {"asin": asin,"title": title,"price": price,"rating": rating,"review_count": review_count,"availability": availability,"img_url": img_url,"scrape_time": pd.Timestamp.utcnow().isoformat()}async def worker(queue: asyncio.Queue, playwright):browser = await playwright.chromium.launch(headless=True)ua = UserAgent()context = await browser.new_context(user_agent=ua.random,viewport={"width": 1280, "height": 720},locale="en-US")page = await context.new_page()while True:asin = await queue.get()if asin is None: # 结束信号queue.task_done()breakdata = await scrape_one(page, asin)await save_one(data)queue.task_done()await context.close()await browser.close()async def save_one(data):file_exists = Path(RESULT).exists()with open(RESULT, "a", newline="", encoding="utf-8") as f:writer = csv.DictWriter(f, fieldnames=HEADERS)if not file_exists:writer.writeheader()writer.writerow(data)logger.info("💾 已保存 {}", data["asin"])async def main():asins = ["B08F7N8PDP", "B08N5WRWNW", "B08N5M7S6K"] # 可换成万级列表queue = asyncio.Queue()for a in asins:await queue.put(a)async with async_playwright() as p:tasks = [asyncio.create_task(worker(queue, p)) for _ in range(CONCURRENCY)]await queue.join() # 等队列空for _ in tasks: await queue.put(None)await asyncio.gather(*tasks)logger.success(">>> 全部完成，共 {} 条", len(asins))if __name__ == "__main__":asyncio.run(main())运行效果：
2025-10-21 23:10:12 | 🚀 正在抓取 B08F7N8PDP
2025-10-21 23:10:14 | 💾 已保存 B08F7N8PDP
...
2025-10-21 23:10:25 | >>> 全部完成，共 3 条CSV 预览：
| asin | title | price | rating | review_count | availability | img_url | scrape_time |
|---|---|---|---|---|---|---|---|
| B08F7N8PDP | Apple AirPods Pro… | $199.00 | 4.6 | 184235 | In stock | https://m.media-amazon.com/images/... | 2025-10-21T23:10:14 |
五、反爬四件套，让你的爬虫“长命百岁”
Amazon 的反爬 = “动态阈值 + 行为检测 + 验证码” 三维立体防御。
下面四件套，亲测能把 429 概率降到 1% 以下：
-
住宅代理池
-
付费：BrightData、Oxylabs、IPRoyal（支持 SOCKS5）
-
自建：Tor + 轻量池（适合日采 <5k）
代码层只需在browser.new_context(proxy={"server": "http://user:pass@ip:port"})动态轮换即可。
-
-
浏览器指纹随机化
-
每次启动 playwright 随机 UA、viewport、timezone、WebGL vendor
-
禁用 WebDriver 属性：
navigator.webdriver = undefined -
屏蔽图片/CSS 加速：
page.route("**/*.{png,jpg,css}", lambda route: route.abort())
-
-
限速 + 重试
-
单 IP 每秒 ≤ 1 请求；随机 sleep 2~5 s
-
返回 429 时指数退避 1s→2s→4s→8s，最多 5 次
-
用
tenacity装饰器一键实现重试：
from tenacity import retry, wait_exponential, stop_after_attempt @retry(wait=wait_exponential(multiplier=1, min=1, max=60), stop=stop_after_attempt(5)) async def scrape_one(page, asin):... -
-
验证码熔断
-
检测到标题含 “Robot Check” 立即丢弃该代理，冷宫 30 min
-
对接 2Captcha / Ruokuai 平台自动打码（成本 ≈ $0.003/次）
-
六、把数据“喂”给业务：4 个真实场景
-
选品决策
每天 06:00 定时跑完耳机类目 5000 SKU，用 Pandas 算出“昨日降价 Top10” → 飞书群推送，运营上班即可决策。 -
动态定价
将抓取到的 FBA 价格、跟卖数丢进自研算法，自动调整 ERP 售价，保证 Buy Box 胜率 ≥ 85%。 -
库存预警
监控对手availability字段，一旦出现 “Only 2 left” 立即发邮件：可以加大广告抢流量！ -
评论情感分析
把reviewText一并收下来，用 SnowNLP / TextBlob 做情感打分，找到 1~2 星差评关键词，反向优化说明书。
七、常见坑合集（血泪史）
| 坑 | 现象 | 解决 |
|---|---|---|
| 价格字段空 | 页面是 JS 渲染，requests 抓不到 | 用 Playwright 等浏览器驱动 |
| 评分 null | 新上架无评论 | 代码里判空，默认值 “N/A” |
| 被重定向验证码 | 标题含 “Robot Check” | 立即丢弃代理，冷宫 30 min |
| 图片 URL 失效 | 带时效参数 | 及时转存到自家 OSS |
| CSV 中文乱码 | Excel 打开是 “???” | 写文件时 utf-8-sig 带 BOM |
八、10 万级 ASIN 分布式方案
单脚本玩 3 个 ASIN 没毛病，老板一句“给我把全站 30 万 SKU 每天扫一遍”怎么办？
把上面的 MVP 拆成“三件套”即可水平扩展：
-
调度层
Airflow + KubernetesPodOperator，把 30 万 ASIN 按热度拆成 4 档，分别给 4 个 DAG（小时级/日级/周级/月级）。 -
抓取节点
K8s 部署 20 个 Pod，每个 Pod 消费 Redis 队列，抓取完回写 S3（Parquet 格式）。 -
监控大盘
Prometheus 采集“成功数 / 429 数 / 平均耗时”，Grafana 一屏展示；超过阈值自动飞书 + 邮件。
九、Docker 一键部署（附 Dockerfile）
dockerfile
FROM mcr.microsoft.com/playwright/python:v1.42.0-focal
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY scraper.py .
CMD ["python", "scraper.py"]构建 & 运行：
bash
docker build -t amz-py .
docker run --rm -v $(pwd)/data:/app/data amz-py十、写在最后的“防吃牢饭”提示
Amazon 的数据受《计算机欺诈与滥用法》（CFAA）及《数字千年版权法》（DMCA）保护，请务必：
-
仅抓取“公开可见、无需登录” 的页面；
-
遵守 robots.txt（Amazon 几乎全站 allow:/，但频率需合理）；
-
数据仅限内部商业分析，不得直接转载、转售或公开 API 化；
-
生产环境先行法律评估，必要时与律师确认合规条款。