你想通过 Python 结合亚马逊商品详情 API 实现商品价格监控，核心需求是自动获取商品价格、对比历史价格、价格达标时触发提醒，我会帮你完成「从 API 请求、数据存储、价格对比到自动提醒」的全流程，兼顾新手可操作性和实用性，同时解决亚马逊 API 的核心坑点。
一、核心前提说明
- API 选择：仍使用亚马逊「Product Advertising API (PA API 5.0)」，需提前获取
Access Key、Secret Key、Partner Tag（跟踪 ID），且商品需用唯一标识「ASIN」（亚马逊商品专属 ID，可从商品详情页 URL 获取）； - 监控核心流程：
定时调用API获取当前价格 → 存储历史价格（本地文件/简易数据库） → 对比当前价格与阈值/历史低价 → 价格达标触发提醒（控制台/邮件） → 循环执行实现持续监控； - 必备 Python 库：
requests（API 请求）、pandas（数据存储 / 分析）、schedule（定时任务，实现自动监控）、orjson（可选，高效解析 JSON），邮件提醒需内置smtplib/email。
二、前置准备工作
1. 安装必备 Python 库
bash
运行
# 核心库：API请求+数据存储+定时任务 pip install requests pandas schedule # 可选：高效JSON解析（大批量数据用） pip install orjson
2. 获取关键信息
- 亚马逊 PA API 5.0 的
Access Key、Secret Key、Partner Tag（需在亚马逊联盟后台申请）； - 目标商品的
ASIN（商品详情页 URL 示例：，其中B08CZJ7Z8L即为 ASIN）； - （可选）邮箱账号（用于邮件提醒，需开启 SMTP 服务，如 QQ 邮箱开启授权码）。
3. 核心预设（新手可直接修改）
- 监控频率：每 1 小时获取一次价格（可调整）；
- 价格阈值：设置目标低价（如 AirPods Pro 2 代低于 200 美元触发提醒）；
- 数据存储：本地 Excel 文件（简单易上手，无需搭建数据库）。
三、步骤 1：封装亚马逊 API 请求，获取商品当前价格
首先封装 API 请求函数，核心解决「多层 JSON 解析」和「安全取值」，提取商品的ASIN、标题、当前价格、库存状态（监控核心数据）。
注意：亚马逊 PA API 5.0 要求AWS4 签名验证，手动实现复杂，新手推荐使用第三方库
amazon-pa-api5-python简化签名流程（避免陷入签名细节），先安装该库：bash
运行
pip install amazon-pa-api5-python
python
运行
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 # ---------------------- 全局配置（修改为你自己的信息） ---------------------- AMAZON_CONFIG = Configuration() AMAZON_CONFIG.access_key = "你的亚马逊Access Key" AMAZON_CONFIG.secret_key = "你的亚马逊Secret Key" AMAZON_CONFIG.host = "webservices.amazon.com" # 美国站点，其他站点需修改（如cn为webservices.amazon.cn） PARTNER_TAG = "你的Partner Tag（跟踪ID）" TARGET_ASIN = "B08CZJ7Z8L" # 要监控的商品ASIN PRICE_THRESHOLD = 200.0 # 价格阈值（低于该价格触发提醒） HISTORY_FILE = "亚马逊商品价格监控历史.xlsx" # 历史价格存储文件 # ---------------------------------------------------------------------------- def get_amazon_product_current_price(asin=TARGET_ASIN): """ 调用亚马逊PA API 5.0，获取商品当前价格 """ try: # 1. 初始化API客户端 api_client = ApiClient(AMAZON_CONFIG) get_items_api = GetItemsApi(api_client) # 2. 构造请求（指定返回字段，减少冗余数据） get_items_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 ] ) # 3. 发送请求，获取响应 response = get_items_api.get_items(get_items_request) # 4. 解析响应数据（安全取值，应对多层嵌套） if response.items and len(response.items) > 0: item = response.items[0] # 提取商品基础信息 asin = item.asin or "未知ASIN" title = item.item_info.title.display_value if (item.item_info and item.item_info.title) else "未知标题" detail_url = item.detail_page_url or "未知链接" # 提取当前价格（核心监控数据） current_price = 0.0 currency = "USD" if item.offers and item.offers.listings and len(item.offers.listings) > 0: listing = item.offers.listings[0] if listing.price: current_price = listing.price.amount or 0.0 currency = listing.price.currency or "USD" # 提取库存状态（确保商品可购买） availability = "未知库存" if item.offers and item.offers.listings and len(item.offers.listings) > 0: listing = item.offers.listings[0] if listing.availability and listing.availability.display_value: availability = listing.availability.display_value # 整理返回结果 return { "ASIN": asin, "商品标题": title[:50] + "..." if len(title) > 50 else title, # 截断过长标题 "当前价格": current_price, "货币单位": currency, "库存状态": availability, "详情页链接": detail_url, "查询时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } else: print("未解析到商品价格数据") return None except ApiException as e: print(f"API调用失败：{e.response_body}") return None except Exception as e: print(f"获取价格失败：{str(e)}") return None
四、步骤 2：封装数据存储函数，记录价格历史
使用 Excel 文件存储历史价格（新手友好，无需数据库），实现「新增数据追加写入」，方便后续查看价格走势。
python
运行
def save_price_history(price_data): """ 保存商品价格数据到Excel，追加历史记录 """ if not price_data: return False # 1. 转换为DataFrame new_df = pd.DataFrame([price_data]) # 2. 判断历史文件是否存在，存在则追加，不存在则新建 try: # 读取历史数据 history_df = pd.read_excel(HISTORY_FILE) # 追加新数据（去重，避免重复记录同一时间的价格） combined_df = pd.concat([history_df, new_df], ignore_index=True).drop_duplicates(subset=["ASIN", "查询时间"], keep="last") except FileNotFoundError: # 文件不存在，直接使用新数据 combined_df = new_df # 3. 保存到Excel combined_df.to_excel(HISTORY_FILE, index=False, encoding="utf-8") print(f"价格历史已更新，当前共 {len(combined_df)} 条记录") return True
五、步骤 3：封装价格对比与提醒函数
实现「价格阈值判断」，当当前价格低于预设阈值且商品有库存时，触发提醒（先实现控制台提醒，再扩展邮件提醒）。
1. 基础控制台提醒
python
运行
def check_price_and_alert(price_data): """ 对比价格阈值，触发提醒 """ if not price_data: return asin = price_data["ASIN"] title = price_data["商品标题"] current_price = price_data["当前价格"] currency = price_data["货币单位"] availability = price_data["库存状态"] query_time = price_data["查询时间"] detail_url = price_data["详情页链接"] # 1. 打印当前价格信息 print(f"\n=== {query_time} 价格监控报告 ===") print(f"商品ASIN：{asin}") print(f"商品标题：{title}") print(f"当前价格：{current_price} {currency}") print(f"库存状态：{availability}") print(f"详情链接：{detail_url}") # 2. 价格判断（低于阈值+有库存，触发提醒） in_stock = "In Stock" in availability or "有货" in availability # 兼容中英文站点 if current_price > 0 and current_price <= PRICE_THRESHOLD and in_stock: print(f"\n🎉 价格达标提醒！🎉") print(f"当前价格 {current_price} {currency} 低于阈值 {PRICE_THRESHOLD} {currency}") print(f"商品可购买，建议立即下单：{detail_url}") # 此处可扩展：邮件提醒、微信提醒等 send_email_alert(price_data) # 后续实现邮件提醒 else: print(f"\n😴 价格未达标，继续监控（目标阈值：{PRICE_THRESHOLD} {currency}）")
2. 扩展邮件提醒（可选，更实用）
python
运行
import smtplib from email.mime.text import MIMEText from email.header import Header # ---------------------- 邮件配置（修改为你自己的邮箱信息） ---------------------- EMAIL_SENDER = "你的发送邮箱@qq.com" # 如QQ邮箱 EMAIL_AUTHOR_CODE = "你的邮箱授权码" # QQ邮箱需开启SMTP，获取授权码 EMAIL_RECEIVER = "接收提醒的邮箱@xxx.com" SMTP_SERVER = "smtp.qq.com" SMTP_PORT = 587 # ---------------------------------------------------------------------------- def send_email_alert(price_data): """ 发送价格达标邮件提醒 """ if not price_data: return # 1. 构造邮件内容 title = price_data["商品标题"] current_price = price_data["当前价格"] currency = price_data["货币单位"] threshold = PRICE_THRESHOLD detail_url = price_data["详情页链接"] query_time = price_data["查询时间"] email_content = f""" <h3>亚马逊商品价格达标提醒</h3> <p>监控时间：{query_time}</p> <p>商品标题：{title}</p> <p>当前价格：{current_price} {currency}</p> <p>目标阈值：{threshold} {currency}</p> <p>库存状态：{price_data['库存状态']}</p> <p>立即购买：<a href="{detail_url}">{detail_url}</a></p> """ # 2. 配置邮件对象 msg = MIMEText(email_content, "html", "utf-8") msg["From"] = Header("亚马逊价格监控机器人", "utf-8") msg["To"] = Header("用户", "utf-8") msg["Subject"] = Header(f"🎉 商品价格达标！{title[:20]}...", "utf-8") # 3. 发送邮件 try: with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server: server.starttls() # 开启TLS加密 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)}")
六、步骤 4：封装定时任务，实现持续监控
使用schedule库实现定时调用（如每 1 小时执行一次），无需手动运行脚本，实现 7x24 小时持续监控。
python
运行
def run_price_monitor(): """ 运行价格监控主流程 """ print(f"\n=== 启动亚马逊商品价格监控 ===") print(f"监控商品ASIN：{TARGET_ASIN}") print(f"价格阈值：{PRICE_THRESHOLD} {AMAZON_CONFIG.host.split('.')[1].upper()}") print(f"历史数据文件：{HISTORY_FILE}") print(f"监控频率：每1小时执行一次") print(f"启动时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print("=" * 50) # 1. 立即执行一次监控（启动时不等待） price_data = get_amazon_product_current_price() if price_data: save_price_history(price_data) check_price_and_alert(price_data) # 2. 配置定时任务（每1小时执行一次，可修改为其他频率） schedule.every(1).hours.do(lambda: [ save_price_history(pd), check_price_and_alert(pd) ] if (pd := get_amazon_product_current_price()) else None) # 3. 循环运行定时任务 while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次任务是否需要执行
七、步骤 5：启动监控，完整运行
添加主函数调用，直接运行脚本即可启动持续监控。
python
运行
if __name__ == "__main__": try: run_price_monitor() except KeyboardInterrupt: print(f"\n监控已被用户手动停止，停止时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") except Exception as e: print(f"\n监控异常停止：{str(e)}")
总结
- 亚马逊商品价格监控的核心流程是「定时获取价格 → 存储历史 → 阈值对比 → 触发提醒」，新手可通过 Excel 存储、控制台 / 邮件提醒快速落地；
- 关键在于「安全解析多层嵌套 JSON」和「合理使用定时任务避免高频调用」，同时借助第三方签名库简化亚马逊 API 的使用；
- 本次代码可直接修改配置信息运行，实现持续监控，后续可扩展多商品、价格走势可视化、微信提醒等功能，满足更复杂的监控需求。
