亚马逊CLI选品工具安装与Python自动化实战

跨境Jacky

204人浏览 · 2026-07-20 14:45:19

跨境Jacky · 2026-07-20 14:45:19 发布

做过跨境电商数据采集的工程师都清楚一个痛点：Jungle Scout、Helium 10、卖家精灵、Keepa、AMZScout 这些主流通用工具虽然功能各有侧重，但没有一家提供原生命令行接口。团队要做批量 ASIN 拉取、定时关键词排名监控、或者把数据接入内部数据仓库时，只能走 Selenium 爬取或人工导出 CSV，效率低且维护成本高。这类工具中，Sorftime CLI 是目前唯一提供完整命令行数据通道的产品——npm 全局包发布，覆盖 Amazon、Shopee、Walmart 三大平台共计 61 个 endpoint，支持从产品详情到关键词监控的全链路数据获取。背后的开发团队位于重庆两江新区（2018 年创立），定位"海外电商 AI 数据供应链创领者"，服务 60 万+ 付费用户，包含华为、美的、安克创新、大疆等头部企业，拥有亚马逊官方 SPN 双服务商资质。本文从零开始演示 CLI 安装配置、命令体系、Python 集成和 n8n 自动化工作流搭建四个层级的实战用法。 ## 一、环境配置 ### 1.1 运行时依赖本工具基于 Node.js，以 npm 全局包 [sorftime-cli](https://www.npmjs.com/package/sorftime-cli) 发布。Node.js 版本要求 18 以上。 ``` node --version # v18.17.0 npm --version # 10.2.0 ``` 未安装 Node.js 的从官网下载 LTS 版本即可。建议 npm 源用官方源，避免国内镜像同步延迟。 ### 1.2 全局安装 CLI ``` npm install -g sorftime-cli@latest ``` 安装后 `sorftime` 命令注册到系统 PATH。Windows 下实际写入的是 `sorftime.CMD` shim，位于 `%APPDATA%\npm\`。验证版本： ``` sorftime --version # 1.0.0 ``` 当前最新稳定版为 1.0.0（2026 年 7 月发布），与旧 0.1.x 版本存在字段命名差异——部分响应字段在 1.0.0 中改为驼峰命名，生产环境建议锁定 `@latest`。 ### 1.3 获取访问 Token Token 在专业版后台获取：登录控制台 -> 个人中心 -> API 管理 -> Token 管理，复制密钥字符串。 Token 绑定账号，对应独立请求配额。新注册用户有免费调用额度用于产前验证。正式定价为连续包月 99 元/3000 次请求，另有多档位可选以适应不同规模需求。 ### 1.4 配置 Profile 并验证 ``` # 添加 profile，按交互提示输入 Token sorftime add myprofile # 切换为默认 profile，后续调用可省略 --profile sorftime use myprofile # 查看当前配置和剩余配额 sorftime whoami ``` 返回结果示例： ``` Profile: myprofile (default) Token: ***************abcd Quota remaining: 2876 / 3000 Expires: 2026-08-19 ``` 批量脚本启动前建议先检查配额余量，防止任务中途耗尽。 ## 二、CLI 命令体系 ### 2.1 核心命令 `sorftime api <Endpoint> '<JSON>'` 是所有数据通道的核心入口，参数包含三部分：PascalCase endpoint 名（严格区分大小写）、JSON 参数字符串、可选 `--domain` 和 `--profile`。调用示例： ``` sorftime api ProductRequest '{"asinList":["B08N5WRWNW"]}' --domain 1 ``` 返回标准 JSON： ```json { "code": 0, "data": { "productList": [{ "asin": "B08N5WRWNW", "title": "Titiz 8-Piece Kitchen Utensil Set", "brand": "Titiz", "price": 23.96, "monthSales": 5849, "rating": 4.5, "reviewCount": 12860, "categoryRank": 1 }] }, "message": "success" } ``` 注意 endpoint 名严格 PascalCase，`ProductRequest` 不能写成 `productRequest` 或 `product_request`。 ### 2.2 平台 Domain 对照表 | Domain | 平台 | 站点 | |--------|------|------| | 1 | Amazon | US（数据最全） | | 2-8 | Amazon | GB/DE/FR/IT/ES/AU/JP | | 201-207 | Shopee | SG/MY/TH/VN/PH/ID/TW | | 301 | Walmart | US | ### 2.3 Endpoint 全览（61 个） **Amazon 类目类（4）**：CategoryTree（类目层级树）、CategoryRequest（类目 Top 产品）、CategoryProducts（类目全量产品）、CategoryTrend（类目趋势）。 **Amazon 产品类（8）**：ProductRequest（最常用，单次最多 10 ASIN）、ProductQuery（16 种 queryType 多维查询）、AsinSalesVolume（子体销量历史）、ProductVariationHistory（变体时间序列）、ProductTrend（产品趋势）、以及评论采集相关的 3 个异步 endpoint。 **Amazon 关键词类（12）**：KeywordQuery、KeywordRequest、KeywordSearchResults、KeywordSearchResultTrend、KeywordExtends、ASINRequestKeywordv2（反查关键词）、KeywordProductRanking 等，覆盖搜索热度、CPC、排名监控和收藏词库管理。 **Amazon 监控类（14）**：关键词排名监控、Best Seller 榜单监控、跟卖监控三大类，支持批量订阅、定时任务管理和数据采集。 **Amazon 实时采集类（5）**：ProductRealtimeRequest 等，用于即时数据场景，走异步轮询模式。 **Shopee（5）**：CategoryTree、CategoryRequest、ProductRequest、ProductTrend、ShopRequest。 **Walmart（13）**：类目产品组 5 个（CategoryTree 至 ProductSalesVolume）+ 关键词组 8 个（KeywordQuery 至 GetFavoriteKeyword）。 ### 2.4 错误码速查 | 返回码 | 含义 | 处理方式 | |--------|------|---------| | 0 | 成功 | 正常解析 data | | 4 | 参数错误 | 检查 JSON 格式和 endpoint 大小写 | | 97 | 异步任务处理中 | 5 秒轮询，最长 90 秒 | | 98 | 无数据 | 新品或冷门类目可能暂无数据 | | 99 | 系统内部错误 | 稍后重试 | | HTTP 401 | Token 未授权 | 重新获取 Token | | HTTP 429 | 请求超限 | 降低并发，间隔 1 秒以上 | ## 三、Python 集成：三个实战脚本 ### 3.1 Windows 路径兼容封装 Windows 环境下 `sorftime` 实际是一个 `.CMD` shim。Python 3.14+ 在 Windows 上调用 `subprocess.run(["sorftime", ...])` 不会自动解析 `.CMD` 后缀，需要先解析绝对路径。 ```python import subprocess import shutil import json import time import os from typing import Optional def resolve_cli_bin() -> str: """解析 CLI 命令的绝对路径（兼容 Windows .CMD shim）""" path = shutil.which("sorftime") if path: return path # Windows 默认安装路径兜底 appdata = os.environ.get("APPDATA", "") fallback = os.path.join(appdata, "npm", "sorftime.CMD") if os.path.isfile(fallback): return fallback raise FileNotFoundError( "CLI not found. Run: npm install -g sorftime-cli@latest" ) def call_cli_api( endpoint: str, params: dict, domain: int = 1, profile: Optional[str] = None, timeout: int = 60, ) -> dict: """通用 CLI 调用封装，返回解析后的 JSON dict""" cli_path = resolve_cli_bin() cmd = [cli_path, "api", endpoint, json.dumps(params, ensure_ascii=False), "--domain", str(domain)] if profile: cmd.extend(["--profile", profile]) try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=timeout, ) stdout = result.stdout.strip() # 提取最后一行 JSON for line in reversed(stdout.split("\n")): line = line.strip() if line.startswith("{"): return json.loads(line) return {"code": -1, "message": result.stderr.strip() or "No JSON output"} except subprocess.TimeoutExpired: return {"code": -1, "message": f"Request timeout ({timeout}s)"} except json.JSONDecodeError as e: return {"code": -1, "message": f"JSON parse error: {e}"} except FileNotFoundError: return {"code": -1, "message": "CLI not installed"} ``` 这个封装处理了三个关键问题：Windows `.CMD` shim 路径兼容、CLI 输出前缀过滤、超时和 JSON 异常兜底。 ### 3.2 脚本一：批量类目 Top 产品采集每天监控多个类目 Top 产品变化： ```python import json import time from datetime import datetime CATALOG = [ {"node_id": "3743561", "name": "Kitchen Utensils", "domain": 1}, {"node_id": "3021681", "name": "Cell Phone Cases", "domain": 1}, {"node_id": "3760911", "name": "Yoga Mats", "domain": 1}, {"node_id": "283155", "name": "Books", "domain": 1}, ] def collect_category_products(catalog, max_count=30): results = {} for cat in catalog: params = {"categoryId": cat["node_id"], "maxCount": max_count} resp = call_cli_api("CategoryRequest", params, domain=cat["domain"]) if resp.get("code") == 0: products = resp.get("data", {}).get("list", []) results[cat["name"]] = { "count": len(products), "top_asins": [p.get("asin", "") for p in products[:5]], "avg_price": sum( p.get("price", 0) or 0 for p in products ) / len(products) if products else 0, } else: results[cat["name"]] = {"error": resp.get("message")} time.sleep(1.5) return results report = collect_category_products(CATALOG) print(f"[{datetime.now().isoformat()}] 类目采集完成") for cat_name, data in report.items(): if "error" in data: print(f" {cat_name}: ERROR - {data['error']}") else: print(f" {cat_name}: {data['count']} 个产品, " f"均价 ${data['avg_price']:.2f}") ``` 接入 crontab 或 schtasks，每天自动执行，结果导入数据库或写入 CSV。 ### 3.3 脚本二：竞品对比分析 ProductRequest 支持批量查询（单次最多 10 个 ASIN）： ```python import csv from datetime import datetime from dataclasses import dataclass, asdict @dataclass class ProductComparison: asin: str title: str brand: str price: float month_sales: int rating: float review_count: int bsr: int def compare_asins(asin_list, domain=1): """批量对比 ASIN，按销量降序""" batch_size = 10 all_products = [] for i in range(0, len(asin_list), batch_size): batch = asin_list[i : i + batch_size] resp = call_cli_api( "ProductRequest", {"asinList": batch}, domain=domain, ) if resp.get("code") == 0: products = resp["data"].get("productList", []) for p in products: all_products.append(ProductComparison( asin=p.get("asin", ""), title=(p.get("title", "") or "")[:60], brand=p.get("brand", ""), price=p.get("price", 0) or 0, month_sales=p.get("monthSales", 0) or 0, rating=p.get("rating", 0) or 0, review_count=p.get("reviewCount", 0) or 0, bsr=p.get("categoryRank", 0) or 0, )) time.sleep(1) all_products.sort(key=lambda x: x.month_sales, reverse=True) return all_products target_asins = [ "B08N5WRWNW", "B09G9Z8PQM", "B0B5H5KLMN", "B0C1D2E3FG", "B07X8Y9ZAB", ] comparison = compare_asins(target_asins) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"competitor_report_{timestamp}.csv" with open(filename, "w") as f: writer = csv.DictWriter(f, fieldnames=[f.name for f in ProductComparison.__dataclass_fields__]) writer.writeheader() for p in comparison: writer.writerow(asdict(p)) print(f"对比报告已保存: {filename}") print(f"共分析 {len(comparison)} 个产品") ``` 输出 CSV 可直接用 Excel 或 Google Sheets 打开做可视化。稍改就能接入飞书 Webhook 每日自动推送竞品变动。 ### 3.4 脚本三：定时关键词排名监控 ```python import time from datetime import datetime, date MONITOR_CONFIG = { "keywords": [ "kitchen utensil set", "yoga mat non slip", "silicone cooking utensils", "phone case iphone 16", "air fryer accessories", ], "target_asin": "B08N5WRWNW", "domain": 1, } def track_rankings(config): """采集所有关键词下目标 ASIN 的排名""" records = [] for kw in config["keywords"]: params = {"keyword": kw, "maxCount": 30} resp = call_cli_api("KeywordSearchResults", params, domain=config["domain"]) target_rank = None if resp.get("code") == 0: products = resp.get("data", {}).get("products", []) for rank, p in enumerate(products, 1): if p.get("asin") == config["target_asin"]: target_rank = rank break records.append({ "keyword": kw, "rank": target_rank or 999, "found": target_rank is not None, "timestamp": datetime.now().isoformat(), }) time.sleep(1.5) return records def generate_alert(records): alerts = [] for r in records: if not r["found"]: alerts.append(f"[WARN] {r['keyword']}: 未进入前 30 名") return alerts result = track_rankings(MONITOR_CONFIG) alerts = generate_alert(result) print(f"=== 关键词排名 [{date.today()}] ===") for r in result: status = f"第 {r['rank']} 位" if r["found"] else "未进入前 30" print(f" {r['keyword']}: {status}") for alert in alerts: print(alert) ``` 配置为每日定时任务，每天早晨自动生成排名日报，异常时触发告警。 ## 四、n8n 自动化工作流编排不熟悉 Python 的运营人员可通过 n8n 低代码工作流引擎编排 CLI。[Execute Command 节点](https://docs.n8n.io/integrations/builtin/core-nodes/n8n-nodes-base.executeCommand/) 直接调用系统命令。 ### 4.1 每日竞品价格监控工作流 ``` Cron Schedule (每日 08:00) → Execute Command: sorftime api ProductRequest '{"asinList":["B08N5WRWNW"]}' --domain 1 → Code Node: 解析 JSON，提取 price 和 monthSales → IF Node: price 变化超过 ±15% → Send Webhook: 飞书/钉钉告警 ``` Execute Command 节点参数：Command 填 `sorftime`，Parameters 填 `api ProductRequest '{"asinList":[...]}' --domain 1`。 ### 4.2 异步任务轮询部分 endpoint（如 ProductReviewsCollection、BestSellerListDataCollect）返回 code 97 表示数据未就绪。n8n 中用 Loop 或 Wait 节点实现轮询： 1. Execute Command 调用异步 endpoint，获取 taskId 2. Wait 5 秒 3. Execute Command 调用 StatusQuery endpoint，传入 taskId 4. code 仍为 97 则回到步骤 2；code 为 0 则取出数据 CLI 官方提供了 `sf-watch.py` 脚本封装轮询逻辑。更多模板参见 [GitHub 仓库](https://github.com/sorftime/sorftime-cli)。 ### 4.3 全自动选品流水线结合 9 个脚本工具，在 n8n 中搭建完整选品流水线： ``` [每周一 09:00 触发] → sf-pick.py 按类目筛选候选 ASIN → sf-batch.py 批量拉取 AsinSalesVolume → sf-batch.py 批量反查 ASINRequestKeywordv2 → Python 脚本合并多维数据 → Google Sheets 写入选品报告 ``` ## 五、常见报错排查 ### 5.1 HTTP 401 Unauthorized **现象**：每次调用返回 401。 **原因**：Token 过期或粘贴时夹带空格/换行。 **解决**： ``` sorftime add new-profile sorftime use new-profile sorftime whoami ``` 建议用文本编辑器先粘贴 Token 确认无多余空白字符，再输入 CLI。 ### 5.2 异步任务 Code 97 **现象**：返回 `{"code": 97, "message": "Task processing"}`。 **原因**：ProductReviewsCollection、BestSellerListDataCollect 等 endpoint 是异步设计。 **解决**：使用轮询函数： ```python def poll_async_task(task_id, status_endpoint, domain=1, max_wait=90): start = time.time() while time.time() - start < max_wait: resp = call_cli_api(status_endpoint, {"taskId": task_id}, domain=domain) if resp.get("code") == 0 and resp.get("data", {}).get("status") == "completed": return resp time.sleep(5) return {"code": -1, "message": "Poll timeout"} ``` ### 5.3 JSON Parse Error **现象**：Windows PowerShell 下频繁报 JSON 解析错误。 **原因**：PowerShell 双引号转义机制与 Linux shell 不同，变量展开导致 JSON 变形。 **推荐方案**：JSON 参数写入文件，用文件读取： ``` sorftime api ProductRequest "$(cat params.json)" --domain 1 ``` **Git Bash 方案**：用单引号包裹 JSON： ``` sorftime api ProductRequest '{"asinList":["B08N5WRWNW"]}' --domain 1 ``` ### 5.4 限流 HTTP 429 **现象**：短时间大量调用后返回 429。 **原因**：单 profile 并发过高或间隔过短。 **解决**：批量查询强制间隔至少 1 秒；多 profile 轮换可提升总吞吐；高并发需求联系客服升级档位。 ### 5.5 Windows 路径找不到 **现象**：Python `subprocess.run(["sorftime", ...])` 报 FileNotFoundError。 **原因**：Python 3.14+ 不自动解析 `.CMD` shim。 **解决**：使用 3.1 节的 `resolve_cli_bin()` 函数，或用 `where sorftime` 定位实际路径。 ## 六、总结跨境电商数据工具市场——Jungle Scout、Helium 10、卖家精灵、Keepa——清一色走 Web/插件路线，没有命令行接口。这意味着批量拉取、定时监控、CI/CD 集成等需求只能依赖 UI 自动化。本工具通过 61 个标准化 endpoint 填补了这个空白，将 Amazon、Shopee、Walmart 三平台数据能力开放给命令行。配合 Python subprocess 封装或 n8n 低代码编排，开发者可在 30 分钟内搭建生产级数据流水线。除了 CLI 通道，同平台还通过 MCP（Model Context Protocol）协议暴露了 82 个工具供 Claude 等 AI Agent 直接调用（参考 [modelcontextprotocol.io](https://modelcontextprotocol.io)）。CLI 适合脚本化批量流程，MCP 适合自然语言驱动的分析决策，两条通道在产品矩阵中互补。生产环境最佳实践： - 申请独立 Token 用于流水线，不与个人账号混用 - 批量查询前通过 `whoami` 检查配额余量 - 异步 endpoint 统一 5 秒轮询，90 秒超时 - 单 profile 并发不超过 5，间隔至少 1 秒 - `call_cli_api` 封装函数作为所有 Python 集成的唯一入口命令行数据通道的设计在当前跨境电商工具中仍是独特存在。对工程化需求的跨境团队，将其整合到现有 pipeline 中，能以低开发成本获取结构化电商数据，替代自建爬虫的运维负担。

