1. 亚马逊数据采集的真实困境与这套方案能解决什么
做亚马逊选品的朋友大概率都遇到过这种场景：手动翻竞品 listing 翻到眼花，好不容易整理出一份表格，过两天价格和排名全变了；写了个 Python 脚本想自动抓，跑不到两百条请求就开始弹验证码，换个 IP 继续跑，第二天发现整个 C 段都被限流。更麻烦的是，数据抓回来是一堆原始 JSON 和 HTML，还得人工清洗、归类、提炼，最后再手动贴到飞书表格里同步给团队——整条链路里真正有价值的分析环节，反而被大量重复劳动挤占了。
这套方案要解决的就是这条链路上的三个断点。第一是出口 IP 的稳定性：用 kookeey 的动态住宅代理做请求轮换，让每次抓取都从真实家庭网络出口发出，避开数据中心 IP 段被平台重点关照的问题。第二是任务编排的自动化：用 OpenClaw 把「接收指令 → 调代理 → 跑爬虫 → 调模型清洗 → 推送飞书」串成一个可复用的 Skill，你只需要发一句自然语言，剩下的它自己跑。第三是模型调用的统一入口：采集回来的原始数据要清洗、去重、生成摘要和选品洞察，这些都要调大模型，通过 TaoToken 的统一 Key 和 API 通道接入，不用在多个厂商的 Key 之间来回切换，也不用为每个模型单独维护一套调用代码。
适合谁看？如果你在做跨境电商选品、竞品监控，或者单纯想搭一套「采集 + AI 分析 + 协同推送」的自动化工作流，这篇的配置和代码可以直接抄。如果你只是想了解动态代理和 Agent 编排怎么配合，第三、四节的参数和排错部分也够用。下面从环境准备开始，一步步把整条链路跑通。
2. TaoToken 统一接入与 OpenClaw 环境前置配置
在动手写爬虫之前，先把「大脑」和「通道」准备好。OpenClaw 本身不带模型，它需要一个能稳定调用的模型接口来理解指令、拆解任务、生成分析结论。这里用 TaoToken 做统一接入层，好处是你可以在一个 Key 下切换不同模型，采集清洗用便宜快的，深度洞察用能力强的，不用改代码。
2.1 获取 TaoToken API Key 与模型通道
先到 TaoToken 控制台创建 API Key。登录后进入 API Keys 页面，点新建，把权限范围设成「模型调用」，复制生成的 Key 保存好——这个 Key 只显示一次。如果你还没账号，从官网入口进：https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，注册后在控制台左侧就能看到 API Keys 入口。
拿到 Key 之后，记下两个地址：Base URL 是https://taotoken.net/api，模型对话的调试页面在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_campaign=rewrite&utm_content=model_chat 对应的控制台里可以先用网页版试跑一条请求，确认 Key 有效再写进配置。模型 ID 方面，清洗和摘要任务用gpt-4o-mini这类性价比高的就够，深度选品分析可以切到claude-3-5-sonnet或deepseek-chat，具体可用列表在控制台的模型列表里查。
2.2 OpenClaw 安装与模型配置
OpenClaw 支持三大平台，用 npm 全局装：
npm install -g openclaw openclaw --version装完后初始化配置目录，默认在~/.openclaw/。核心配置文件是openclaw.json，模型通道写在这里：
{ "models": { "default": { "provider": "openai-compatible", "baseUrl": "https://taotoken.net/api", "apiKey": "sk-你的TaoToken密钥", "modelId": "gpt-4o-mini" }, "analysis": { "provider": "openai-compatible", "baseUrl": "https://taotoken.net/api", "apiKey": "sk-你的TaoToken密钥", "modelId": "claude-3-5-sonnet" } } }这里default用于日常指令理解和任务拆解，analysis用于采集后的深度分析。两个通道共用同一个 Key，切换模型只改modelId。配置写完后跑一条测试：
openclaw model test --config ~/.openclaw/openclaw.json返回model response ok就说明通道通了。如果报 401，先检查 Key 有没有多余空格；如果报连接超时，确认 Base URL 结尾没有多写/v1——TaoToken 的兼容接口路径已经内置，直接填https://taotoken.net/api即可。
2.3 kookeey 动态代理凭证准备
代理这块，登录 kookeey 后台，进入「动态住宅代理」页面，点提取动态 IP，生成一组连接参数。你会拿到accessid、signature密钥、以及 API 提取地址。把这些存成环境变量，别硬编码进脚本：
export KOOKEEY_ACCESSID="你的accessid" export KOOKEEY_SECRET="你的签名密钥" export KOOKEEY_API="https://www.kookeey.com/pickdynamicips"动态代理的提取接口支持按数量、地区、格式拉取 IP 列表。format=4表示返回明文文本，每个 IP 用回车换行分隔，脚本直接 split 就能用；g参数指定地区代码，比如美国站填us，德国站填de。这些参数在下一节的爬虫配置里会具体用到。
3. 可复制的采集配置：代理轮换参数与爬虫脚本
这一节是整篇的核心，所有配置都能直接复制。先讲代理轮换的参数设计，再给完整的爬虫脚本，最后是 OpenClaw Skill 的封装方式。
3.1 动态代理轮换参数配置
kookeey 动态代理的提取请求需要签名，签名算法是 HMAC-SHA1，用你的密钥对参数串做哈希。下面这个 Python 函数负责生成签名并拉取 IP 列表：
import hmac, hashlib, time, requests, os def fetch_proxies(n=10, region="us"): accessid = os.environ["KOOKEEY_ACCESSID"] secret = os.environ["KOOKEEY_SECRET"] ts = str(int(time.time())) params = { "accessid": accessid, "ts": ts, "n": n, "p": "2", # 动态住宅代理类型 "g": region, # 地区代码 "format": "4", # 明文文本格式 "dl": "\r\n" } # 按参数名排序拼接后做 HMAC-SHA1 sign_str = "&".join(f"{k}={v}" for k, v in sorted(params.items())) signature = hmac.new(secret.encode(), sign_str.encode(), hashlib.sha1).hexdigest() params["signature"] = signature resp = requests.get(os.environ["KOOKEEY_API"], params=params, timeout=15) resp.raise_for_status() return [line.strip() for line in resp.text.splitlines() if line.strip()]几个关键参数说明：n是一次提取的 IP 数量，建议 10 到 20 个，太多会拉长提取时间；p=2对应动态住宅，别填错成静态；g按目标站点选，美国us、英国gb、德国de、日本jp。提取回来的 IP 列表放进一个队列，每次请求前 pop 一个，用完放回队尾，实现轮换。
轮换策略上，我实测下来「每请求一换」比「每 N 请求一换」更稳。亚马逊的风控对同一 IP 的连续请求很敏感，单 IP 连续超过 5 次搜索请求就容易触发验证。所以脚本里每次requests.get都从队列取一个新 IP。同时配合随机延迟：
import random, time time.sleep(random.uniform(1.5, 3.5))这个延迟模拟真人浏览的停顿，是绕过基础频率检测的关键，别省。
3.2 亚马逊搜索接口请求配置
亚马逊搜索页的请求头必须模拟真实浏览器。核心字段：
HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/122.0.0.0 Safari/537.36", "Accept-Language": "en-US,en;q=0.9", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Referer": "https://www.amazon.com/", "Connection": "keep-alive" }User-Agent用常见的 Chrome 版本，Accept-Language按目标站点设，美国站en-US，德国站de-DE。Referer设成亚马逊主站，让请求看起来是从站内跳转过来的。Cookie 方面，建议先访问一次首页拿初始 Cookie，再带着 Cookie 访问搜索页，维持会话连续性。
搜索请求的 URL 结构：
def build_search_url(keyword, page=1, domain="amazon.com"): kw = keyword.replace(" ", "+") return f"https://www.{domain}/s?k={kw}&page={page}"分页用page参数控制，每页大约 16 到 20 个商品。解析时优先用 BeautifulSoup 配合 CSS 选择器：
from bs4 import BeautifulSoup def parse_products(html): soup = BeautifulSoup(html, "html.parser") items = [] for card in soup.select("div[data-component-type='s-search-result']"): asin = card.get("data-asin", "") title_el = card.select_one("h2 span") price_el = card.select_one("span.a-price-whole") rating_el = card.select_one("span.a-icon-alt") review_el = card.select_one("span.a-size-base.s-underline-text") items.append({ "asin": asin, "title": title_el.get_text(strip=True) if title_el else "", "price": price_el.get_text(strip=True) if price_el else "", "rating": rating_el.get_text(strip=True) if rating_el else "", "reviews": review_el.get_text(strip=True) if review_el else "" }) return items3.3 封装为 OpenClaw Skill
把上面的逻辑包成一个 Skill，OpenClaw 就能通过自然语言触发。Skill 目录结构：
skills/amazon-scraper/ ├── SKILL.md ├── scraper.py └── config.jsonSKILL.md描述能力和参数：
# Amazon Scraper Skill ## 功能 采集亚马逊指定站点、指定关键词的商品列表数据。 ## 参数 - keyword: 搜索关键词 - region: 站点地区代码（us/gb/de/jp） - pages: 采集页数 ## 输出 CSV 文件，字段：asin, title, price, rating, reviewsconfig.json放环境变量映射，scraper.py是执行入口，接收参数、调代理、跑采集、存 CSV。封装完成后，在 OpenClaw 里发一句「帮我采集亚马逊德国站 kitchen utensils 前 5 页」，Agent 会自动解析出关键词、站点、页数，传给脚本执行。
4. 验证请求与飞书推送：从采集到协同的完整链路
配置写完必须验证，不然你不知道是代理没生效还是解析写错了。这一节给验证动作和飞书推送的完整配置。
4.1 代理连通性与采集结果验证
先单独验证代理是否生效。写个小脚本，通过代理请求一个显示出口 IP 的接口：
import requests def check_proxy(proxy): proxies = {"http": f"http://{proxy}", "https": f"http://{proxy}"} try: r = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=10) return r.json().get("origin") except Exception as e: return f"failed: {e}" proxies = fetch_proxies(n=5, region="us") for p in proxies: print(p, "->", check_proxy(p))如果每个代理返回的origin都不一样，说明轮换生效。如果全是同一个 IP，检查提取接口的n参数是不是被忽略了。
采集验证跑一条真实请求：
python scraper.py --keyword "yoga mat" --region us --pages 1成功的话会在当前目录生成amazon_products.csv，打开看应该有 16 到 20 行，字段完整。如果 CSV 是空的，大概率是 CSS 选择器失效——亚马逊的 class 名会变，用浏览器 F12 重新确认s-search-result这个data-component-type还在不在。
4.2 模型清洗与摘要生成
原始数据里有重复、有缺失、有格式不统一的价格。调 TaoToken 的模型通道做清洗：
import requests, json def clean_with_model(raw_items): prompt = f"""以下是亚马逊商品原始数据，请去重、补全缺失字段、 统一价格格式为数字，并输出 JSON 数组： {json.dumps(raw_items, ensure_ascii=False)}""" resp = requests.post( "https://taotoken.net/api/chat/completions", headers={"Authorization": "Bearer sk-你的TaoToken密钥"}, json={ "model": "gpt-4o-mini", "messages": [{"role": "user", "content": prompt}], "temperature": 0.2 }, timeout=60 ) return resp.json()["choices"][0]["message"]["content"]注意响应解析路径是choices[0].message.content，如果报reading 'choices'错误，说明返回体结构不对，先打印完整响应看是不是被限流或 Key 无效。
4.3 飞书 Webhook 与字段映射
飞书这边用自定义机器人 Webhook 最省事。在飞书群设置里添加「自定义机器人」，拿到 Webhook 地址。推送消息：
def push_to_feishu(webhook, records): card = { "msg_type": "interactive", "card": { "header": {"title": {"tag": "plain_text", "content": "亚马逊采集结果"}}, "elements": [ {"tag": "div", "text": {"tag": "lark_md", "content": f"共采集 {len(records)} 条商品数据"}} ] } } requests.post(webhook, json=card, timeout=10)字段映射到飞书多维表格时，用飞书开放平台的 Bitable API。先在多维表格里建好列：ASIN、标题、价格、评分、评论数、采集时间。然后调bitable/v1/apps/{app_token}/tables/{table_id}/records批量写入。字段名要和表格列名完全一致，否则会报field not found。
4.4 限速与失败重试验证
限速策略：单关键词单页请求间隔 2 到 4 秒，不同关键词之间间隔 5 到 8 秒。重试用指数退避：
def request_with_retry(url, proxies, max_retry=3): for i in range(max_retry): try: r = requests.get(url, headers=HEADERS, proxies=proxies, timeout=15) if r.status_code == 200: return r.text if r.status_code in (429, 503): time.sleep(2 ** i * 5) except requests.RequestException: time.sleep(2 ** i * 3) return None验证重试是否生效：故意把代理设成一个无效地址，跑一次采集，观察日志里有没有重试记录，最终是否优雅失败而不是直接崩溃。
5. 本篇常见报错排查：401、代理失败与解析异常
跑这套链路最容易卡在几个固定位置，下面按报错原文对照排查。
401 Unauthorized（TaoToken 通道）：报错体通常是{"error": {"message": "Invalid API key"}}。先确认openclaw.json里的apiKey没有多余空格或换行，再确认 Base URL 是https://taotoken.net/api而不是带/v1的变体。如果 Key 是从控制台复制的，注意别把前后引号也复制进去。还有一种情况是 Key 权限范围没勾「模型调用」，回控制台改权限即可。
local proxy failed / Connection refused（kookeey 代理）：这个报错说明代理地址连不上。先检查提取回来的 IP 格式是不是ip:port，有些接口返回带协议前缀，需要手动去掉。再确认代理是否已过期——动态住宅 IP 有存活时间，提取后放太久会失效，建议每次采集前重新提取。如果所有 IP 都连不上，检查accessid和签名密钥是否匹配，签名算法里参数排序错了也会导致提取失败。
reading 'choices' of undefined（模型响应解析）：这个错误出现在resp.json()["choices"]这一步，说明返回体里没有choices字段。常见原因是请求被限流返回了错误对象，或者模型 ID 写错了。先print(resp.text)看完整返回，如果是{"error": ...}就按错误信息处理；如果返回正常但结构不同，检查是不是用了非兼容格式的接口路径。
OAuth / token expired（飞书推送）：飞书自建应用的tenant_access_token有有效期，默认两小时。如果推送报 token 失效，需要在代码里加自动刷新逻辑，用 App ID 和 App Secret 重新换取。Webhook 方式不涉及这个问题，但 Webhook 只能发消息不能写多维表格，写表格必须走 API 加 token。
CSV 为空或字段缺失（解析层）：亚马逊的页面结构会调整，s-search-result这个选择器偶尔会变。排查方法是在浏览器里打开搜索页，F12 检查商品卡片的data-component-type属性值，同步更新脚本里的选择器。价格字段如果抓不到，检查是不是遇到了「暂无报价」的商品，这类卡片结构不同，需要单独处理。
CC Switch / Cline MCP / Codex auth.json 配置三件套：如果你在 OpenClaw 之外还用其他工具接同一套通道，配置时三件套必须齐全——Base URL 填https://taotoken.net/api，Key 填 TaoToken 的 Key，Model ID 填控制台里确认可用的模型名。缺任何一个都会报连接失败，尤其是 Model ID 写成了不存在的名字，报错信息往往很模糊。
6. 长期跑这套工作流的接入建议
如果你只是偶尔采一次数据，上面的配置够用了。但如果要长期跑竞品监控，有几个点值得提前规划。
模型通道方面，日常清洗用gpt-4o-mini控制成本，每周做一次深度选品分析时切到能力更强的模型。TaoToken 的统一 Key 让你不用改代码就能切换，把模型 ID 做成配置项就行。需要长期编码和 Agent 编排的，可以看下 Coding Plan 的额度方案：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_campaign=rewrite&utm_content=coding_plan ，按调用量走比单次充值更划算。
代理方面，动态住宅适合高频轮换的采集场景，但如果你同时还在用 ChatGPT 或 Claude 的网页端，建议单独配一条静态 ISP 住宅线路，避免和采集流量混在一起触发风控。kookeey 后台可以按业务场景分配 IP，购买时选对应场景，系统会优先匹配近期没跑过同类业务的资源。
飞书这边，建议把采集任务做成定时触发，OpenClaw 支持 cron 表达式配置。每天早上自动跑一轮目标关键词，结果推到群里，团队上班就能看到最新的价格和排名变化。多维表格里加一列「采集日期」，方便做时间序列对比。
最后提醒一句：采集频率别贪高。亚马逊对高频请求的容忍度有限，即使有住宅代理，单账号单站点每天控制在合理范围内更稳妥。把精力放在数据分析和选品决策上，采集本身交给自动化跑就行。
