1. 项目概述淘宝作为国内最大的电商平台之一其商品数据对于市场分析、价格监控和竞品研究具有重要价值。通过Python调用淘宝店铺商品API获取全店商品数据可以快速建立商品数据库为后续的数据分析和商业决策提供支持。这个教程将带你从零开始完整实现通过Python获取淘宝全店商品的功能。不同于简单的网页爬虫我们使用的是淘宝开放平台提供的官方API接口这种方式更加稳定可靠且完全符合平台规则。2. 环境准备与API申请2.1 Python环境配置首先确保你的电脑上安装了Python 3.6或更高版本。推荐使用Anaconda来管理Python环境它可以方便地创建隔离的开发环境conda create -n taobao_api python3.8 conda activate taobao_api安装必要的依赖库pip install requests pandas pycryptodome注意pycryptodome库用于API签名加密这是淘宝API调用必需的步骤。2.2 淘宝开放平台账号申请访问淘宝开放平台官网(需自行搜索)注册开发者账号(个人或企业均可)创建应用选择网站应用类型获取App Key和App Secret申请过程中需要填写详细的应用信息包括回调地址等。对于本地测试可以使用http://127.0.0.1作为回调地址。2.3 获取API权限在应用管理后台找到接口管理部分申请taobao.items.onsale.get(获取出售中的商品列表)和taobao.items.inventory.get(获取仓库中的商品列表)这两个API的权限。3. API调用核心实现3.1 签名算法实现淘宝API要求所有请求都必须进行签名。以下是签名算法的Python实现import hashlib import hmac import urllib.parse from datetime import datetime def generate_sign(secret, params): # 1. 排序所有请求参数 sorted_params sorted(params.items(), keylambda x: x[0]) # 2. 拼接参数字符串 query_string for k, v in sorted_params: query_string f{k}{v} # 3. 使用HMAC-SHA256加密 sign hmac.new(secret.encode(utf-8), query_string.encode(utf-8), hashlib.sha256).hexdigest().upper() return sign3.2 基础请求函数创建一个通用的API请求函数处理签名、错误重试等逻辑import requests import time def call_taobao_api(method, app_key, app_secret, session_keyNone, **kwargs): base_url https://eco.taobao.com/router/rest # 公共参数 params { method: method, app_key: app_key, timestamp: datetime.now().strftime(%Y-%m-%d %H:%M:%S), format: json, v: 2.0, sign_method: hmac-sha256, } if session_key: params[session] session_key # 添加业务参数 params.update(kwargs) # 生成签名 params[sign] generate_sign(app_secret, params) # 发送请求 max_retry 3 for i in range(max_retry): try: response requests.get(base_url, paramsparams) result response.json() if error_response in result: error result[error_response] if error.get(code) 7: # 频率限制 time.sleep(2) # 等待2秒后重试 continue raise Exception(fAPI Error: {error.get(msg)}) return result except Exception as e: if i max_retry - 1: raise time.sleep(1)3.3 获取店铺商品列表现在我们可以实现获取全店商品的核心功能了。由于淘宝API有分页限制我们需要处理分页逻辑def get_all_shop_items(app_key, app_secret, session_key, shop_id): all_items [] page_no 1 page_size 100 # 每页最大100条 while True: # 获取在售商品 result call_taobao_api( taobao.items.onsale.get, app_key, app_secret, session_key, fieldsnum_iid,title,price,list_time,modified, page_nopage_no, page_sizepage_size ) items result.get(items_onsale_get_response, {}).get(items, {}).get(item, []) if not items: break all_items.extend(items) # 检查是否还有下一页 total_results result[items_onsale_get_response][total_results] if page_no * page_size total_results: break page_no 1 time.sleep(0.5) # 避免触发频率限制 return all_items4. 完整流程与授权4.1 获取Session Key要调用店铺相关的API需要先获取店铺的授权Session Key。这需要通过OAuth2.0授权流程引导用户访问授权URLdef get_auth_url(app_key, redirect_uri): return fhttps://oauth.taobao.com/authorize?response_typecodeclient_id{app_key}redirect_uri{redirect_uri}stateinit用户授权后使用返回的code换取Session Keydef get_session_key(app_key, app_secret, code): params { grant_type: authorization_code, client_id: app_key, client_secret: app_secret, code: code, redirect_uri: http://127.0.0.1 # 与申请时一致 } response requests.post(https://oauth.taobao.com/token, paramsparams) result response.json() if error in result: raise Exception(fAuth Error: {result[error_description]}) return result[access_token]4.2 完整示例代码将以上所有部分组合起来形成一个完整的示例def main(): # 配置你的应用信息 APP_KEY 你的AppKey APP_SECRET 你的AppSecret SHOP_ID 目标店铺ID # 1. 获取授权URL auth_url get_auth_url(APP_KEY, http://127.0.0.1) print(f请访问以下URL授权: {auth_url}) # 2. 用户授权后会跳转到回调地址URL中会包含code参数 code input(请输入回调URL中的code参数: ) # 3. 使用code换取Session Key session_key get_session_key(APP_KEY, APP_SECRET, code) print(f获取到的Session Key: {session_key}) # 4. 获取全店商品 print(开始获取店铺商品...) items get_all_shop_items(APP_KEY, APP_SECRET, session_key, SHOP_ID) # 5. 保存结果 import pandas as pd df pd.DataFrame(items) df.to_excel(shop_items.xlsx, indexFalse) print(f共获取到{len(items)}个商品已保存到shop_items.xlsx) if __name__ __main__: main()5. 常见问题与解决方案5.1 API调用频率限制淘宝API有严格的频率限制常见的错误代码和解决方法错误代码原因解决方案7调用频率超限降低请求频率增加sleep时间15远程服务错误稍后重试40缺少必要参数检查参数是否完整41参数非法检查参数格式和值建议在代码中加入适当的延迟避免频繁调用time.sleep(0.5) # 每次调用后暂停0.5秒5.2 数据不完整问题有时获取的商品列表不完整可能的原因分页处理不正确 - 确保正确处理total_results和page_no的关系商品状态变化 - 考虑同时获取在售和仓库中的商品API限制 - 某些字段需要额外权限改进版的商品获取函数def get_complete_shop_items(app_key, app_secret, session_key): # 获取在售商品 onsale_items get_all_shop_items(app_key, app_secret, session_key) # 获取仓库中的商品 inventory_items [] page_no 1 while True: result call_taobao_api( taobao.items.inventory.get, app_key, app_secret, session_key, fieldsnum_iid,title,price,list_time,modified, page_nopage_no, page_size100 ) items result.get(items_inventory_get_response, {}).get(items, {}).get(item, []) if not items: break inventory_items.extend(items) total_results result[items_inventory_get_response][total_results] if page_no * 100 total_results: break page_no 1 time.sleep(0.5) # 合并结果去重 all_items onsale_items inventory_items unique_items {item[num_iid]: item for item in all_items}.values() return list(unique_items)5.3 数据存储优化对于大量商品数据建议使用数据库存储而不是简单的Excel文件。以下是使用SQLite的示例import sqlite3 def save_to_db(items, db_filetaobao_items.db): conn sqlite3.connect(db_file) c conn.cursor() # 创建表 c.execute(CREATE TABLE IF NOT EXISTS items (num_iid TEXT PRIMARY KEY, title TEXT, price REAL, list_time TEXT, modified TEXT)) # 插入数据 for item in items: c.execute(INSERT OR REPLACE INTO items VALUES (:num_iid, :title, :price, :list_time, :modified), item) conn.commit() conn.close()6. 高级技巧与优化6.1 异步请求加速使用aiohttp库可以实现异步请求显著提高数据获取速度import aiohttp import asyncio async def async_call_taobao_api(session, method, app_key, app_secret, session_keyNone, **kwargs): base_url https://eco.taobao.com/router/rest params { method: method, app_key: app_key, timestamp: datetime.now().strftime(%Y-%m-%d %H:%M:%S), format: json, v: 2.0, sign_method: hmac-sha256, } if session_key: params[session] session_key params.update(kwargs) params[sign] generate_sign(app_secret, params) async with session.get(base_url, paramsparams) as response: result await response.json() if error_response in result: error result[error_response] raise Exception(fAPI Error: {error.get(msg)}) return result async def async_get_shop_items(app_key, app_secret, session_key): async with aiohttp.ClientSession() as session: tasks [] # 先获取总数量 result await async_call_taobao_api( session, taobao.items.onsale.get, app_key, app_secret, session_key, fieldsnum_iid, page_no1, page_size1 ) total result[items_onsale_get_response][total_results] pages (total // 100) 1 # 创建所有页面的任务 for page in range(1, pages 1): task async_call_taobao_api( session, taobao.items.onsale.get, app_key, app_secret, session_key, fieldsnum_iid,title,price,list_time,modified, page_nopage, page_size100 ) tasks.append(task) # 并行执行所有请求 results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理结果 items [] for r in results: if isinstance(r, Exception): print(f请求失败: {r}) continue items.extend(r.get(items_onsale_get_response, {}).get(items, {}).get(item, [])) return items6.2 数据更新策略对于定期更新的需求可以实现增量更新def incremental_update(db_filetaobao_items.db): conn sqlite3.connect(db_file) c conn.cursor() # 获取最后更新时间 c.execute(SELECT MAX(modified) FROM items) last_update c.fetchone()[0] # 获取更新的商品 params { fields: num_iid,title,price,list_time,modified, page_size: 100 } if last_update: params[start_modified] last_update updated_items [] page_no 1 while True: params[page_no] page_no result call_taobao_api( taobao.items.onsale.get, APP_KEY, APP_SECRET, SESSION_KEY, **params ) items result.get(items_onsale_get_response, {}).get(items, {}).get(item, []) if not items: break updated_items.extend(items) total result[items_onsale_get_response][total_results] if page_no * 100 total: break page_no 1 time.sleep(0.5) # 更新数据库 for item in updated_items: c.execute(INSERT OR REPLACE INTO items VALUES (:num_iid, :title, :price, :list_time, :modified), item) conn.commit() conn.close() print(f更新了{len(updated_items)}条商品记录)6.3 异常处理与监控在生产环境中完善的异常处理和监控是必不可少的import logging from logging.handlers import RotatingFileHandler # 配置日志 logging.basicConfig( handlers[RotatingFileHandler(taobao_api.log, maxBytes1e6, backupCount3)], levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) def monitored_get_items(): try: start time.time() items get_complete_shop_items(APP_KEY, APP_SECRET, SESSION_KEY) duration time.time() - start logging.info( f成功获取{len(items)}条商品数据耗时{duration:.2f}秒, extra{items_count: len(items), duration: duration} ) return items except Exception as e: logging.error(f获取商品数据失败: {str(e)}, exc_infoTrue) raise7. 实际应用案例7.1 价格监控系统利用获取的商品数据可以构建一个简单的价格监控系统def detect_price_changes(db_filetaobao_items.db): conn sqlite3.connect(db_file) # 获取当前价格 current_items pd.read_sql(SELECT num_iid, price FROM items, conn) # 获取历史价格 history pd.read_sql(SELECT * FROM price_history, conn) # 检测价格变化 merged pd.merge(current_items, history.groupby(num_iid)[price].last().reset_index(), onnum_iid, suffixes(_current, _previous)) changed merged[merged[price_current] ! merged[price_previous]] if not changed.empty: print(发现价格变动的商品:) print(changed[[num_iid, price_previous, price_current]]) # 记录价格变动 changed[record_time] datetime.now().strftime(%Y-%m-%d %H:%M:%S) changed[[num_iid, price_current, record_time]].rename( columns{price_current: price} ).to_sql(price_history, conn, if_existsappend, indexFalse) conn.close()7.2 商品数据分析使用pandas进行简单的数据分析def analyze_items(db_filetaobao_items.db): conn sqlite3.connect(db_file) df pd.read_sql(SELECT * FROM items, conn) conn.close() if df.empty: print(没有商品数据可供分析) return # 价格分布分析 print(\n价格分布统计:) print(df[price].describe()) # 价格区间统计 bins [0, 50, 100, 200, 500, 1000, float(inf)] labels [50, 50-100, 100-200, 200-500, 500-1000, 1000] df[price_range] pd.cut(df[price], binsbins, labelslabels) print(\n价格区间分布:) print(df[price_range].value_counts().sort_index()) # 上架时间分析 df[list_date] pd.to_datetime(df[list_time]).dt.date print(\n每日上架商品数量:) print(df[list_date].value_counts().sort_index().head(10))7.3 自动化报告生成结合Jinja2模板引擎可以生成精美的HTML报告from jinja2 import Environment, FileSystemLoader import webbrowser def generate_html_report(db_filetaobao_items.db): conn sqlite3.connect(db_file) df pd.read_sql(SELECT * FROM items, conn) conn.close() # 准备数据 price_stats df[price].describe().to_dict() price_dist df[price_range].value_counts().sort_index().to_dict() # 设置模板环境 env Environment(loaderFileSystemLoader(.)) template env.get_template(report_template.html) # 渲染模板 html template.render( total_itemslen(df), price_statsprice_stats, price_distprice_dist, update_timedatetime.now().strftime(%Y-%m-%d %H:%M:%S) ) # 保存并打开报告 with open(taobao_report.html, w, encodingutf-8) as f: f.write(html) webbrowser.open(taobao_report.html)提示report_template.html需要提前创建包含适当的HTML和CSS样式8. 性能优化与扩展8.1 缓存机制实现对于不常变动的数据可以添加缓存层减少API调用import pickle import os from functools import wraps def cache_response(cache_file, expire_hours24): def decorator(func): wraps(func) def wrapper(*args, **kwargs): if os.path.exists(cache_file): mtime os.path.getmtime(cache_file) if (time.time() - mtime) expire_hours * 3600: with open(cache_file, rb) as f: return pickle.load(f) result func(*args, **kwargs) with open(cache_file, wb) as f: pickle.dump(result, f) return result return wrapper return decorator # 使用示例 cache_response(shop_items_cache.pkl) def get_cached_shop_items(app_key, app_secret, session_key): return get_complete_shop_items(app_key, app_secret, session_key)8.2 分布式爬取架构对于大规模数据采集可以考虑分布式架构import redis from rq import Queue # 设置Redis任务队列 redis_conn redis.Redis() task_queue Queue(connectionredis_conn) def enqueue_shop_items_task(shop_id): task_queue.enqueue( get_complete_shop_items, APP_KEY, APP_SECRET, SESSION_KEY, shop_id, result_ttl86400 ) print(f已加入队列: 店铺{shop_id}商品采集任务)8.3 API调用监控使用Prometheus实现API调用监控from prometheus_client import start_http_server, Counter, Histogram # 定义指标 API_CALLS Counter(taobao_api_calls_total, Total API calls, [method, status]) API_DURATION Histogram(taobao_api_duration_seconds, API call duration, [method]) def monitored_call_taobao_api(method, app_key, app_secret, **kwargs): start time.time() try: result call_taobao_api(method, app_key, app_secret, **kwargs) API_CALLS.labels(methodmethod, statussuccess).inc() return result except Exception as e: API_CALLS.labels(methodmethod, statuserror).inc() raise finally: duration time.time() - start API_DURATION.labels(methodmethod).observe(duration) # 启动监控服务器 start_http_server(8000) Python调用淘宝API获取全店商品数据实战
1. 项目概述淘宝作为国内最大的电商平台之一其商品数据对于市场分析、价格监控和竞品研究具有重要价值。通过Python调用淘宝店铺商品API获取全店商品数据可以快速建立商品数据库为后续的数据分析和商业决策提供支持。这个教程将带你从零开始完整实现通过Python获取淘宝全店商品的功能。不同于简单的网页爬虫我们使用的是淘宝开放平台提供的官方API接口这种方式更加稳定可靠且完全符合平台规则。2. 环境准备与API申请2.1 Python环境配置首先确保你的电脑上安装了Python 3.6或更高版本。推荐使用Anaconda来管理Python环境它可以方便地创建隔离的开发环境conda create -n taobao_api python3.8 conda activate taobao_api安装必要的依赖库pip install requests pandas pycryptodome注意pycryptodome库用于API签名加密这是淘宝API调用必需的步骤。2.2 淘宝开放平台账号申请访问淘宝开放平台官网(需自行搜索)注册开发者账号(个人或企业均可)创建应用选择网站应用类型获取App Key和App Secret申请过程中需要填写详细的应用信息包括回调地址等。对于本地测试可以使用http://127.0.0.1作为回调地址。2.3 获取API权限在应用管理后台找到接口管理部分申请taobao.items.onsale.get(获取出售中的商品列表)和taobao.items.inventory.get(获取仓库中的商品列表)这两个API的权限。3. API调用核心实现3.1 签名算法实现淘宝API要求所有请求都必须进行签名。以下是签名算法的Python实现import hashlib import hmac import urllib.parse from datetime import datetime def generate_sign(secret, params): # 1. 排序所有请求参数 sorted_params sorted(params.items(), keylambda x: x[0]) # 2. 拼接参数字符串 query_string for k, v in sorted_params: query_string f{k}{v} # 3. 使用HMAC-SHA256加密 sign hmac.new(secret.encode(utf-8), query_string.encode(utf-8), hashlib.sha256).hexdigest().upper() return sign3.2 基础请求函数创建一个通用的API请求函数处理签名、错误重试等逻辑import requests import time def call_taobao_api(method, app_key, app_secret, session_keyNone, **kwargs): base_url https://eco.taobao.com/router/rest # 公共参数 params { method: method, app_key: app_key, timestamp: datetime.now().strftime(%Y-%m-%d %H:%M:%S), format: json, v: 2.0, sign_method: hmac-sha256, } if session_key: params[session] session_key # 添加业务参数 params.update(kwargs) # 生成签名 params[sign] generate_sign(app_secret, params) # 发送请求 max_retry 3 for i in range(max_retry): try: response requests.get(base_url, paramsparams) result response.json() if error_response in result: error result[error_response] if error.get(code) 7: # 频率限制 time.sleep(2) # 等待2秒后重试 continue raise Exception(fAPI Error: {error.get(msg)}) return result except Exception as e: if i max_retry - 1: raise time.sleep(1)3.3 获取店铺商品列表现在我们可以实现获取全店商品的核心功能了。由于淘宝API有分页限制我们需要处理分页逻辑def get_all_shop_items(app_key, app_secret, session_key, shop_id): all_items [] page_no 1 page_size 100 # 每页最大100条 while True: # 获取在售商品 result call_taobao_api( taobao.items.onsale.get, app_key, app_secret, session_key, fieldsnum_iid,title,price,list_time,modified, page_nopage_no, page_sizepage_size ) items result.get(items_onsale_get_response, {}).get(items, {}).get(item, []) if not items: break all_items.extend(items) # 检查是否还有下一页 total_results result[items_onsale_get_response][total_results] if page_no * page_size total_results: break page_no 1 time.sleep(0.5) # 避免触发频率限制 return all_items4. 完整流程与授权4.1 获取Session Key要调用店铺相关的API需要先获取店铺的授权Session Key。这需要通过OAuth2.0授权流程引导用户访问授权URLdef get_auth_url(app_key, redirect_uri): return fhttps://oauth.taobao.com/authorize?response_typecodeclient_id{app_key}redirect_uri{redirect_uri}stateinit用户授权后使用返回的code换取Session Keydef get_session_key(app_key, app_secret, code): params { grant_type: authorization_code, client_id: app_key, client_secret: app_secret, code: code, redirect_uri: http://127.0.0.1 # 与申请时一致 } response requests.post(https://oauth.taobao.com/token, paramsparams) result response.json() if error in result: raise Exception(fAuth Error: {result[error_description]}) return result[access_token]4.2 完整示例代码将以上所有部分组合起来形成一个完整的示例def main(): # 配置你的应用信息 APP_KEY 你的AppKey APP_SECRET 你的AppSecret SHOP_ID 目标店铺ID # 1. 获取授权URL auth_url get_auth_url(APP_KEY, http://127.0.0.1) print(f请访问以下URL授权: {auth_url}) # 2. 用户授权后会跳转到回调地址URL中会包含code参数 code input(请输入回调URL中的code参数: ) # 3. 使用code换取Session Key session_key get_session_key(APP_KEY, APP_SECRET, code) print(f获取到的Session Key: {session_key}) # 4. 获取全店商品 print(开始获取店铺商品...) items get_all_shop_items(APP_KEY, APP_SECRET, session_key, SHOP_ID) # 5. 保存结果 import pandas as pd df pd.DataFrame(items) df.to_excel(shop_items.xlsx, indexFalse) print(f共获取到{len(items)}个商品已保存到shop_items.xlsx) if __name__ __main__: main()5. 常见问题与解决方案5.1 API调用频率限制淘宝API有严格的频率限制常见的错误代码和解决方法错误代码原因解决方案7调用频率超限降低请求频率增加sleep时间15远程服务错误稍后重试40缺少必要参数检查参数是否完整41参数非法检查参数格式和值建议在代码中加入适当的延迟避免频繁调用time.sleep(0.5) # 每次调用后暂停0.5秒5.2 数据不完整问题有时获取的商品列表不完整可能的原因分页处理不正确 - 确保正确处理total_results和page_no的关系商品状态变化 - 考虑同时获取在售和仓库中的商品API限制 - 某些字段需要额外权限改进版的商品获取函数def get_complete_shop_items(app_key, app_secret, session_key): # 获取在售商品 onsale_items get_all_shop_items(app_key, app_secret, session_key) # 获取仓库中的商品 inventory_items [] page_no 1 while True: result call_taobao_api( taobao.items.inventory.get, app_key, app_secret, session_key, fieldsnum_iid,title,price,list_time,modified, page_nopage_no, page_size100 ) items result.get(items_inventory_get_response, {}).get(items, {}).get(item, []) if not items: break inventory_items.extend(items) total_results result[items_inventory_get_response][total_results] if page_no * 100 total_results: break page_no 1 time.sleep(0.5) # 合并结果去重 all_items onsale_items inventory_items unique_items {item[num_iid]: item for item in all_items}.values() return list(unique_items)5.3 数据存储优化对于大量商品数据建议使用数据库存储而不是简单的Excel文件。以下是使用SQLite的示例import sqlite3 def save_to_db(items, db_filetaobao_items.db): conn sqlite3.connect(db_file) c conn.cursor() # 创建表 c.execute(CREATE TABLE IF NOT EXISTS items (num_iid TEXT PRIMARY KEY, title TEXT, price REAL, list_time TEXT, modified TEXT)) # 插入数据 for item in items: c.execute(INSERT OR REPLACE INTO items VALUES (:num_iid, :title, :price, :list_time, :modified), item) conn.commit() conn.close()6. 高级技巧与优化6.1 异步请求加速使用aiohttp库可以实现异步请求显著提高数据获取速度import aiohttp import asyncio async def async_call_taobao_api(session, method, app_key, app_secret, session_keyNone, **kwargs): base_url https://eco.taobao.com/router/rest params { method: method, app_key: app_key, timestamp: datetime.now().strftime(%Y-%m-%d %H:%M:%S), format: json, v: 2.0, sign_method: hmac-sha256, } if session_key: params[session] session_key params.update(kwargs) params[sign] generate_sign(app_secret, params) async with session.get(base_url, paramsparams) as response: result await response.json() if error_response in result: error result[error_response] raise Exception(fAPI Error: {error.get(msg)}) return result async def async_get_shop_items(app_key, app_secret, session_key): async with aiohttp.ClientSession() as session: tasks [] # 先获取总数量 result await async_call_taobao_api( session, taobao.items.onsale.get, app_key, app_secret, session_key, fieldsnum_iid, page_no1, page_size1 ) total result[items_onsale_get_response][total_results] pages (total // 100) 1 # 创建所有页面的任务 for page in range(1, pages 1): task async_call_taobao_api( session, taobao.items.onsale.get, app_key, app_secret, session_key, fieldsnum_iid,title,price,list_time,modified, page_nopage, page_size100 ) tasks.append(task) # 并行执行所有请求 results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理结果 items [] for r in results: if isinstance(r, Exception): print(f请求失败: {r}) continue items.extend(r.get(items_onsale_get_response, {}).get(items, {}).get(item, [])) return items6.2 数据更新策略对于定期更新的需求可以实现增量更新def incremental_update(db_filetaobao_items.db): conn sqlite3.connect(db_file) c conn.cursor() # 获取最后更新时间 c.execute(SELECT MAX(modified) FROM items) last_update c.fetchone()[0] # 获取更新的商品 params { fields: num_iid,title,price,list_time,modified, page_size: 100 } if last_update: params[start_modified] last_update updated_items [] page_no 1 while True: params[page_no] page_no result call_taobao_api( taobao.items.onsale.get, APP_KEY, APP_SECRET, SESSION_KEY, **params ) items result.get(items_onsale_get_response, {}).get(items, {}).get(item, []) if not items: break updated_items.extend(items) total result[items_onsale_get_response][total_results] if page_no * 100 total: break page_no 1 time.sleep(0.5) # 更新数据库 for item in updated_items: c.execute(INSERT OR REPLACE INTO items VALUES (:num_iid, :title, :price, :list_time, :modified), item) conn.commit() conn.close() print(f更新了{len(updated_items)}条商品记录)6.3 异常处理与监控在生产环境中完善的异常处理和监控是必不可少的import logging from logging.handlers import RotatingFileHandler # 配置日志 logging.basicConfig( handlers[RotatingFileHandler(taobao_api.log, maxBytes1e6, backupCount3)], levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) def monitored_get_items(): try: start time.time() items get_complete_shop_items(APP_KEY, APP_SECRET, SESSION_KEY) duration time.time() - start logging.info( f成功获取{len(items)}条商品数据耗时{duration:.2f}秒, extra{items_count: len(items), duration: duration} ) return items except Exception as e: logging.error(f获取商品数据失败: {str(e)}, exc_infoTrue) raise7. 实际应用案例7.1 价格监控系统利用获取的商品数据可以构建一个简单的价格监控系统def detect_price_changes(db_filetaobao_items.db): conn sqlite3.connect(db_file) # 获取当前价格 current_items pd.read_sql(SELECT num_iid, price FROM items, conn) # 获取历史价格 history pd.read_sql(SELECT * FROM price_history, conn) # 检测价格变化 merged pd.merge(current_items, history.groupby(num_iid)[price].last().reset_index(), onnum_iid, suffixes(_current, _previous)) changed merged[merged[price_current] ! merged[price_previous]] if not changed.empty: print(发现价格变动的商品:) print(changed[[num_iid, price_previous, price_current]]) # 记录价格变动 changed[record_time] datetime.now().strftime(%Y-%m-%d %H:%M:%S) changed[[num_iid, price_current, record_time]].rename( columns{price_current: price} ).to_sql(price_history, conn, if_existsappend, indexFalse) conn.close()7.2 商品数据分析使用pandas进行简单的数据分析def analyze_items(db_filetaobao_items.db): conn sqlite3.connect(db_file) df pd.read_sql(SELECT * FROM items, conn) conn.close() if df.empty: print(没有商品数据可供分析) return # 价格分布分析 print(\n价格分布统计:) print(df[price].describe()) # 价格区间统计 bins [0, 50, 100, 200, 500, 1000, float(inf)] labels [50, 50-100, 100-200, 200-500, 500-1000, 1000] df[price_range] pd.cut(df[price], binsbins, labelslabels) print(\n价格区间分布:) print(df[price_range].value_counts().sort_index()) # 上架时间分析 df[list_date] pd.to_datetime(df[list_time]).dt.date print(\n每日上架商品数量:) print(df[list_date].value_counts().sort_index().head(10))7.3 自动化报告生成结合Jinja2模板引擎可以生成精美的HTML报告from jinja2 import Environment, FileSystemLoader import webbrowser def generate_html_report(db_filetaobao_items.db): conn sqlite3.connect(db_file) df pd.read_sql(SELECT * FROM items, conn) conn.close() # 准备数据 price_stats df[price].describe().to_dict() price_dist df[price_range].value_counts().sort_index().to_dict() # 设置模板环境 env Environment(loaderFileSystemLoader(.)) template env.get_template(report_template.html) # 渲染模板 html template.render( total_itemslen(df), price_statsprice_stats, price_distprice_dist, update_timedatetime.now().strftime(%Y-%m-%d %H:%M:%S) ) # 保存并打开报告 with open(taobao_report.html, w, encodingutf-8) as f: f.write(html) webbrowser.open(taobao_report.html)提示report_template.html需要提前创建包含适当的HTML和CSS样式8. 性能优化与扩展8.1 缓存机制实现对于不常变动的数据可以添加缓存层减少API调用import pickle import os from functools import wraps def cache_response(cache_file, expire_hours24): def decorator(func): wraps(func) def wrapper(*args, **kwargs): if os.path.exists(cache_file): mtime os.path.getmtime(cache_file) if (time.time() - mtime) expire_hours * 3600: with open(cache_file, rb) as f: return pickle.load(f) result func(*args, **kwargs) with open(cache_file, wb) as f: pickle.dump(result, f) return result return wrapper return decorator # 使用示例 cache_response(shop_items_cache.pkl) def get_cached_shop_items(app_key, app_secret, session_key): return get_complete_shop_items(app_key, app_secret, session_key)8.2 分布式爬取架构对于大规模数据采集可以考虑分布式架构import redis from rq import Queue # 设置Redis任务队列 redis_conn redis.Redis() task_queue Queue(connectionredis_conn) def enqueue_shop_items_task(shop_id): task_queue.enqueue( get_complete_shop_items, APP_KEY, APP_SECRET, SESSION_KEY, shop_id, result_ttl86400 ) print(f已加入队列: 店铺{shop_id}商品采集任务)8.3 API调用监控使用Prometheus实现API调用监控from prometheus_client import start_http_server, Counter, Histogram # 定义指标 API_CALLS Counter(taobao_api_calls_total, Total API calls, [method, status]) API_DURATION Histogram(taobao_api_duration_seconds, API call duration, [method]) def monitored_call_taobao_api(method, app_key, app_secret, **kwargs): start time.time() try: result call_taobao_api(method, app_key, app_secret, **kwargs) API_CALLS.labels(methodmethod, statussuccess).inc() return result except Exception as e: API_CALLS.labels(methodmethod, statuserror).inc() raise finally: duration time.time() - start API_DURATION.labels(methodmethod).observe(duration) # 启动监控服务器 start_http_server(8000)