抓取亚马逊价格:用 Python 做价格监控
用 Python、requests 和 BeautifulSoup 抓取亚马逊价格:按 ASIN 解析标题和价格,识别机器人验证页,控制节奏,结果存进 CSV,适合跨境电商选品与竞品监控。
抓取亚马逊价格的做法是:按 ASIN 逐个请求公开的商品页,保持缓慢而稳定的节奏;用 BeautifulSoup 配合几组备用选择器解析标题和价格;一旦出现机器人验证页就立刻停下;每条结果追加写入 CSV。使用该站点所在国家的静态 IP,才能保证今天和昨天的数字可以放在一起比。
做跨境电商的卖家,常常要盯着美国站、德国站、日本站上竞品的价格变化,或者在选品阶段记录一批商品的价格走势。这篇教程用大约一百行 Python 搭一个这样的亚马逊价格监控脚本,内容包括:哪些数据可以合理地采集、先用哪条代理线路、解析器怎么写、节奏怎么控制、怎么区分机器人验证页和商品页,以及这项任务大概要用多少流量。
抓取亚马逊价格可以吗?
写代码之前先把这件事说清楚。亚马逊的使用条件(Conditions of Use)不允许在网站上使用机器人、数据挖掘或类似的数据提取工具。价格监控脚本就属于这类工具,即使它读取的每个页面都是公开的。下面介绍的是一种礼貌地采集公开价格的方法,但这并不意味着亚马逊同意了这种做法;这类条款对你是否有约束力,取决于你所在的地区和你怎么使用这些数据。一般情况见使用代理合法吗?;如果用于商业用途,请咨询专业人士。
如果决定要做,请守住这几条:
- 只抓公开商品页。 不登录、不用账号、不碰购物车,不碰任何需要登录才能看到的内容。绝不要在登录自己亚马逊账号(尤其是卖家账号)的状态下抓取。
- 低频率。 每天一两次、几百个商品,是价格监控;每小时几千个页面,就是在给别人的网站添负担。
- 阅读该站点的 robots.txt,避开其中禁止访问的路径。
- 先考虑官方渠道。 如果你是卖家或联盟推广者,亚马逊提供可以直接返回价格数据的 API,不需要做这些。
监控任何访客都能看到的价格,在我们这边是允许的,价格监控使用场景有详细说明。为了看到会员价而批量注册账号则不允许。
亚马逊价格监控适合哪条代理线路?
亚马逊显示的价格、配送选项,有时连展示的报价,都取决于两件事:你访问的是哪个站点域名,以及它根据你的 IP 推测出的配送地址。监控脚本每次都应该从同一个国家发出请求,否则昨天的价格和今天的价格就不是同一种测量。
所以静态线路更合适:
- 先从数据中心代理开始。 下单时可以选择国家、地区和城市,而且按 IP 算是最便宜的一种。先拿少量 ASIN 测几天。
- 如果放慢节奏后,数据中心 IP 仍然频繁遇到机器人验证,换成 ISP 代理。 ISP IP 登记在面向家庭用户的宽带运营商名下,对服务器流量有戒心的网站,往往更愿意把它当成普通在家购物的用户。位置同样可以自选。两者的比较见 ISP 还是数据中心代理?
- 住宅代理不太适合这个任务。 它在家庭网络之间轮换,而且不能选择国家、地区或城市,同一个商品的两次检查可能来自两个不同的国家。
静态 IP 可选的地点见地点页面。怎么针对自己的目标网站做决定,住宅还是数据中心?里有一个五分钟就能做完的测试。
用 Python 写亚马逊价格抓取脚本
先安装两个库:
pip install requests beautifulsoup4
然后把下面的代码保存为 amazon_prices.py。把你的 ASIN 填进 ASINS,把 MARKETPLACE 设成对应站点的域名,再从服务页面填入代理信息。
import csv
import os
import random
import re
import time
from datetime import datetime, timezone
import requests
from bs4 import BeautifulSoup
PROXY = "http://USERNAME:PASSWORD@HOST:PORT"
PROXIES = {"http": PROXY, "https": PROXY}
MARKETPLACE = "MARKETPLACE_DOMAIN"
ASINS = ["ASIN_1", "ASIN_2", "ASIN_3"]
OUTPUT = "prices.csv"
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
}
TITLE_SELECTORS = ["#productTitle", "#title"]
PRICE_SELECTORS = [
"#corePriceDisplay_desktop_feature_div .a-price .a-offscreen",
"#corePrice_feature_div .a-price .a-offscreen",
"#apex_desktop .a-price .a-offscreen",
"#price_inside_buybox",
"#priceblock_ourprice",
"#priceblock_dealprice",
]
ROBOT_MARKERS = ("validatecaptcha", "type the characters you see", "robot check")
class Blocked(Exception):
pass
def first_text(soup, selectors):
for selector in selectors:
node = soup.select_one(selector)
if node and node.get_text(strip=True):
return node.get_text(strip=True)
return None
def to_number(text):
digits = re.sub(r"[^\d.,]", "", text or "")
if not digits:
return None
last = max(digits.rfind("."), digits.rfind(","))
if last != -1 and len(digits) - last - 1 == 2:
whole, fraction = digits[:last], digits[last + 1:]
else:
whole, fraction = digits, "0"
return float(re.sub(r"[.,]", "", whole) + "." + fraction)
def parse_product(html):
if any(marker in html.lower() for marker in ROBOT_MARKERS):
raise Blocked("robot check page")
soup = BeautifulSoup(html, "html.parser")
price_text = first_text(soup, PRICE_SELECTORS)
return {
"title": first_text(soup, TITLE_SELECTORS),
"price_text": price_text,
"price": to_number(price_text),
}
def fetch(asin):
url = f"https://{MARKETPLACE}/dp/{asin}"
response = requests.get(url, headers=HEADERS, proxies=PROXIES, timeout=(5, 30))
if response.status_code in (429, 503):
raise Blocked(f"HTTP {response.status_code}")
if response.status_code == 404:
return None
response.raise_for_status()
item = parse_product(response.text)
if item["title"] is None:
with open(f"unparsed-{asin}.html", "w", encoding="utf-8") as f:
f.write(response.text)
return item
def main():
new_file = not os.path.exists(OUTPUT)
with open(OUTPUT, "a", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
if new_file:
writer.writerow(["checked_at", "marketplace", "asin", "title", "price_text", "price"])
for asin in ASINS:
try:
item = fetch(asin)
except Blocked as reason:
print(f"Stopping at {asin}: {reason}")
break
except requests.RequestException as error:
print(f"{asin}: {error}")
item = None
if item:
writer.writerow([
datetime.now(timezone.utc).isoformat(timespec="seconds"),
MARKETPLACE, asin, item["title"], item["price_text"], item["price"],
])
print(asin, item["price_text"], item["title"])
time.sleep(random.uniform(8, 15))
if __name__ == "__main__":
main()
每一部分做什么
- 网址由站点域名和 ASIN 拼成。ASIN 是每个亚马逊商品链接里
/dp/后面那串十位的商品编码,/dp/ASIN是商品页最短、最稳定的写法。 PRICE_SELECTORS是一个按顺序尝试的列表。 这些年亚马逊把主价格在好几个区块之间挪过,不同类型的商品至今也用着不同的区块。第一个取到文字的选择器生效。列表特意没有用单独的.a-price兜底,因为它还会匹配“买了该商品的顾客还买了”轮播里的价格,结果会记成别的商品。to_number同时保留原始文字和解析后的数字。 各站点的价格写法不同:有的写成1,299.00,有的写成1.299,00,日本站则是不带小数的整数。函数只在最后一个分隔符后面正好有两位数字时,才把它当作小数点。原始的price_text也一并写进 CSV,随时可以核对解析结果。- 标题为空,说明拿到的不是预期的页面。 这时 HTML 会保存到
unparsed-ASIN.html,方便你查看返回了什么。页面改版往往就是这样被发现的,而不是等攒了一周的空行才察觉。 - 价格为空不算错误。 缺货商品、只有其他卖家报价的商品,常常没有主价格。这一行照样写入,价格留空,这本身也是有用的记录。
选择器一定会变
亚马逊的页面结构经常变,而且不会提前通知,这个列表下个月是否还管用,谁也不能保证。当结果开始出现空行时,在浏览器里打开一个商品页,右键点价格,选择“检查”,找离它最近、带稳定 id 的元素,把它的选择器加到 PRICE_SELECTORS 的最前面。
改完之后,先用保存下来的文件测试,不花任何流量:
from amazon_prices import parse_product
with open("unparsed-ASIN_1.html", encoding="utf-8") as f:
print(parse_product(f.read()))
请求头和节奏
- User-Agent: 从你自己的浏览器里复制。默认的
python-requests标识,第一个请求就告诉对方这是脚本。 - Accept-Language: 与站点匹配。美国站用
en-US,德国站用de-DE,日本站用ja-JP,以此类推。 - 节奏: 脚本在两个商品之间等待 8 到 15 秒,并加入随机性,避免节奏过于机械。不要在同一个 IP 上并行运行多份脚本;需要更高吞吐量时,多加几个 IP,每个 IP 分一部分 ASIN。
- 频率: 大多数价格不会每分钟都变。用 cron 或 Windows 任务计划程序每天跑一两次,对多数监控需求已经足够。
出现机器人验证,就停下
亚马逊怀疑是自动程序时,会返回一个要求你输入图片中字符的页面,有时状态码是 200,有时是 503。parse_product 会查找这种页面的特征文字,循环在遇到第一个时就停止,不再继续。
停下来正是关键。机器人验证说明网站已经怀疑这个 IP 或这批流量,之后的每一个请求都会加深怀疑。靠重试硬闯,会把短暂的冷却期变成长期封锁,在按 GB 计费的线路上还要为每个验证页付流量费。等几个小时,放慢节奏,再试。如果节奏已经很温和,验证页仍然不断出现,就该从数据中心换到 ISP 了。其他常见原因见爬虫为什么突然开始被封了,暂时性错误的退避方法见处理封锁和重试,别白白烧掉代理流量。
亚马逊价格监控要用多少流量?
即使不下载图片(这个脚本从不下载图片),亚马逊商品页的 HTML 也很重。规划之前先实测:按 GB 计费的代理怎么算钱里有一个脚本,可以经代理称量一批样本页面的大小。
举一个用整数凑出来的例子:如果一个商品页在线路上是 150 KB,你每天两次监控 200 个 ASIN、持续 30 天,一共是 12,000 个页面,约 1.8 GB。数据中心和 ISP IP 按 IP 租用一段期限,价格页面上列出了每种 IP 套餐包含的流量,请确认这个任务放得下。
读取 CSV
每次运行都会追加新行,prices.csv 就成了一份价格历史,用任何表格软件都能打开。只打印和上一次相比价格有变化的商品:
import csv
from collections import defaultdict
history = defaultdict(list)
with open("prices.csv", newline="", encoding="utf-8") as f:
for row in csv.DictReader(f):
if row["price"]:
history[row["asin"]].append(float(row["price"]))
for asin, prices in history.items():
if len(prices) > 1 and prices[-1] != prices[-2]:
print(asin, prices[-2], "->", prices[-1])
快速问答
为什么脚本看到的价格和我浏览器里的不一样? 你的浏览器带着你的配送地址、登录状态和 Cookie。脚本看到的是代理所在国家、未登录访客看到的价格。用不登录的无痕窗口对比一下。
不用代理能抓亚马逊价格吗? 偶尔抓几个商品可以。用家里或服务器的地址反复检查,会更早遇到机器人验证,而且你永远只能看到自己所在国家的视图。
为什么有些 ASIN 的价格是空的? 商品可能缺货、只有其他卖家的报价,或者用了你的选择器没覆盖到的页面布局。用浏览器打开看看是哪一种。
可以每几分钟跑一次吗? 可以,这也是让 IP 被标记的好办法。价格变得慢,就按它变化的节奏去查。
从小规模开始
在对应站点的国家租一个数据中心 IP,就足够对几十个商品跑一周,看看返回的页面是否干净。各期限的价格见价格页面;如果目标网站给你出了难题,先到 Discord 描述一下,再决定要不要多花钱。