如何使用Python结合亚马逊商品详情API实现商品价格监控？

你想通过 Python 结合亚马逊商品详情 API 实现商品价格监控，核心是自动定时获取商品价格、对比预设阈值、价格达标时触发提醒，我会基于亚马逊官方 PA API 5.0（最合规的方式），提供一套「新手可落地、功能完整、稳定性高」的监控方案，涵盖从 API 配置到自动提醒的全流程。

一、核心前提（必看）

1. API 基础：使用亚马逊 Product Advertising API (PA API 5.0)，需先在亚马逊联盟后台申请Access Key、Secret Key、Partner Tag（跟踪 ID），商品用唯一标识 ASIN（从商品详情页 URL 提取，如https://www.amazon.com/dp/B08CZJ7Z8L中的B08CZJ7Z8L）；
2. 监控逻辑：定时调用 API → 解析价格数据 → 存储历史价格 → 对比阈值 → 触发提醒（控制台 / 邮件）→ 循环执行；
3. 必备库：amazon-pa-api5-python（简化 API 签名）、requests、pandas（数据存储）、schedule（定时任务）、smtplib（邮件提醒）。

二、前置准备

1. 安装依赖库

bash

运行

# 核心：亚马逊API签名+请求 pip install amazon-pa-api5-python # 辅助：数据存储+定时任务 pip install pandas schedule

2. 配置关键信息

• 亚马逊 PA API 5.0 的Access Key/Secret Key/Partner Tag（亚马逊联盟后台获取）；
• 目标商品 ASIN、价格阈值（如 AirPods Pro 低于 200 美元提醒）；
• （可选）邮箱账号（开启 SMTP 服务，如 QQ 邮箱需获取授权码）。

三、完整实现代码（可直接运行）

python

运行

# 亚马逊API依赖 from paapi5_python_sdk import ApiClient, ApiException, Configuration, GetItemsApi from paapi5_python_sdk.models import GetItemsRequest, PartnerType, Resource # 基础工具 import pandas as pd from datetime import datetime import schedule import time # 邮件提醒（可选） import smtplib from email.mime.text import MIMEText from email.header import Header # ===================== 全局配置（修改为你的信息） ===================== # 亚马逊API配置 AMAZON_CONFIG = Configuration() AMAZON_CONFIG.access_key = "你的Access Key" AMAZON_CONFIG.secret_key = "你的Secret Key" AMAZON_CONFIG.host = "webservices.amazon.com" # 美国站，中国站改webservices.amazon.cn PARTNER_TAG = "你的Partner Tag（跟踪ID）" # 监控配置 TARGET_ASIN = "B08CZJ7Z8L" # 监控商品ASIN PRICE_THRESHOLD = 200.0 # 价格阈值（低于此价触发提醒） MONITOR_INTERVAL = 1 # 监控频率（小时） HISTORY_FILE = "amazon_price_monitor.xlsx" # 历史价格存储文件 # 邮件配置（可选，不需要则注释相关代码） EMAIL_SENDER = "你的邮箱@qq.com" EMAIL_AUTHOR_CODE = "你的邮箱授权码" EMAIL_RECEIVER = "接收提醒的邮箱@xxx.com" SMTP_SERVER = "smtp.qq.com" SMTP_PORT = 587 # ==================================================================== def get_amazon_product_price(asin=TARGET_ASIN): """调用亚马逊API，获取商品当前价格（核心函数）""" try: # 初始化API客户端 api_client = ApiClient(AMAZON_CONFIG) get_items_api = GetItemsApi(api_client) # 构造请求（仅获取必要字段，减少数据量） request = GetItemsRequest( partner_tag=PARTNER_TAG, partner_type=PartnerType.ASSOCIATES, item_ids=[asin], resources=[ Resource.ITEMS_TITLE, Resource.ITEMS_OFFERS_LISTINGS_PRICE, Resource.ITEMS_OFFERS_LISTINGS_AVAILABILITY, Resource.ITEMS_DETAIL_PAGE_URL ] ) # 发送请求并解析响应 response = get_items_api.get_items(request) if not response.items or len(response.items) == 0: print("未获取到商品数据") return None item = response.items[0] # 安全提取核心数据（多层嵌套用判断避免报错） price_data = { "ASIN": item.asin or "未知", "标题": (item.item_info.title.display_value[:50] + "...") if (item.item_info and item.item_info.title) else "未知", "当前价格": item.offers.listings[0].price.amount if (item.offers and item.offers.listings and item.offers.listings[0].price) else 0.0, "货币": item.offers.listings[0].price.currency if (item.offers and item.offers.listings and item.offers.listings[0].price) else "USD", "库存状态": item.offers.listings[0].availability.display_value if (item.offers and item.offers.listings and item.offers.listings[0].availability) else "未知", "详情链接": item.detail_page_url or "未知", "查询时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } return price_data except ApiException as e: print(f"API调用失败：{e.response_body}") return None except Exception as e: print(f"解析数据失败：{str(e)}") return None def save_price_history(price_data): """保存价格数据到Excel（追加模式）""" if not price_data: return False new_df = pd.DataFrame([price_data]) # 追加到历史文件，避免重复 try: history_df = pd.read_excel(HISTORY_FILE) combined_df = pd.concat([history_df, new_df], ignore_index=True).drop_duplicates(subset=["ASIN", "查询时间"]) except FileNotFoundError: combined_df = new_df combined_df.to_excel(HISTORY_FILE, index=False, encoding="utf-8") print(f"历史数据已更新，共{len(combined_df)}条记录") return True def send_email_alert(price_data): """价格达标时发送邮件提醒（可选）""" if not price_data: return # 构造邮件内容 content = f""" <h2>亚马逊商品价格提醒</h2> <p>监控时间：{price_data['查询时间']}</p> <p>商品：{price_data['标题']}</p> <p>当前价格：{price_data['当前价格']} {price_data['货币']}</p> <p>目标阈值：{PRICE_THRESHOLD} {price_data['货币']}</p> <p>库存状态：{price_data['库存状态']}</p> <p>购买链接：<a href="{price_data['详情链接']}">{price_data['详情链接']}</a></p> """ msg = MIMEText(content, "html", "utf-8") msg["From"] = Header("亚马逊价格监控", "utf-8") msg["To"] = Header("用户", "utf-8") msg["Subject"] = Header(f"🎉 {price_data['标题'][:20]} 价格达标！", "utf-8") # 发送邮件 try: with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server: server.starttls() server.login(EMAIL_SENDER, EMAIL_AUTHOR_CODE) server.sendmail(EMAIL_SENDER, EMAIL_RECEIVER, msg.as_string()) print("提醒邮件已发送！") except Exception as e: print(f"邮件发送失败：{str(e)}") def check_price_and_alert(price_data): """对比价格阈值，触发提醒""" if not price_data: return # 打印监控日志 print(f"\n=== {price_data['查询时间']} 监控结果 ===") print(f"商品：{price_data['标题']}") print(f"当前价格：{price_data['当前价格']} {price_data['货币']}") print(f"库存状态：{price_data['库存状态']}") # 价格判断（低于阈值+有库存才提醒） in_stock = "In Stock" in price_data["库存状态"] or "有货" in price_data["库存状态"] if price_data["当前价格"] > 0 and price_data["当前价格"] <= PRICE_THRESHOLD and in_stock: print(f"\n🎯 价格达标！当前价格低于{PRICE_THRESHOLD} {price_data['货币']}") send_email_alert(price_data) # 触发邮件提醒 else: print(f"\n⏳ 价格未达标（目标：≤{PRICE_THRESHOLD} {price_data['货币']}）") def run_monitor(): """启动监控主程序""" print(f"\n=== 亚马逊价格监控已启动 ===") print(f"监控商品：{TARGET_ASIN}") print(f"价格阈值：{PRICE_THRESHOLD} {AMAZON_CONFIG.host.split('.')[1].upper()}") print(f"监控频率：每{MONITOR_INTERVAL}小时") print(f"启动时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print("="*60) # 首次运行立即执行（不等待） price_data = get_amazon_product_price() if price_data: save_price_history(price_data) check_price_and_alert(price_data) # 配置定时任务 schedule.every(MONITOR_INTERVAL).hours.do(lambda: [ save_price_history(pd), check_price_and_alert(pd) ] if (pd := get_amazon_product_price()) else None) # 循环执行定时任务 while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次任务 if __name__ == "__main__": try: run_monitor() except KeyboardInterrupt: print(f"\n监控已手动停止（停止时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}）") except Exception as e: print(f"\n监控异常停止：{str(e)}")

