用 Python 给 Amazon 关键词搜索做“全身 CT”——可量产、可扩展的爬虫实战
一、技术选型：为什么选 Python 而不是 Java？
| 维度 | Python | Java |
|---|---|---|
| 开发效率 | 语法简洁，requests / Playwright / Scrapy 一键即用 | 啰嗦，但类型安全 |
| 动态页面 | 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 + 虚拟环境 -
依赖一次性装完
bash
python -m venv venv
source venv/bin/activate
pip install playwright pandas tqdm loguru fake-useragent aiofiles
playwright install chromium # 自动下载浏览器-
目标字段 & CSS 选择器
| 字段 | 选择器 | |---|---| | 标题 |[data-component-type="s-search-result"] h2 a span| | 价格 |.a-price .a-offscreen| | 评分 |.a-icon-alt| | 评论数 |.a-size-base| | 库存/配送 |.a-color-base| | 图片 |.s-image| | 详情页链接 |h2 a的href|
四、MVP：150 行代码即可跑通
单文件脚本，支持异步并发 10 个关键词，自动翻页、重试 429，结果直接写
keyword_search.csv。
Python
import asyncio, csv, re, random, time
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 = "keyword_search.csv"
HEADERS = ["kw","rank","title","price","rating","review_count","availability","img_url","detail_link","scrape_time"]async def scrape_one_page(page, kw: str, page_no: int) -> list:url = f"https://www.amazon.com/s?k={kw}&page={page_no}"logger.info("🚀 {} 第 {} 页", kw, page_no)for attempt in range(1, RETRY+1):try:await page.goto(url, wait_until="domcontentloaded", timeout=TIMEOUT)await page.wait_for_selector('[data-component-type="s-search-result"]', timeout=8000)breakexcept Exception as e:logger.warning("⚠️ {} 第 {} 页 第 {} 次失败: {}", kw, page_no, attempt, e)await asyncio.sleep(random.uniform(2, 4))else:logger.error("❌ {} 第 {} 页 超过最大重试", kw, page_no)return []rows = []products = await page.query_selector_all('[data-component-type="s-search-result"]')for rank, prod in enumerate(products, start=1):try:title = await prod.query_selector('h2 a span')title = await title.text_content() if title else "N/A"link = await prod.query_selector('h2 a')link = "https://amazon.com" + (await link.get_attribute('href')) if link else "N/A"price = await prod.query_selector('.a-price .a-offscreen')price = await price.text_content() if price else "N/A"rating = await prod.query_selector('.a-icon-alt')rating = await rating.text_content() if rating else "N/A"rating = re.search(r"([\d.]+)\sout", rating).group(1) if rating != "N/A" else "N/A"review = await prod.query_selector('.a-size-base')review = await review.text_content() if review else "N/A"review = review.replace(",", "").split()[0] if review != "N/A" else "N/A"avail = await prod.query_selector('.a-color-base')avail = await avail.text_content() if avail else "N/A"img = await prod.query_selector('.s-image')img = await img.get_attribute('src') if img else "N/A"rows.append({"kw": kw,"rank": (page_no-1)*48 + rank,"title": title,"price": price,"rating": rating,"review_count": review,"availability": avail,"img_url": img,"detail_link": link,"scrape_time": pd.Timestamp.utcnow().isoformat()})except Exception as e:logger.warning("解析单品失败: {}", e)continuereturn rowsasync def worker(queue: asyncio.Queue, playwright):browser = await playwright.chromium.launch(headless=True)ua = UserAgent()context = await browser.new_context(user_agent=ua.random, locale="en-US")page = await context.new_page()while True:item = await queue.get()if item is None:queue.task_done()breakkw, page_no = itemrows = await scrape_one_page(page, kw, page_no)await save_rows(rows)queue.task_done()await context.close()await browser.close()async def save_rows(rows):if not rows: returnfile_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.writerows(rows)logger.info("💾 保存 {} 条", len(rows))async def main():keywords = ["coffee maker", "office desk"] # 可换成万级列表max_page = 3 # 每关键词抓前 3 页queue = asyncio.Queue()for kw in keywords:for p in range(1, max_page+1):await queue.put((kw, p))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(">>> 全部完成，结果见 {}", RESULT)if __name__ == "__main__":asyncio.run(main())运行效果：
2025-10-22 08:12:10 | 🚀 coffee maker 第 1 页
2025-10-22 08:12:13 | 💾 保存 48 条
...
2025-10-22 08:13:25 | >>> 全部完成，结果见 keyword_search.csvCSV 预览：
| kw | rank | title | price | rating | review_count | availability | img_url | detail_link | scrape_time |
|---|---|---|---|---|---|---|---|---|---|
| coffee maker | 1 | Keurig K-Classic Coffee Maker | $89.99 | 4.6 | 84235 | In stock | https://m.media-amazon.com/images/... | https://amazon.com/dp/... | 2025-10-22T08:12:13 |
五、反爬四件套，让你的爬虫“长命百岁”
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(page, kw, page_no):... -
-
验证码熔断
-
检测到标题含 “Robot Check” 立即丢弃该代理，冷宫 30 min
-
对接 2Captcha / Ruokuai 平台自动打码（成本 ≈ $0.003/次）
-
六、把数据“喂”给业务：4 个真实场景
-
选品决策
每天 06:00 定时跑完 1000 个关键词，用 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 万级关键词分布式方案
单脚本玩 2 个关键词没毛病，老板一句“给我把全站 30 万关键词每天扫一遍”怎么办？
把上面的 MVP 拆成“三件套”即可水平扩展：
-
调度层
Airflow + KubernetesPodOperator，把 30 万关键词按热度拆成 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-keyword-py .
docker run --rm -v $(pwd)/data:/app/data amz-keyword-py十、写在最后的“防吃牢饭”提示
Amazon 的数据受《计算机欺诈与滥用法》（CFAA）及《数字千年版权法》（DMCA）保护，请务必：
-
仅抓取“公开可见、无需登录” 的页面；
-
遵守 robots.txt（Amazon 几乎全站 allow:/，但频率需合理）；
-
数据仅限内部商业分析，不得直接转载、转售或公开 API 化；
-
生产环境先行法律评估，必要时与律师确认合规条款。