亚马逊 ASIN 的价格监控对于电子商务卖家、交易搜寻者和市场情报团队至关重要。直接抓取亚马逊产品页面会快速触发机器人检测，并且需要持续维护。 Scavio API 提供了一个稳定的产品端点，可以返回任何 ASIN 的当前价格、评级、评论计数和可用性。本教程构建一个轮询脚本，用于检查 ASIN 列表、将价格与存储的基准进行比较，并在价格低于配置的阈值时打印警报。
前置条件
- Python 3.8 或更高版本
- 请求已安装库
- Scavio API 密钥
- 要监控的亚马逊 ASIN 列表
操作指南
步骤 1: 定义您的 ASIN 监视列表
创建将 ASIN 映射到目标价格阈值的字典。当实时价格跌至或低于阈值时，脚本将发出警报。
Python
WATCHLIST = {
"B09G9FPHY6": {"name": "Echo Dot 5th Gen", "threshold": 35.0},
"B07FZ8S74R": {"name": "Fire TV Stick 4K", "threshold": 25.0},
}步骤 2: 获取 ASIN 的当前价格
使用平台 amazon 和 ASIN 作为查询，POST 到 Scavio 端点。响应包括产品对象下的价格字段。
Python
def get_amazon_price(asin: str) -> dict:
response = requests.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": API_KEY},
json={"platform": "amazon", "query": asin, "marketplace": "US"}
)
response.raise_for_status()
return response.json()步骤 3: 将价格与阈值进行比较
从响应中提取当前价格并将其与存储的阈值进行比较。优雅地处理缺失或无价格。
Python
def check_price(asin: str, meta: dict, data: dict) -> None:
product = data.get("product", {})
price_str = product.get("price", "")
if not price_str:
return
price = float(price_str.replace("$", "").replace(",", ""))
if price <= meta["threshold"]:
print(f"ALERT: {meta['name']} is ${price} (threshold ${meta['threshold']})")步骤 4: 运行监控循环
迭代所有 ASIN、获取价格并检查阈值。在民意调查之间休息以保持在速率限制之内。
Python
import time
while True:
for asin, meta in WATCHLIST.items():
data = get_amazon_price(asin)
check_price(asin, meta, data)
print("Cycle complete. Sleeping 3600s...")
time.sleep(3600)Python 示例
Python
import os
import time
import requests
API_KEY = os.environ.get("SCAVIO_API_KEY", "your_scavio_api_key")
ENDPOINT = "https://api.scavio.dev/api/v1/search"
WATCHLIST = {
"B09G9FPHY6": {"name": "Echo Dot 5th Gen", "threshold": 35.0},
"B07FZ8S74R": {"name": "Fire TV Stick 4K", "threshold": 25.0},
}
def get_price(asin: str) -> float | None:
r = requests.post(
ENDPOINT,
headers={"x-api-key": API_KEY},
json={"platform": "amazon", "query": asin, "marketplace": "US"}
)
r.raise_for_status()
price_str = r.json().get("product", {}).get("price", "")
if price_str:
return float(price_str.replace("$", "").replace(",", ""))
return None
def monitor():
for asin, meta in WATCHLIST.items():
price = get_price(asin)
if price and price <= meta["threshold"]:
print(f"[ALERT] {meta['name']}: ${price}")
else:
print(f"[OK] {meta['name']}: ${price}")
if __name__ == "__main__":
while True:
monitor()
time.sleep(3600)JavaScript 示例
JavaScript
const API_KEY = process.env.SCAVIO_API_KEY || "your_scavio_api_key";
const ENDPOINT = "https://api.scavio.dev/api/v1/search";
const WATCHLIST = {
"B09G9FPHY6": { name: "Echo Dot 5th Gen", threshold: 35.0 },
"B07FZ8S74R": { name: "Fire TV Stick 4K", threshold: 25.0 },
};
async function getPrice(asin) {
const res = await fetch(ENDPOINT, {
method: "POST",
headers: { "x-api-key": API_KEY, "Content-Type": "application/json" },
body: JSON.stringify({ platform: "amazon", query: asin, marketplace: "US" })
});
const data = await res.json();
const priceStr = data?.product?.price || "";
return priceStr ? parseFloat(priceStr.replace(/[$,]/g, "")) : null;
}
async function monitor() {
for (const [asin, meta] of Object.entries(WATCHLIST)) {
const price = await getPrice(asin);
if (price !== null && price <= meta.threshold) {
console.log(`[ALERT] ${meta.name}: $${price}`);
} else {
console.log(`[OK] ${meta.name}: $${price}`);
}
}
}
monitor().catch(console.error);预期输出
JSON
{
"product": {
"asin": "B09G9FPHY6",
"title": "Echo Dot (5th Gen)",
"price": "$29.99",
"original_price": "$49.99",
"rating": "4.7",
"reviews_count": 284521,
"availability": "In Stock",
"marketplace": "US"
}
}