4大核心组件：教你如何构建Amazon产品监控系统并实现数据自动化

CREATE TABLE products_to_monitor (
    id SERIAL PRIMARY KEY,
    asin VARCHAR(20) UNIQUE NOT NULL,
    url TEXT NOT NULL,
    last_known_price DECIMAL(10, 2),
    last_checked_at TIMESTAMP WITH TIME ZONE,
    is_active BOOLEAN DEFAULT TRUE,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

CREATE TABLE price_history (
    id SERIAL PRIMARY KEY,
    product_id INTEGER REFERENCES products_to_monitor(id),
    price DECIMAL(10, 2) NOT NULL,
    scraped_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

# scheduler_setup.py
from apscheduler.schedulers.blocking import BlockingScheduler
import time

def my_monitoring_task():
    """
    这是一个伪任务，代表了我们希望周期性执行的监控主逻辑。
    """
    print(f"监控任务正在执行... 时间: {time.ctime()}")
    # 在这里，我们将调用数据采集和变更检测的逻辑
    # run_monitoring_logic()

# 创建一个调度器实例
scheduler = BlockingScheduler()

# 添加一个任务，设定为每小时执行一次 my_monitoring_task 函数
scheduler.add_job(my_monitoring_task, 'interval', hours=1, id='amazon_monitor_job')

print("调度器已启动，将每小时执行一次监控任务。按 Ctrl+C 退出。")

try:
    scheduler.start()
except (KeyboardInterrupt, SystemExit):
    pass

# data_acquirer.py
import requests
import os

# 从环境变量或配置文件中读取API密钥，更安全
API_TOKEN = os.getenv("PANGOLIN_API_TOKEN") 
API_BASE_URL = "http://scrapeapi.pangolinfo.com/api/v1"

def get_product_data_with_scrapeapi(asin: str, zipcode: str = "10041") -> dict:
    """
    使用Pangolin Scrape API获取单个ASIN的商品数据。
    """
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {API_TOKEN}"
    }
    
    payload = {
        "url": f"https://www.amazon.com/dp/{asin}",
        "formats": ["json"],
        "parserName": "amzProductDetail", # 使用预设的亚马逊商品详情解析器
        "bizContext": {
            "zipcode": zipcode
        }
    }
    
    try:
        response = requests.post(API_BASE_URL, headers=headers, json=payload, timeout=60)
        response.raise_for_status()
        
        api_response = response.json()
        if api_response.get("code") == 0:
            # 假设解析后的JSON数据在 data['json'][0] 中
            # 注意：根据实际API返回结构，可能需要调整这里的解析路径
            parsed_data = api_response.get("data", {}).get("json", [None])[0]
            if parsed_data:
                return parsed_data
            else:
                print(f"API成功返回，但未能解析出JSON数据 for ASIN: {asin}")
                return None
        else:
            print(f"API调用出错 for ASIN {asin}: {api_response.get('message')}")
            return None

    except requests.exceptions.RequestException as e:
        print(f"请求API时发生网络错误 for ASIN {asin}: {e}")
        return None

# main_monitor.py
from apscheduler.schedulers.blocking import BlockingScheduler
from data_acquirer import get_product_data_with_scrapeapi # 导入我们的数据采集函数
import database_connector # 假设这是一个您自己编写的，用于连接和操作数据库的模块

def main_monitoring_task():
    print("开始执行主监控任务...")
    
    # 1. 从数据库获取所有需要监控的活跃商品ASIN
    products = database_connector.get_active_products()
    
    for product in products:
        asin = product['asin']
        last_price = product['last_known_price']
        
        print(f"正在检查 ASIN: {asin}...")
        
        # 2. 使用Scrape API获取最新商品数据
        current_data = get_product_data_with_scrapeapi(asin)
        
        if not current_data:
            print(f"无法获取 ASIN: {asin} 的数据，跳过本次检查。")
            continue
            
        # 3. 变更检测：对比价格
        # 注意：current_data['price']中的字段名需要与API返回的JSON结构一致
        current_price = current_data.get('price') 
        
        if current_price and current_price != last_price:
            print(f"!!! 价格变动警报 for ASIN: {asin} !!!")
            print(f"旧价格: {last_price}, 新价格: {current_price}")
            
            # 4. 触发通知 (这里仅做打印，实际应用中会调用邮件/Slack等)
            # send_alert(f"ASIN {asin} 价格从 {last_price} 变为 {current_price}")
            
            # 5. 更新数据库
            database_connector.update_product_price(asin, current_price)
            database_connector.log_price_history(asin, current_price)
        else:
            print(f"ASIN: {asin} 价格未变。")
            
    print("主监控任务执行完毕。")

# --- 调度器设置 ---
scheduler = BlockingScheduler()
scheduler.add_job(main_monitoring_task, 'interval', hours=1, id='amazon_monitor_job')
print("监控系统已启动，将每小时运行。")
scheduler.start()