摘要
本文面向两类读者：准备把亚马逊数据接入自有系统的技术负责人，以及参与供应商选型与合同条款审核的采购与法务角色。市面上的教程停在「请求返回 200、字段打印出来」那一刻，但生产环境要在请求之外补八个工程环节，否则第三到第五周就会出现不报错的失效。
本文给出一套可运行的 Python 结构与上线前自查清单，并说明供应商与采购方之间的权责边界。读完你应能判断：一份数据采集方案在字段契约、失败计费、速率额度三处是否写清了关键条款，以及自己团队还要在落地与观测上投入多少工程量。需要采集亚马逊数据的接口服务，可看pangolinfo亚马逊数据采集 API的方案说明。
一、为什么 demo 活不过第三周
把教程脚本放进定时任务，前几天报表通常好看。问题集中在第三到第五周浮出水面，且都不是报错形式。这类失效的代价不在修复本身，而在返工：笔记本上的脚本错了，改一行再跑一次是三十秒；库里混了三周的重复行，代价是追回所有下游报表，且你未必知道哪些报表用过这批数据。
| 现象 | 账面表现 | 缺口层 |
|---|---|---|
| 页面改版，选择器失效 | 请求返 200，字段为 None | 缺字段非空校验 |
| 搜索翻页到上限 | 数据增长变慢 | 缺单次任务产出估算 |
| 失败后重试成功 | 成功率维持 99% | 缺失败分类 |
| CSV 持续追加 | 行数增长，业务侧投诉重复 | 缺主键与幂等重跑 |
四类共同点是监控不会叫。状态码、异常数、退出码都正常，只有报表数字悄悄偏离。指望报警发现它们，等于让监控去猜业务语义。结构上省下的几个小时，会在数据事故里连本带利还回去。
二、五条 Python 路线，各自把多少工作留给你
选型不是挑 API，是挑自己还要写多少代码。同一份任务，五条路线留给团队的工程量差别在十倍以上。
| 路线 | 你要自己实现的部分 | 卡在哪里 |
|---|---|---|
| 官方 SP-API | LWA 授权刷新、角色切换、配额排队、字段拼装 | 授权边界：只能取自己的数据 |
| requests + BeautifulSoup | 反爬对抗、IP 轮换、渲染、解析、重试、调度 | 亚马逊改版一次就要改一次选择器 |
| Scrapy + 代理池 | 规则维护、指纹伪装、分布式去重、降级 | 并发有了，数据语义仍缺 |
| 通用抓取 / SERP API | 字段映射、亚马逊特有语义 | 拿到通用网页结果，不是电商对象 |
| 专用亚马逊数据 API | 契约定义、质量门、去重、落地与观测 | 上游收走解析与反爬，其余仍在你侧 |
最后一行要分清：换供应商省掉的是「怎么把字节变成字段」，省不掉「哪些字段必须非空」「同一商品今天和昨天怎么对齐」「这批数据值多少钱」。这三件事的定义权在业务侧，任何 API 都替不了。自建采集与直接调用 API 在工程量上的差异，可参考自建采集与调用 API 的口径对比。
三、请求层：超时、凭据、重试收进一个类
第一段代码先收拢三件易散落的事：超时来源、密钥位置、重试责任。散落的结果是某夜任务用了默认超时，挂住整个调度。
# pip install httpx
import os, time, random
import httpx
BASE_URL = os.environ["AMZ_API_BASE"] # 换成供应商域名
API_KEY = os.environ["AMZ_API_KEY"] # 密钥只从环境变量读
ENDPOINTS = {
# 端点集中维护，改版只改这一处
"product": "/v1/amazon/product",
"search": "/v1/amazon/search",
"review": "/v1/amazon/reviews",
}
TIMEOUT = httpx.Timeout(connect=5.0, read=30.0, write=5.0, pool=10.0)
class AmazonDataClient:
def __init__(self, base_url: str = BASE_URL, key: str = API_KEY) -> None:
self._http = httpx.Client(
base_url=base_url,
timeout=TIMEOUT,
headers={
"Authorization": f"Bearer {key}",
"User-Agent": "amz-pipeline/1.0 (+ops@example.com)",
},
)
def get_json(self, endpoint: str, params: dict, attempts: int = 3) -> dict:
path = ENDPOINTS[endpoint]
last = None
for attempt in range(1, attempts + 1):
try:
resp = self._http.get(path, params=params)
except httpx.TransportError as exc: # 连接层问题，可重试
last = exc
else:
if resp.status_code < 400:
return resp.json()
if resp.status_code not in (429, 500, 502, 503, 504):
raise ValueError(f"{endpoint} rejected: {resp.status_code} {resp.text[:200]}")
last = ValueError(f"HTTP {resp.status_code}")
time.sleep(min(2 ** attempt, 30) + random.uniform(0, 1)) # 退避 + 抖动
raise RuntimeError(f"{endpoint} failed after {attempts} attempts") from last
超时拆成连接、读取、写入、连接池四段，读取给 30 秒是渲染型端点要等异步加载。退避之后加抖动，避免几十个并发任务同一秒集体重试。User-Agent 里留联系邮箱，供应商封禁或限流时能找到人。
四、字段契约与 P0 字段
拿到字典后不直接入库。中间放一层模型：把供应商字段名映射成内部名，版本变化时让旧数据自动失效。
from dataclasses import dataclass
from datetime import date
from typing import Optional
CONTRACT_VERSION = "2026-09-01" # 字段增减必须改这里
@dataclass(frozen=True)
class ProductRow:
contract_version: str
asin: str
marketplace: str
captured_at: date
title: Optional[str]
price: Optional[float]
currency: Optional[str]
rating: Optional[float]
review_count: Optional[int]
is_sponsored_slot: bool
P0_FIELDS = ("title", "price", "currency") # 缺任何一个，这条记录不可用
P0_FIELDS 这行是整套结构的收益来源：它把「数据好不好」从主观判断变成可写进 CI 的断言。contract_version 进入主键后，升级字段定义会触发历史数据重跑，不必担心新旧口径混在同一张表。字段该取哪些、怎么验，按字段而不是按宣传页去比较供应商那篇给了一份逐字段对照方法。
五、质量门：覆盖率与填充率分开算
字段存在不等于字段有值。覆盖率衡量结构里有没有路径，填充率衡量实际交付取到没取到。两者差距靠肉眼难发现，因为它混在正常数据里。
def quality_gate(rows: list[ProductRow], min_fill: float = 0.95) -> dict:
n = len(rows) or 1
report = {
}
for field in P0_FIELDS:
filled = sum(1 for r in rows if getattr(r, field) is not None)
report[field] = round(filled / n, 4)
report["usable"] = all(report[f] >= min_fill for f in P0_FIELDS)
return report
# 用法：gate = quality_gate(rows)
# 返回 {'title': 1.0, 'price': 0.61, 'currency': 1.0, 'usable': False}
# 看到 price 掉到 0.61 就该停下手上的活去查，而不是继续往库里写
这段检查跑在入库之前，不是跑在报表里。顺序反过来，脏数据已进库，清理成本高几倍。供应商计费系统把返 200 记作成功，业务侧把字段为空记作缺失，两套账对不上，唯一的办法是在自己管道里给出第三种定义。这个定义的完整形态，以及为什么它决定「那些之后甩不掉的治理工作」，在《数据管道里剩下的自建部分》那篇按层拆过。
六、翻页与跨页去重
先定主键：(asin, marketplace, captured_at, contract_version)。这个元组唯一决定一条记录，重跑同一天只覆盖同一行，不追加。定键之前讨论翻页是在浪费时间。
翻页有两个现实约束。深度限制：单一搜索词通常翻不到二十页之后，要更深只能按类目、价格区间或品牌拆词，把任务切成若干并行子任务。跨页重复：同一件商品在相邻两页重复出现是常态，尤其广告位外的自然结果。
去重率本身是有用信号。某天从 3% 涨到 40%，多半不是亚马逊改了排序，而是去重键写错或分页参数失效。records_per_call 是后面算钱的直接输入，把每次任务的实际重复比例记下来。
七、失败四类处置
except Exception: retry 是最贵的一行。它把参数错误、鉴权失败、配额耗尽、网络抖动混成一件事，结果是参数错误也重试三次，然后带着 429 让整批任务排队。
| 类别 | 例子 | 处置 |
|---|---|---|
| 瞬时故障 | 连接超时、502/503/504 | 退避重试，计入尝试次数 |
| 限流 | 429 带 Retry-After | 按头里秒数等待，不加码 |
| 契约错误 | 400、缺必填参数、密钥失效 | 直接失败告警，不重试 |
| 内容缺失 | 200 但 P0 字段为空 | 入库前拦截，进死信队列 |
第三类出现就该停整批任务，重试只耗额度。第四类靠质量门拦住，不能指望状态码。attempts 计数必须写进日志——失败的那次在多数供应商那里也计费，直接出现在月底账单上。
八、并发用信号量，而非 random.sleep
time.sleep(random.uniform(1, 3)) 把任务变慢，不把速率控制住。随机延时下实际 QPS 随并发数波动，撞上限流只是时间问题。
import asyncio, httpx
async def fetch_all(client: httpx.AsyncClient, jobs: list[dict], concurrency: int = 8):
sem = asyncio.Semaphore(concurrency)
results = []
async def one(job):
async with sem:
for attempt in range(1, 4):
try:
r = await client.get(ENDPOINTS[job["endpoint"]], params=job["params"])
except httpx.TransportError:
await asyncio.sleep(2 ** attempt)
continue
if r.status_code == 429:
wait = int(r.headers.get("Retry-After", 5))
await asyncio.sleep(wait)
continue
if r.status_code < 400:
results.append((job, r.json()))
return
if r.status_code not in (500, 502, 503, 504):
raise RuntimeError(f"job {job} rejected: {r.status_code}")
raise RuntimeError(f"job {job} exhausted retries")
await asyncio.gather(*[one(j) for j in jobs])
return results
并发数从 4 起步，观察 429 比例与端到端时延两条曲线再上调。异步只是把等待时间做别的事，不创造容量；上游给每秒 10 次额度，八个并发和十六个并发天花板一样。
九、快照表落地与成本埋点
存当前价格那一刻也丢掉了最有价值的信息：价格何时变的。每次运行当快照追加，主键带 captured_at，历史自然留表。
# pip install duckdb
import duckdb
con = duckdb.connect("amazon.duckdb")
con.execute("""
CREATE TABLE IF NOT EXISTS product_snapshot (
asin VARCHAR, marketplace VARCHAR, captured_at DATE,
contract_version VARCHAR, title VARCHAR, price DOUBLE,
currency VARCHAR, rating DOUBLE, review_count INTEGER,
is_sponsored_slot BOOLEAN,
PRIMARY KEY (asin, marketplace, captured_at, contract_version)
)
""")
con.executemany(
"INSERT OR REPLACE INTO product_snapshot VALUES (?,?,?,?,?,?,?,?,?,?)",
[(r.asin, r.marketplace, r.captured_at, r.contract_version, r.title,
r.price, r.currency, r.rating, r.review_count, r.is_sponsored_slot) for r in rows],
)
INSERT OR REPLACE 四个字符解决「任务失败是否污染数据」：同一天重跑覆盖同一批主键，不留半旧半新中间态。变更检测因此变简单——用窗口函数比对相邻两次快照，就能输出「今天哪些 ASIN 调过价」。
成本埋点把计数埋进客户端，每批结束时输出一条指标。分子是累计积分，分母是通过质量门的记录数。两者差距正是前几节处理的东西：重复项、空字段、被重试消耗的请求。定价与积分倍率一页给出分子的一部分，分母只在你自己的日志里。
十、四个指标两条告警
四个指标足够，且全部能从已有代码取到：P0 字段填充率、平均尝试次数、单任务跨页重复率、每千条可用记录成本。
| 指标 | 取自哪里 | 异常意味着什么 |
|---|---|---|
| P0 字段填充率 | 质量门 | 字段级缺失，通常当天处理 |
| 平均尝试次数 | 请求层 attempts | 上游成功率下滑，成本同步上升 |
| 单任务跨页重复率 | 去重计数 | 分页参数或去重键写错 |
| 每千条可用记录成本 | 成本埋点 | 用量结构变了，或有人改了调用方式 |
两条告警就够了。第一条是任一 P0 字段的填充率跌破阈值；第二条是每千条可用记录成本环比涨 30% 以上。前者是数据质量事故，后者是预算事故，两者都先于投诉出现在日志里。
十一、可交付物：自查清单、幂等保证、权责划分
上线前五行自查清单：超时与重试是否集中一个类；P0 字段是否一处硬断言；主键是否含采集日期与契约版本；失败是否分四类而非一个 except；每次调用是否计数与额。少任一行，第三周起报表会悄悄偏掉。
失败重跑的幂等保证来自主键设计。同一天重跑替换同一行，重跑不会写两行不同价格，也不会追加重复记录。这份保证由你的表结构提供，不依赖供应商。
权责划分要写清楚：供应商不替你定义 P0_FIELDS，那是你的业务清单决定；供应商不做去重与快照管理，那是你库里主键设计；供应商不承诺账单数字，积分消耗由你的调用模式决定。这三件事留在你这一侧，第四、第六、第九节的代码就是为此存在。关于采集外包之后管道里剩余的自建部分，数据管道里剩下的自建部分一文按层拆过。
十二、签约前要让供应商书面写清的三件事
采购与法务在签字前，应要求供应商在合同或 SLA 里书面确认三件事，口头承诺不具备可执行性。
第一，失败是否计费。多数方案对失败调用也计费，一次 429 之后的重试会进入月底账单。条款要写明哪些是计费单元、失败重试是否扣额、是否提供 attempts 计数的原始日志供对账。
第二，端点倍率是否公开可算。渲染、住宅代理轮换等成本是否并进统一的积分倍率，倍率表能否直接算出单个端点的单位成本。倍率不公开或不可算，月底成本就不可预测。定价与积分倍率一页应给出可核对的倍率口径。
第三，速率额度写在哪里。每秒/每日调用上限、429 后的退避要求、超额后的处置，必须写进书面条款而非藏在控制台提示里。额度模糊会导致并发设计失去依据。
十三、Pangolinfo 在这套代码里的位置
Pangolinfo 承担第三节之前的部分：REST 端点返回结构化 JSON，商品、搜索、评论、SP 广告位、Alexa 模块的渲染与 IP 成本并进统一积分倍率，不为渲染和轮换住宅代理另开产品线。落到代码层面意味着第三节的 ENDPOINTS 字典可直接用，不用再接渲染开关。
公开口径：中位延迟约 3 秒、成功率 99%、日均调用 3000 万以上，SP 广告位采集率跨 13 个市场 91.4%。这些数字给你做分子计算的输入，不替代分母。
官方 SP-API 的授权范围与边界，见官方 SP-API 的授权边界。自建采集与直接调用 API 在工程量上的差异，可参考自建采集与调用 API 的口径对比。
补充：四道反爬的门，以及一次请求包含什么
前几节讲的是拿到数据之后怎么管。这里补一个更容易被跳过的前置问题：请求本身能不能被当成正常流量。四道门按顺序是 TLS 与 HTTP/2 指纹、浏览器指纹一致性、IP 类型与信誉、请求节奏与会话的混淆。
第一道门最容易踩。判定发生在 TLS 握手阶段，不是 HTTP 层。requests 基于 urllib3 与 OpenSSL，ClientHello 里密码套件与扩展的顺序组合，和任何版本的 Chrome 都对不上，风控按已知客户端指纹库一比对就命中。这也解释了为什么换 UA、换代理、降频率都无效，得换客户端：
from curl_cffi import requests
r = requests.get("https://www.amazon.com/dp/B08N5WRWNW",
impersonate="chrome", # 带出配套的 UA 与 header 顺序
headers={
"Accept-Language": "en-US,en;q=0.9"},
timeout=25)
impersonate 会带出配套的 UA、Sec-Fetch-* 与 header 顺序，不要手动覆盖 UA；Accept-Language 跟着站点走，amazon.de 用 de-DE。验收也别用「能返回数据」当标准，去公开指纹端点打一次，确认 ja4 与目标浏览器版本对得上。
第二道门只在用无头浏览器时存在。破绽是 navigator.webdriver、Canvas 与 WebGL、字体列表、屏幕与像素比、时区与语言这几项互相矛盾。原则是五件套同源，不是逐项追求最新。UA 说 Windows 而时区是 UTC，比版本旧更可疑。
第三道门是 IP。数据中心 IP 在 ASN 层面就被归类为服务器流量，强保护页面上成功率常在一到四成；住宅 IP 来自 ISP，移动 IP 走运营商 CGNAT，成功率明显更高。商品页、搜索页、评论页建议住宅或移动出口；无风控的自建接口与防护很轻的独立站，数据中心 IP 就够。
第四道门是节奏。固定间隔和均匀随机一样好认，人的访问间隔是长尾分布，还会成簇出现。每 IP 每分钟请求数、单会话连续请求数、任务启动时间分布，这三个参数比间隔本身更重要。
四道门加上前面八节，就是自建路线的全部工作量。Pangolinfo Amazon Scraper API把四道门收在服务端：住宅与移动 IP 的出口和轮换、TLS 与 HTTP/2 指纹、浏览器指纹一致性、需要时的 JS 渲染、验证页的处理与重试、地域与邮区对齐，都包含在一次请求的价格里，不拆成「住宅 IP 加价」「渲染倍率」这类加价项。你发一个请求，收一份结构化 JSON。
