1. 电商评论监控的真实痛点：手动翻页翻到怀疑人生
做电商运营或者竞品调研的朋友，大概率都经历过这种场景：每天早上打开后台，把自家店铺和几个竞品的商品评价一条条翻过去，想看看有没有新的差评、用户最近在吐槽什么。商品少的时候还能忍，一旦要盯十几款甚至几十款商品，光翻页复制就能耗掉大半天。更麻烦的是，等你人工整理完，负面评价可能已经挂了两三天，客服和产品团队完全来不及反应。
这个问题的本质不是"要不要监控"，而是"怎么把采集、清洗、分析、预警串成一条自动化的链路"。手动做，效率低还容易漏；自己从零写爬虫，又要处理分页、去重、反爬、文本清洗一大堆底层逻辑，开发周期长，维护起来也头疼。我试过用纯 requests 硬怼页面接口，结果频繁触发访问限制，数据漏采、报错中断是常态。
这篇教程要解决的，就是把这套链路用 Python 脚本 + 统一 Key 的方式落地下来。核心思路是：用 TaoToken 统一管理模型调用的 Key，把评论采集后的文本分析（情感判断、高频词提取、差评归因）交给大模型来处理，脚本本身只负责采集、存储和调度。这样你不需要自己维护一套复杂的分词和情感词典，也能拿到结构化的分析结论。适合个人卖家做竞品调研，也适合运营团队做全店铺口碑监控。
整篇会交付一份可复制的config.toml骨架、统一 Key 的配置示例，以及脚本跑起来之后的验证动作。跟着做，你能搭出一条"定时拉取评论 → 增量去重 → 文本分析 → 负面预警"的实时追踪链路。
2. 前置准备：TaoToken 统一 Key 与 Python 环境
2.1 为什么用统一 Key 而不是每个模型单独配
做文本分析时，你可能会用到不同能力的模型：有的擅长情感分类，有的擅长关键词抽取，有的适合做长文本摘要。如果每个模型都单独申请 Key、单独配环境变量，脚本里就会散落一堆凭证，换模型时改起来很烦。TaoToken 的做法是提供一个统一的 API 入口，你用同一个 Key 就能调用不同的模型，脚本里只需要维护一份配置。
对电商评论监控这个场景来说，好处很直接：采集模块和分析模块解耦，分析模块换模型时，采集脚本一行都不用动。你只需要在config.toml里改一下模型名，重启脚本就生效。
2.2 获取 Key 与依赖安装
先到 TaoToken 控制台创建一个 API Key，建议单独建一个用于脚本调用的 Key，方便后续做权限隔离和用量统计。拿到 Key 之后，本地安装依赖：
pip install requests pandas jieba tomlitomli是用来解析config.toml的，Python 3.11 以上版本内置了tomllib，可以不用装。jieba用于中文分词，pandas用于数据持久化和统计。
2.3 目录结构约定
为了让脚本好维护，建议按下面的结构组织文件：
comment_monitor/ ├── config.toml # 统一配置：Key、模型、商品ID、轮询间隔 ├── goods_id.txt # 待监控商品ID清单，一行一个 ├── main.py # 主程序：采集 + 分析 + 调度 ├── data/ │ └── comments.csv # 原始评论持久化 └── output/ └── analysis.csv # 分析结果输出把配置和代码分开，后面调轮询频率、换模型、加商品，都只动config.toml和goods_id.txt，不用碰主逻辑。
3. 可复制的 config.toml 骨架与统一 Key 配置
3.1 config.toml 完整骨架
下面这份配置可以直接复制，把api_key换成你自己的即可。模型名按你实际开通的能力填，这里用占位符表示。
[taotoken] # 统一 Key，所有模型调用共用这一份 api_key = "sk-你的TaoToken密钥" # API 基础地址，不要加多余路径 base_url = "https://taotoken.net/api" # 文本分析使用的模型，按需替换 model = "你的模型名" # 单次请求超时，秒 timeout = 30 # 失败重试次数 max_retries = 3 [monitor] # 商品ID清单文件 goods_file = "goods_id.txt" # 轮询间隔，秒。竞品重点监控可设 300，常规监控设 3600 loop_interval = 3600 # 单商品单页拉取条数 page_size = 50 # 每页之间的间隔，避免请求过密 page_sleep = 1.0 [storage] # 原始评论存储路径 comment_csv = "data/comments.csv" # 分析结果输出路径 analysis_csv = "output/analysis.csv" [alert] # 是否开启负面预警 enabled = true # 评分小于等于该值视为差评 bad_score_threshold = 2 # 预警推送地址，企业微信或钉钉机器人 webhook webhook_url = ""3.2 统一 Key 的读取方式
主程序里读取配置，把 Key 注入到请求头。注意不要把 Key 硬编码在代码里，也不要提交到公开仓库。
import tomli def load_config(path="config.toml"): with open(path, "rb") as f: return tomli.load(f) CFG = load_config() API_KEY = CFG["taotoken"]["api_key"] BASE_URL = CFG["taotoken"]["base_url"] MODEL = CFG["taotoken"]["model"]3.3 商品 ID 清单
goods_id.txt一行一个商品 ID，脚本启动时读取，支持批量监控：
723456123456 723456123789 723456123999这样加竞品只需要往文件里追加一行，不用改代码。
4. 采集 + 文本分析脚本完整实现
4.1 采集模块：增量拉取与去重
采集部分的核心是"只拉新增"。脚本启动时先读取本地 CSV 里已有的评论 ID，存进一个集合，每次拉取时过滤掉已存在的，避免重复存储。
import time import requests import pandas as pd def load_history_ids(csv_path): try: df = pd.read_csv(csv_path, encoding="utf-8-sig") return set(df["comment_id"].astype(str).tolist()) except FileNotFoundError: return set() def fetch_comments(item_id, page, cfg): url = f"{cfg['taotoken']['base_url']}/comment/query" headers = {"Authorization": f"Bearer {cfg['taotoken']['api_key']}"} params = { "item_id": item_id, "page": page, "page_size": cfg["monitor"]["page_size"], } resp = requests.get( url, headers=headers, params=params, timeout=cfg["taotoken"]["timeout"] ) resp.raise_for_status() return resp.json() def collect_item(item_id, history_ids, cfg): new_comments = [] page = 1 while True: data = fetch_comments(item_id, page, cfg) items = data.get("data", {}).get("list", []) if not items: break for it in items: cid = str(it.get("comment_id")) if cid not in history_ids: history_ids.add(cid) new_comments.append(it) page += 1 time.sleep(cfg["monitor"]["page_sleep"]) return new_comments这里把history_ids作为可变集合传入，采集过程中实时更新，避免同一轮里重复。
4.2 文本分析模块：调用统一 Key 做情感与关键词
采集到的评论文本，直接丢给模型做结构化分析。这里的关键是设计好 prompt，让模型返回 JSON，方便脚本解析。
import json ANALYSIS_PROMPT = """你是电商评论分析助手。请对下面的评论做分析，只返回 JSON，不要输出其他内容。 JSON 格式： { "sentiment": "positive/neutral/negative", "keywords": ["关键词1", "关键词2"], "reason": "一句话说明判断依据" } 评论内容： {content} """ def analyze_comment(content, cfg): url = f"{cfg['taotoken']['base_url']}/chat/completions" headers = { "Authorization": f"Bearer {cfg['taotoken']['api_key']}", "Content-Type": "application/json", } payload = { "model": cfg["taotoken"]["model"], "messages": [ {"role": "user", "content": ANALYSIS_PROMPT.format(content=content)} ], "temperature": 0.2, } for attempt in range(cfg["taotoken"]["max_retries"]): try: resp = requests.post( url, headers=headers, json=payload, timeout=cfg["taotoken"]["timeout"] ) resp.raise_for_status() text = resp.json()["choices"][0]["message"]["content"] return json.loads(text) except Exception as e: if attempt == cfg["taotoken"]["max_retries"] - 1: return {"sentiment": "unknown", "keywords": [], "reason": str(e)} time.sleep(2)注意temperature设低一点，让输出更稳定，方便解析。
4.3 持久化与预警
采集和分析的结果分别落盘，差评触发预警：
def save_comments(comments, cfg): if not comments: return df_new = pd.DataFrame(comments) path = cfg["storage"]["comment_csv"] try: df_old = pd.read_csv(path, encoding="utf-8-sig") df_total = pd.concat([df_old, df_new], ignore_index=True) except FileNotFoundError: df_total = df_new df_total.drop_duplicates(subset=["comment_id"], keep="last", inplace=True) df_total.to_csv(path, index=False, encoding="utf-8-sig") def send_alert(text, cfg): if not cfg["alert"]["enabled"] or not cfg["alert"]["webhook_url"]: return requests.post( cfg["alert"]["webhook_url"], json={"msgtype": "text", "text": {"content": text}}, timeout=10, )4.4 主循环
把上面几块串起来，定时轮询：
def main(): cfg = load_config() history_ids = load_history_ids(cfg["storage"]["comment_csv"]) with open(cfg["monitor"]["goods_file"], encoding="utf-8") as f: goods = [line.strip() for line in f if line.strip()] while True: all_new = [] for gid in goods: new = collect_item(gid, history_ids, cfg) all_new.extend(new) if all_new: save_comments(all_new, cfg) bad = [c for c in all_new if c.get("score", 5) <= cfg["alert"]["bad_score_threshold"]] if bad: send_alert(f"检测到 {len(bad)} 条新增差评，请及时查看", cfg) else: print("本轮无新增评论") time.sleep(cfg["monitor"]["loop_interval"]) if __name__ == "__main__": main()5. 验证请求与成功结果
5.1 先做一次单次调用验证
在跑主循环之前，建议先单独验证 Key 和模型是否通。写一个最小测试：
import requests resp = requests.post( "https://taotoken.net/api/chat/completions", headers={"Authorization": "Bearer sk-你的密钥"}, json={ "model": "你的模型名", "messages": [{"role": "user", "content": "回复 OK 两个字母"}], }, timeout=30, ) print(resp.status_code) print(resp.json())如果返回 200 且内容里有OK，说明 Key 和模型都正常。如果返回 401，检查 Key 是否复制完整；返回 404，检查base_url是否写成了带多余路径的地址。
5.2 跑一次采集验证
把loop_interval临时改成 60，goods_id.txt里只放一个商品，运行main.py。观察控制台输出：
开始采集商品：723456123456 新增 37 条评论 检测到 2 条新增差评，请及时查看然后检查data/comments.csv，应该能看到评论正文、评分、时间等字段；output/analysis.csv里能看到情感标签和关键词。如果 CSV 用 Excel 打开中文乱码，确认写入时用的是utf-8-sig编码。
5.3 验证增量逻辑
第一次跑完之后，不要删数据，直接再跑一次。如果商品没有新评论，控制台应该输出"本轮无新增评论"，CSV 行数不变。这一步是验证去重是否生效的关键，很多脚本的 bug 就出在这里。
6. 本篇常见错误排查
6.1 请求返回 401 或 403
最常见的原因是 Key 没带对。检查config.toml里api_key是否有多余空格，请求头里是不是写成了Bearer sk-xxx的格式。另外确认这个 Key 有没有被禁用或者额度耗尽。
6.2 模型返回的内容不是合法 JSON
大模型偶尔会在 JSON 外面包一层说明文字，导致json.loads失败。解决办法有两个：一是 prompt 里强调"只返回 JSON，不要输出其他内容"；二是在解析前做一次清洗，把第一个{到最后一个}之间的内容截出来再解析。
def safe_parse(text): start = text.find("{") end = text.rfind("}") if start == -1 or end == -1: return None return json.loads(text[start:end + 1])6.3 采集速度慢或者被限流
把page_sleep调大一点，比如从 1.0 改成 2.0，给服务端留出喘息时间。同时确认page_size没有设得过大，一般 50 比较稳妥。如果商品评论量特别大，可以把轮询间隔拉长，避免同一时间发起太多请求。
6.4 分词结果里全是无意义词
如果你在本地也用了 jieba 做辅助分词，记得维护停用词表，把"的""了""就是""感觉"这类词过滤掉。不过更推荐的做法是直接用模型做关键词抽取，省去维护词典的麻烦，效果也更贴近语义。
6.5 预警没有推送
先检查webhook_url是否填了、机器人是否开启了消息接收。企业微信和钉钉的机器人对消息格式有要求，msgtype和text.content字段不能少。如果还是收不到，把send_alert里的请求单独拿出来测一下，看返回的 errcode 是什么。
7. 把链路跑起来之后，你可以这样扩展
这套脚本跑通之后，最直接的扩展方向是分析维度的细化。比如把差评按关键词聚类，自动归因到"物流""材质""尺寸""客服"几个大类，运营每天早上看一份归类报表就够了。再进一步，可以把历史评论按周做趋势对比，看某个吐槽点是在上升还是在收敛。
另一个方向是接入更多数据源。现在只监控了商品评论，其实问答区、追评、晒图描述里也有大量用户反馈，采集逻辑稍作调整就能覆盖。分析模块因为走的是统一 Key，换模型或者加模型都不影响采集侧。
最后提醒一句：脚本长期跑的时候，记得定期清理comments.csv，或者按月份分文件存储。数据量大了之后，单文件读写会变慢，分片存储能让分析模块跑得更轻快。