四、关键步骤解析

1. API 调用与数据解析

• 使用amazon-pa-api5-python库简化 AWS4 签名（手动实现签名极其复杂，新手必用）；
• 「安全取值」是核心：亚马逊 JSON 层级深，用if判断 +or默认值避免KeyError，比如item.offers.listings[0].price.amount需先判断offers/listings是否存在；
• 仅请求必要字段（通过resources参数），减少数据传输和解析耗时。

2. 历史数据存储

• 用 Excel 文件存储（新手友好，无需数据库），通过pandas实现「追加写入 + 去重」，避免重复记录；
• 文件名固定，可随时打开查看价格走势。

3. 价格提醒逻辑

• 双重判断：价格≤阈值且商品有库存（In Stock/ 有货），避免提醒无货商品；
• 邮件提醒：通过smtplib发送 HTML 格式邮件，包含商品链接，直接点击即可购买。

4. 定时任务

• 用schedule库实现定时调用，every(MONITOR_INTERVAL).hours可灵活调整频率（如every(30).minutes每 30 分钟）；
• 循环中time.sleep(60)减少 CPU 占用，仅每分钟检查一次任务是否需要执行。

五、使用说明

1. 修改配置：替换代码中Access Key、Secret Key、Partner Tag、TARGET_ASIN、PRICE_THRESHOLD等参数；
2. 运行脚本：终端执行python amazon_price_monitor.py，启动后会立即执行一次监控，之后按设定频率运行；
3. 查看结果：
   • 控制台实时打印监控日志；
   • 生成amazon_price_monitor.xlsx文件，记录所有历史价格；
   • 价格达标时自动发送邮件提醒。

六、避坑指南

1. API 调用失败：
   • 检查Access Key/Secret Key是否正确，Partner Tag是否绑定对应站点；
   • 确认 API 配额未耗尽（亚马逊 PA API 5.0 每天有调用次数限制，避免高频调用）；
2. 价格为 0.0：商品无报价 / 下架，检查 ASIN 是否有效，或等待商品重新上架；
3. 邮件发送失败：
   • 开启邮箱 SMTP 服务（如 QQ 邮箱在「设置 - 账户」中开启）；
   • 确认授权码正确，SMTP 服务器 / 端口匹配
4. 中文乱码：保存 Excel 时指定encoding="utf-8"，邮件内容用utf-8编码。

总结

1. 亚马逊价格监控核心是「合规调用 API + 安全解析数据 + 定时触发 + 阈值提醒」，本次代码覆盖全流程，新手只需修改配置即可运行；
2. 关键优化点：仅请求必要字段、安全取值避免报错、Excel 追加存储、合理控制监控频率（避免 API 限流）；
3. 扩展方向：支持多商品监控（遍历 ASIN 列表）、价格走势可视化（用 matplotlib 绘图）、微信 / 短信提醒（对接第三方短信接口）。