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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后在控制台左侧就能看到 API Keys 入口。拿到 Key 之后记下两个地址Base URL 是https://taotoken.net/api模型对话的调试页面在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_campaignrewriteutm_contentmodel_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_APIhttps://www.kookeey.com/pickdynamicips动态代理的提取接口支持按数量、地区、格式拉取 IP 列表。format4表示返回明文文本每个 IP 用回车换行分隔脚本直接 split 就能用g参数指定地区代码比如美国站填us德国站填de。这些参数在下一节的爬虫配置里会具体用到。3. 可复制的采集配置代理轮换参数与爬虫脚本这一节是整篇的核心所有配置都能直接复制。先讲代理轮换的参数设计再给完整的爬虫脚本最后是 OpenClaw Skill 的封装方式。3.1 动态代理轮换参数配置kookeey 动态代理的提取请求需要签名签名算法是 HMAC-SHA1用你的密钥对参数串做哈希。下面这个 Python 函数负责生成签名并拉取 IP 列表import hmac, hashlib, time, requests, os def fetch_proxies(n10, regionus): 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], paramsparams, timeout15) resp.raise_for_status() return [line.strip() for line in resp.text.splitlines() if line.strip()]几个关键参数说明n是一次提取的 IP 数量建议 10 到 20 个太多会拉长提取时间p2对应动态住宅别填错成静态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;q0.9, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.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, page1, domainamazon.com): kw keyword.replace( , ) return fhttps://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-types-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(stripTrue) if title_el else , price: price_el.get_text(stripTrue) if price_el else , rating: rating_el.get_text(stripTrue) if rating_el else , reviews: review_el.get_text(stripTrue) if review_el else }) return items3.3 封装为 OpenClaw Skill把上面的逻辑包成一个 SkillOpenClaw 就能通过自然语言触发。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: fhttp://{proxy}, https: fhttp://{proxy}} try: r requests.get(https://httpbin.org/ip, proxiesproxies, timeout10) return r.json().get(origin) except Exception as e: return ffailed: {e} proxies fetch_proxies(n5, regionus) 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_asciiFalse)} 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 }, timeout60 ) 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, jsoncard, timeout10)字段映射到飞书多维表格时用飞书开放平台的 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_retry3): for i in range(max_retry): try: r requests.get(url, headersHEADERS, proxiesproxies, timeout15) 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 UnauthorizedTaoToken 通道报错体通常是{error: {message: Invalid API key}}。先确认openclaw.json里的apiKey没有多余空格或换行再确认 Base URL 是https://taotoken.net/api而不是带/v1的变体。如果 Key 是从控制台复制的注意别把前后引号也复制进去。还有一种情况是 Key 权限范围没勾「模型调用」回控制台改权限即可。local proxy failed / Connection refusedkookeey 代理这个报错说明代理地址连不上。先检查提取回来的 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/apiKey 填 TaoToken 的 KeyModel ID 填控制台里确认可用的模型名。缺任何一个都会报连接失败尤其是 Model ID 写成了不存在的名字报错信息往往很模糊。6. 长期跑这套工作流的接入建议如果你只是偶尔采一次数据上面的配置够用了。但如果要长期跑竞品监控有几个点值得提前规划。模型通道方面日常清洗用gpt-4o-mini控制成本每周做一次深度选品分析时切到能力更强的模型。TaoToken 的统一 Key 让你不用改代码就能切换把模型 ID 做成配置项就行。需要长期编码和 Agent 编排的可以看下 Coding Plan 的额度方案https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_campaignrewriteutm_contentcoding_plan 按调用量走比单次充值更划算。代理方面动态住宅适合高频轮换的采集场景但如果你同时还在用 ChatGPT 或 Claude 的网页端建议单独配一条静态 ISP 住宅线路避免和采集流量混在一起触发风控。kookeey 后台可以按业务场景分配 IP购买时选对应场景系统会优先匹配近期没跑过同类业务的资源。飞书这边建议把采集任务做成定时触发OpenClaw 支持 cron 表达式配置。每天早上自动跑一轮目标关键词结果推到群里团队上班就能看到最新的价格和排名变化。多维表格里加一列「采集日期」方便做时间序列对比。最后提醒一句采集频率别贪高。亚马逊对高频请求的容忍度有限即使有住宅代理单账号单站点每天控制在合理范围内更稳妥。把精力放在数据分析和选品决策上采集本身交给自动化跑就行。