1. TikTok 电商采集为什么需要一个 CLI Agent
做 TikTok 电商选品和竞品调研,最头疼的不是分析,而是数据怎么稳定拿到手。商品页、店铺页、关键词搜索页,结构各不相同,字段又多,手动复制粘贴根本不现实。我试过直接写 requests 脚本去抓,结果动态加载、人机验证、浏览器指纹轮番上阵,脚本跑十分钟就挂。后来我把思路换成「CLI 智能体 + 统一 Key 通道」:采集逻辑封装成命令行工具,鉴权和模型调用全部走 TaoToken 统一通道,一行命令拿到结构化 CSV,再顺手让模型生成分析报告。
这套方案适合三类人:一是做跨境电商选品、需要批量拉公开商品数据的运营;二是想学 Agent 工程化、但不想被鉴权和密钥管理拖住的后端/数据同学;三是已经在用 Claude Code、OpenClaw 这类 CLI 工具,想把采集能力接进自己工作流的开发者。核心检索词就三个:CLI、Python、TikTok 爬虫 Agent。它解决的问题很具体——把「配置密钥、发请求、拿快照、下数据、调模型分析」这条链路收敛成一条命令,配置只写一次,Key 只放一处。
我这次的做法是:Python 负责 CLI 框架和数据处理,采集接口负责拿公开商品数据,模型分析走 TaoToken 的统一 Key 通道。这样密钥不散落在各个脚本里,换模型、换 Key 只改一个配置文件。下面从环境准备到跑通验证,一步步来。
2. TaoToken 前置:统一 Key 通道怎么接
在动手写采集之前,先把模型调用这条线理顺。TaoToken 在这里扮演的角色是「统一 Key / API 通道」:你不需要在采集脚本里硬编码各家模型的密钥,而是把 base_url 指向 TaoToken 的 API 地址,用一把 Key 管理请求鉴权与配置。官网入口在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 地址是 https://taotoken.net/api (这个不加 UTM)。
为什么采集 Agent 要配这个?因为采集完数据后,分析环节要调模型。如果每个脚本各自维护一份 Key,改起来就是灾难。统一通道的好处是:CLI 里所有模型请求都走同一个 base_url 和同一把 Key,config.toml 里改一处,全链路生效。
你需要先拿到 Key。进入控制台创建 API Key,地址是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite 。创建完复制保存,后面写进配置文件。如果你还想先验证模型通不通,可以用模型对话页面快速试一条请求: https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 。接入细节和参数说明看文档: https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。
注意:Key 只放在本地 config.toml 或环境变量里,不要提交到 Git,也不要在采集脚本里明文写死。CLI Agent 的价值之一就是配置集中,别把这点优势浪费掉。
如果你后续要做长期编码或 Agent 任务,可以了解 Coding Plan: https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。Key 管理页在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite 。这些入口先记着,配置阶段会用到。
3. 可复制配置:config.toml 骨架与 CLI 启动
3.1 项目结构与依赖
先建目录,结构保持简单,三个职责分开:采集、下载、分析。
mkdir tiktok-agent && cd tiktok-agent python3 -m venv venv source venv/bin/activate pip install requests openai tomli目录长这样:
tiktok-agent/ ├── config.toml ├── tiktok_scraper.py ├── modules/ │ ├── collect.py # 提交采集任务,拿快照 ID │ ├── download.py # 按快照 ID 下载数据集 │ └── analyze.py # 调模型生成分析报告 └── input_sample.csv3.2 config.toml 骨架
这是整篇最该抄走的部分。采集接口的 Key 和 TaoToken 的 Key 分开写,模型 base_url 指向统一通道。
[collector] # 采集接口的鉴权 Key,从采集服务后台获取 api_key = "your_collector_api_key" # 采集接口基础地址 base_url = "https://api.example-collector.com" # 单次提交的最大 URL 数,避免一次塞太多 batch_size = 20 # 轮询快照状态的间隔(秒) poll_interval = 5 # 最大轮询次数,超时退出 max_poll = 60 [llm] # 统一走 TaoToken 通道 api_key = "your_taotoken_api_key" base_url = "https://taotoken.net/api" model = "deepseek-chat" temperature = 0.3 max_tokens = 4096 [output] # 采集结果落盘目录 data_dir = "./data" # 分析报告目录 report_dir = "./reports"3.3 采集模块:提交任务拿快照 ID
采集接口的通用模式是「提交任务 → 返回快照 ID → 轮询状态 → 下载数据集」。下面这段是提交逻辑,按商品 URL 采集,也支持关键词和店铺 URL,靠输入类型自动分流。
# modules/collect.py import time import requests import tomli def load_config(path="config.toml"): with open(path, "rb") as f: return tomli.load(f) def submit_job(cfg, inputs, input_type="url"): """提交采集任务,返回快照 ID""" url = f"{cfg['collector']['base_url']}/trigger" headers = {"Authorization": f"Bearer {cfg['collector']['api_key']}"} payload = {"input_type": input_type, "inputs": inputs} resp = requests.post(url, json=payload, headers=headers, timeout=30) resp.raise_for_status() return resp.json()["snapshot_id"] def wait_snapshot(cfg, snapshot_id): """轮询快照状态直到就绪""" url = f"{cfg['collector']['base_url']}/snapshot/{snapshot_id}" headers = {"Authorization": f"Bearer {cfg['collector']['api_key']}"} for _ in range(cfg["collector"]["max_poll"]): resp = requests.get(url, headers=headers, timeout=30) status = resp.json().get("status") if status == "ready": return True time.sleep(cfg["collector"]["poll_interval"]) raise TimeoutError(f"快照 {snapshot_id} 超时未就绪")3.4 下载模块:按快照 ID 拉数据
# modules/download.py import requests import csv import os def download_snapshot(cfg, snapshot_id, out_path): url = f"{cfg['collector']['base_url']}/snapshot/{snapshot_id}/download" headers = {"Authorization": f"Bearer {cfg['collector']['api_key']}"} resp = requests.get(url, headers=headers, timeout=60) resp.raise_for_status() rows = resp.json().get("data", []) os.makedirs(os.path.dirname(out_path), exist_ok=True) if not rows: raise ValueError("快照返回空数据集") with open(out_path, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=rows[0].keys()) writer.writeheader() writer.writerows(rows) return out_path3.5 分析模块:走 TaoToken 统一通道
这里用 openai 库,base_url 指向 TaoToken,Key 从 config 读。模型名按你实际开通的填。
# modules/analyze.py from openai import OpenAI import tomli def load_config(path="config.toml"): with open(path, "rb") as f: return tomli.load(f) def analyze_csv(cfg, csv_path, out_md): client = OpenAI( api_key=cfg["llm"]["api_key"], base_url=cfg["llm"]["base_url"], ) with open(csv_path, "r", encoding="utf-8") as f: content = f.read() prompt = f"""你是电商选品分析师。以下是 TikTok 公开商品数据 CSV: {content[:12000]} 请输出 Markdown 报告,包含:数据概览、品类分布、价格区间、爆款 TOP10、店铺排行、选品建议。""" resp = client.chat.completions.create( model=cfg["llm"]["model"], messages=[{"role": "user", "content": prompt}], temperature=cfg["llm"]["temperature"], max_tokens=cfg["llm"]["max_tokens"], ) report = resp.choices[0].message.content with open(out_md, "w", encoding="utf-8") as f: f.write(report) return out_md3.6 CLI 入口:一行命令跑通
用 argparse 把三个模块串起来,支持-i输入、-o输出、--analyze触发分析。
# tiktok_scraper.py import argparse import csv import os from modules.collect import load_config, submit_job, wait_snapshot from modules.download import download_snapshot from modules.analyze import analyze_csv def read_inputs(path): with open(path, "r", encoding="utf-8") as f: return [row[0] for row in csv.reader(f) if row] def main(): parser = argparse.ArgumentParser(description="TikTok 商品采集 CLI Agent") parser.add_argument("-i", "--input", required=True, help="输入 CSV,每行一个 URL 或关键词") parser.add_argument("-o", "--output", default="./data/output.csv", help="采集结果 CSV") parser.add_argument("--type", default="url", choices=["url", "keyword", "shop"], help="输入类型") parser.add_argument("--analyze", action="store_true", help="采集后生成分析报告") args = parser.parse_args() cfg = load_config() inputs = read_inputs(args.input) print(f"[1/3] 提交 {len(inputs)} 条采集任务,类型={args.type}") sid = submit_job(cfg, inputs, args.type) print(f"[2/3] 快照 ID={sid},等待就绪...") wait_snapshot(cfg, sid) out = download_snapshot(cfg, sid, args.output) print(f"[3/3] 数据已落盘:{out}") if args.analyze: md = os.path.join(cfg["output"]["report_dir"], "report.md") analyze_csv(cfg, out, md) print(f"分析报告:{md}") if __name__ == "__main__": main()启动命令就一条:
python3 tiktok_scraper.py -i input_sample.csv -o ./data/output.csv --type url --analyze4. 验证请求与成功结果
4.1 准备输入文件
input_sample.csv每行一个目标,第一行可以是表头也可以直接是数据,代码里按行读第一列。
https://www.tiktok.com/@shop/product/1234567890 https://www.tiktok.com/@shop/product/09876543214.2 先单独验证模型通道
在跑完整链路前,先确认 TaoToken 通道是通的,避免采集跑完才发现模型调不通。写个最小脚本:
from openai import OpenAI client = OpenAI(api_key="你的Key", base_url="https://taotoken.net/api") resp = client.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": "回复 OK 两个字母"}], ) print(resp.choices[0].message.content)返回OK就说明 Key 和 base_url 都对。这一步能省掉后面大量排查时间。
4.3 跑完整采集
python3 tiktok_scraper.py -i input_sample.csv -o ./data/output.csv --type url正常输出类似:
[1/3] 提交 2 条采集任务,类型=url [2/3] 快照 ID=snap_abc123,等待就绪... [3/3] 数据已落盘:./data/output.csv打开output.csv,字段通常包含商品名称、描述、价格、币种、销量、评分、店铺名、商品链接等。字段数量取决于采集接口返回,一般几十个。确认行数和输入条数对得上,就说明采集链路通了。
4.4 触发分析
python3 tiktok_scraper.py -i input_sample.csv -o ./data/output.csv --type url --analyze报告落在./reports/report.md,结构大致是数据概览、品类分布、价格区间、爆款 TOP10、店铺排行、选品建议。你可以改analyze.py里的 prompt 来定制维度,比如加上「物流时效关键词」「评论区高频词」这类你关心的字段。
5. 本篇常见错排查
5.1 快照一直不 ready
最常见原因是轮询次数不够或间隔太短。采集任务本身需要时间,max_poll=60、poll_interval=5意味着最多等 5 分钟。如果数据量大,把这两个值调大。另外确认input_type和输入内容匹配——用关键词采集却传了 URL,任务会一直卡住。
5.2 401 / 403 鉴权失败
分两种:采集接口的 Key 错了,或者 TaoToken 的 Key 错了。看报错发生在哪一步。提交任务阶段报错,查[collector].api_key;分析阶段报错,查[llm].api_key和base_url是否写成https://taotoken.net/api。注意 base_url 结尾不要多加斜杠,openai 库对路径拼接比较敏感。
5.3 CSV 字段错位或乱码
下载时用utf-8编码写入,如果源数据含特殊字符,读的时候也要指定编码。字段错位通常是DictWriter的 fieldnames 和实际数据 key 不一致,打印rows[0].keys()确认一下。空数据集会直接抛ValueError,这是有意为之,避免生成空报告。
5.4 模型返回被截断
分析长 CSV 时,prompt 太长会撞上max_tokens。我在analyze.py里做了content[:12000]截断,你可以按模型上下文调大。如果报告明显不完整,先调max_tokens,再考虑分批分析后合并。
5.5 采集频率与合规
采集公开数据要控制频率,batch_size别设太大,轮询间隔别太短。个人测试用极少量公开数据即可,遵守目标站点的 robots 与相关法规约定。CLI Agent 的便利不等于可以无节制抓取,这点心里要有数。
6. 把采集能力接进你的工作流
跑通之后,这个 CLI 的价值在于可组合。你可以把它接进定时任务,每天拉一次关键词商品数据;也可以把analyze.py换成走 Coding Plan 的通道,做更长的 Agent 任务。模型对话入口用来快速验证 prompt 效果: https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ,Key 管理在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite 。
如果你用 Claude Code 这类工具做开发,Anthropic 兼容入口在 https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claude_code&utm_campaign=rewrite ,可以把采集 Agent 的代码生成和维护也纳入统一通道。长期做编码和 Agent 任务的话,Coding Plan 页面值得看一眼: https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。
最后给个实用建议:把config.toml里的 Key 换成环境变量读取,比如api_key = "${TAOTOKEN_KEY}",在代码里做一次替换。这样配置文件可以放心提交,Key 留在本地环境。采集接口那边同理。CLI Agent 的工程化程度,往往就体现在这些配置细节上。