实操教程2026年9月29日阅读约 9 分钟

抓取亚马逊价格:用 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 描述一下,再决定要不要多花钱。

还有后续问题?

去 Discord 问。答案还能帮到下一个读到这个帖子的人。

加入 Discorddiscord.gg/proxypanda
$5 起步去 Discord 问