作为电商数据开发中最常用的接口之一淘宝商品评论 API 是实现商品口碑分析、竞品监控、用户画像构建的核心能力。本文基于Python完整实现淘宝商品评论 API 的调用、数据解析、结构化存储全流程代码可直接复用配套详细注释和避坑指南适合电商数据开发者、爬虫工程师、数据分析从业者参考。一、淘宝商品评论 API 核心使用场景在实际开发中评论 API 是电商数据体系的基础模块主要落地于以下场景1. 商品口碑与质量分析批量抓取评论数据通过 NLP 提取差评关键词如 “续航差”“做工粗糙”、统计评分分布实现商品质量异常预警替代人工逐条审核的低效模式。2. 电商选品决策为店群运营 / 供应链平台提供数据支撑筛选好评率95%、复购反馈多的潜力商品规避差评集中的滞销款。3. 竞品监控定时抓取竞品评论实时识别负面舆情如 “售后差”“假货”并推送告警通知快速响应市场变化。4. 内容运营与推荐提取优质好评、晒单内容作为导购平台的素材结合用户评价标签实现个性化商品推荐提升转化。5. AI 数据集构建构建商品评论情感分析数据集用于训练 NLP 模型如文本分类、观点抽取支撑智能客服、智能质检等场景。二、API 调用前置知识Python 版1. 接口通用规范以主流第三方合规接口为例项⽬说明请求方式POST/GET推荐 POST避免参数暴露数据格式JSON核心请求参数appkey应用密钥、item_id商品 ID、page页码、page_size每页条数、sign签名、timestamp时间戳核心返回字段user_name用户名、score评分、content评论内容、create_time评论时间、reply商家回复、total_count总评论数鉴权方式appkey appsecret 生成签名MD5/SHA2562. 环境准备首先安装必备依赖执行以下命令pip install requests python-dotenv pymysql apscheduler logururequests发送 HTTP 请求python-dotenv管理配置文件避免硬编码密钥pymysql存储数据到 MySQLapscheduler实现定时抓取loguru简化日志记录三、完整 Python 实战代码1. 配置文件.env新建.env文件存放敏感配置避免硬编码泄露# API 配置 APP_KEY你的appkey APP_SECRET你的appsecret API_URLhttps://api.example.com/taobao/comment # 替换为实际API地址 # MySQL 配置 MYSQL_HOST127.0.0.1 MYSQL_PORT3306 MYSQL_USERroot MYSQL_PASSWORD你的密码 MYSQL_DBtaobao_comment # 抓取配置 PAGE_SIZE20 MAX_RETRY3 # 最大重试次数2. 核心实现代码taobao_comment_api.pyimport requests import time import hashlib import json import os import pymysql from dotenv import load_dotenv from loguru import logger from apscheduler.schedulers.blocking import BlockingScheduler from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry # 加载配置文件 load_dotenv() # 初始化日志 logger.add( taobao_comment.log, rotation500MB, retention7 days, encodingutf-8, levelINFO ) class TaobaoCommentAPI: def __init__(self): 初始化配置和请求会话 # API 配置 self.app_key os.getenv(APP_KEY) self.app_secret os.getenv(APP_SECRET) self.api_url os.getenv(API_URL) self.page_size int(os.getenv(PAGE_SIZE)) self.max_retry int(os.getenv(MAX_RETRY)) # MySQL 配置 self.mysql_config { host: os.getenv(MYSQL_HOST), port: int(os.getenv(MYSQL_PORT)), user: os.getenv(MYSQL_USER), password: os.getenv(MYSQL_PASSWORD), db: os.getenv(MYSQL_DB), charset: utf8mb4 # 支持emoji } # 初始化请求会话带重试机制 self.session self._init_session() # 初始化数据库连接 self.db_conn self._init_mysql() def _init_session(self): 初始化请求会话配置超时和重试 session requests.Session() # 重试策略连接失败、超时、5xx错误重试 retry_strategy Retry( totalself.max_retry, backoff_factor1, # 重试间隔1s, 2s, 4s... status_forcelist[429, 500, 502, 503, 504], allowed_methods[GET, POST] ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(https://, adapter) session.mount(http://, adapter) # 设置超时时间 session.timeout 10 return session def _init_mysql(self): 初始化MySQL连接创建评论表如果不存在 try: conn pymysql.connect(**self.mysql_config) # 创建表 create_table_sql CREATE TABLE IF NOT EXISTS tb_comment ( id INT AUTO_INCREMENT PRIMARY KEY COMMENT 自增ID, item_id VARCHAR(64) NOT NULL COMMENT 商品ID, user_name VARCHAR(64) COMMENT 用户名, score TINYINT COMMENT 评分(1-5), content TEXT COMMENT 评论内容, create_time DATETIME COMMENT 评论时间, reply_content TEXT COMMENT 商家回复, has_image TINYINT DEFAULT 0 COMMENT 是否有图片(0-无,1-有), crawl_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT 抓取时间, UNIQUE KEY uk_item_user_time (item_id, user_name, create_time) # 防止重复存储 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT淘宝商品评论表; with conn.cursor() as cursor: cursor.execute(create_table_sql) conn.commit() logger.info(MySQL连接成功评论表初始化完成) return conn except Exception as e: logger.error(fMySQL初始化失败{str(e)}) raise def _generate_sign(self, params): 生成API签名MD5方式按接口要求调整 # 1. 按key排序参数 sorted_params sorted(params.items(), keylambda x: x[0]) # 2. 拼接参数字符串 sign_str for k, v in sorted_params: if v: # 跳过空值 sign_str f{k}{v} # 3. 拼接app_secret sign_str self.app_secret sign_str self.app_secret # 4. MD5加密并转大写 sign hashlib.md5(sign_str.encode(utf-8)).hexdigest().upper() return sign def get_comments(self, item_id, page1): 调用API获取单页评论 :param item_id: 商品ID :param page: 页码 :return: 结构化的评论列表 try: # 1. 构造基础参数 timestamp str(int(time.time())) params { appkey: self.app_key, item_id: item_id, page: page, page_size: self.page_size, timestamp: timestamp } # 2. 生成签名 params[sign] self._generate_sign(params) # 3. 发送请求 response self.session.post(self.api_url, jsonparams) response.raise_for_status() # 抛出HTTP错误 result response.json() # 4. 校验返回结果 if result.get(code) ! 200: logger.error(fAPI调用失败{result.get(msg)}商品ID{item_id}页码{page}) return [] # 5. 解析评论数据 comments result.get(data, {}).get(comments, []) parsed_comments [] for comment in comments: parsed_comment { item_id: item_id, user_name: comment.get(user_name, ), score: comment.get(score, 0), content: comment.get(content, ).strip(), create_time: comment.get(create_time, ), reply_content: comment.get(reply, {}).get(content, ).strip(), has_image: 1 if comment.get(images) else 0 } parsed_comments.append(parsed_comment) logger.info(f成功抓取商品{item_id}第{page}页评论共{len(parsed_comments)}条) return parsed_comments except requests.exceptions.RequestException as e: logger.error(f请求异常{str(e)}商品ID{item_id}页码{page}) return [] except Exception as e: logger.error(f解析异常{str(e)}商品ID{item_id}页码{page}) return [] def save_comments(self, comments): 保存评论到MySQL :param comments: 解析后的评论列表 if not comments: return try: insert_sql INSERT INTO tb_comment (item_id, user_name, score, content, create_time, reply_content, has_image) VALUES (%s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE scoreVALUES(score), contentVALUES(content), reply_contentVALUES(reply_content); with self.db_conn.cursor() as cursor: # 批量插入 cursor.executemany( insert_sql, [( c[item_id], c[user_name], c[score], c[content], c[create_time], c[reply_content], c[has_image] ) for c in comments] ) self.db_conn.commit() logger.info(f成功保存{len(comments)}条评论到MySQL) except Exception as e: self.db_conn.rollback() logger.error(f保存评论失败{str(e)}) def crawl_item_comments(self, item_id, max_page5): 抓取单个商品多页评论 :param item_id: 商品ID :param max_page: 最大抓取页数 logger.info(f开始抓取商品{item_id}评论最大页数{max_page}) for page in range(1, max_page 1): comments self.get_comments(item_id, page) if not comments: logger.warning(f商品{item_id}第{page}页无评论停止抓取) break self.save_comments(comments) # 限速避免触发接口限流 time.sleep(1) def start_scheduled_crawl(self, item_ids, max_page5, cron0 0 * * *): 启动定时抓取每天0点执行 :param item_ids: 商品ID列表 :param max_page: 最大抓取页数 :param cron: cron表达式 scheduler BlockingScheduler(timezoneAsia/Shanghai) # 添加定时任务 scheduler.add_job( funcself._scheduled_task, triggercron, args[item_ids, max_page], cron_expressioncron, name淘宝评论定时抓取 ) logger.info(定时抓取任务已启动Cron表达式{}监控商品{}, cron, item_ids) try: scheduler.start() except KeyboardInterrupt: logger.info(定时任务被手动终止) scheduler.shutdown() def _scheduled_task(self, item_ids, max_page): 定时任务执行函数 for item_id in item_ids: self.crawl_item_comments(item_id, max_page) def close(self): 关闭数据库连接和会话 if self.db_conn: self.db_conn.close() self.session.close() logger.info(资源已释放数据库连接、请求会话) # 测试代码 if __name__ __main__: # 初始化API客户端 comment_api TaobaoCommentAPI() try: # 1. 单次抓取指定商品评论示例商品ID替换为实际值 target_item_id 123456789 comment_api.crawl_item_comments(target_item_id, max_page3) # 2. 启动定时抓取注释掉单次抓取打开下面的代码即可 # monitor_item_ids [123456789, 987654321] # comment_api.start_scheduled_crawl(monitor_item_ids, max_page5) finally: # 确保资源释放 comment_api.close()3. 代码运行效果说明执行代码后你会看到以下效果日志输出taobao_comment.log文件中会记录抓取进度、成功 / 失败信息数据库存储MySQL 的taobao_comment表中会新增商品评论数据重复评论会自动更新单次抓取快速获取指定商品 3 页评论可调整max_page定时抓取注释掉单次抓取代码打开定时任务后每天 0 点自动抓取指定商品评论。4. 常见问题与避坑指南问题现象原因分析解决方案签名验证失败签名生成规则与接口要求不一致核对接口文档调整_generate_sign函数接口返回 429请求频率过高触发限流增加time.sleep()间隔降低抓取速度评论内容乱码编码未设置为 utf8mb4MySQL 表字符集必须为 utf8mb4支持 emoji数据库连接失败配置错误 / 服务未启动检查.env文件中的 MySQL 配置启动 MySQL接口返回空数据商品 ID 错误 / 无评论 / 接口权限不足核对商品 ID检查接口权限四、开发者核心注意事项CSDN 重点提醒1. 合规性重中之重仅使用合规授权的 API 接口禁止爬取淘宝官方未开放的接口评论数据仅限内部分析 / 商用授权场景禁止泄露用户隐私、用于非法用途遵守《电子商务法》《网络数据安全管理条例》等法律法规。2. 稳定性优化加入熔断降级当 API 连续失败时暂停抓取避免无效请求增加缓存对高频访问的商品评论做本地缓存减少 API 调用监控告警对接钉钉 / 企业微信机器人异常时推送告警信息。3. CSDN 读者专属优化建议多语言适配如需 Java/PHP 版本可参考本文的签名、解析逻辑替换 HTTP 客户端即可数据可视化结合pandas matplotlib实现评论评分分布、关键词云图分布式部署如需大规模抓取可基于Celery Redis实现分布式任务调度。五、总结本文基于 Python 实现了淘宝商品评论 API 的完整调用链路核心要点如下通过配置文件管理敏感信息符合开发最佳实践封装通用 API 调用类支持单次抓取、定时抓取两种模式完善的异常处理、重试机制保证抓取稳定性结构化存储到 MySQL支持重复数据自动更新附带避坑指南解决开发者常见问题。