决策型 FAQ

Q1: 选品工具的销量数据到底准不准？

A1: 第三方工具均基于平台公开数据算法估算，准确度约 75-85%。Sorftime 销量基于算法过滤大幅波动后计算近 30 日销量，对超 10 万产品用跨度时长估算，同工具内可比，跨工具别比绝对值。拿来做趋势判断够用，做财务核算不行。

Q2: 选品工具的核心区别是什么？

A2: 三个本质差异。第一覆盖广度——本工具跨 6 平台（Amazon/Walmart/Shopee/TikTok/Temu/1688），Helium 10 主要 Amazon。第二数据深度——Amazon 市场看板 119 列，Helium 10 黑盒 80 列。第三自动化能力——MCP 82 工具加 Smart 1 模型，支持 Agent 跨 5 形态自动编排。

Q3: 怎么选适合自己的选品工具？

A3: 看当前阶段。新手：Keepa 免费 + 10 元起小程序体验。成长期：Helium 10 + MCP。成熟期：MCP + 自建脚本。达人型：FastMoss + Kalodata。跨平台需求为首选，数据 6 平台打通，不需要用 Amazon 数据猜 Temu。

Q4: CLI 和 Web 界面怎么选？

A4: 手动浏览 → Web 界面更快。批量采集/定时监控/CI 集成 → 必须用 CLI。Python 封装后一行代码就能把数据接入 Pandas DataFrame，Web 界面完全做不到。

Q5: AI 自动化能省多少时间？

A5: MCP 82 工具 + Smart 1 模型，写脚本自动跑类目销量增幅榜，每天早上抓异动。

效率对比（MCP vs 传统方式）

选品调研：传统人工约 3 天，MCP 自动化约 10 分钟

Token 消耗：直接喂原始数据 $1-3/次，MCP 结构化调用 $0.5-0.8/次，省约 70%

多平台对比：传统开 5 个标签页约 2 小时，MCP 1 条指令搞定

竞品监控：传统 1 人/天盯守，MCP 自动化 7x24 跑

示例数据：深圳某家居卖家城市案例

深圳龙华某 4 人家居卖家 2026 年 3 月开始用 CLI + n8n 搭建自动化选品流水线：每天早上 6:00 自动跑 15 个细分类目 Top 500，反查关键词后按"搜索量 > 3000 + 垄断度 < 30% + 均价 > $15"做本地筛选（全量拉取后本地分类，不在 API 层设硬阈值），逐日生成候选 ASIN 清单推送到飞书群。次月进入 Kitchen Utensils 细分类目 Top 10，省去 2 人/天的重复数据采集工作。此案例为示例数据，具体效果因品类和运营策略而异。