How to Build Cross-Platform Product Price Alerts

import os, requests, json
from datetime import datetime

API_KEY = os.environ['SCAVIO_API_KEY']
SH = {'x-api-key': API_KEY, 'Content-Type': 'application/json'}

MONITOR_LIST = [
    {'name': 'AirPods Pro 2', 'query': 'Apple AirPods Pro 2nd generation', 'target_price': 180.00},
    {'name': 'Keychron K8 Pro', 'query': 'Keychron K8 Pro wireless mechanical keyboard', 'target_price': 95.00},
    {'name': 'Anker 737 Charger', 'query': 'Anker 737 portable charger 24000mah', 'target_price': 80.00},
]

PLATFORMS = ['amazon', 'walmart', None]  # None = Google Shopping

def extract_price(text):
    if not text:
        return None
    try:
        cleaned = str(text).replace('$', '').replace(',', '').split('-')[0].split(' ')[0].strip()
        return float(cleaned)
    except (ValueError, IndexError):
        return None

def check_prices(product):
    results = []
    for platform in PLATFORMS:
        body = {'query': product['query'], 'country_code': 'us'}
        if platform:
            body['platform'] = platform
        data = requests.post('https://api.scavio.dev/api/v1/search',
            headers=SH, json=body, timeout=10).json()
        for r in data.get('organic_results', [])[:3]:
            price = extract_price(r.get('price', r.get('extracted_price', '')))
            if price:
                results.append({
                    'platform': platform or 'google',
                    'title': r.get('title', '')[:50],
                    'price': price,
                    'link': r.get('link', ''),
                })
    return results

for product in MONITOR_LIST:
    prices = check_prices(product)
    print(f'\n{product["name"]} (target: ${product["target_price"]:.2f})')
    for p in sorted(prices, key=lambda x: x['price'])[:5]:
        below = ' << BELOW TARGET' if p['price'] <= product['target_price'] else ''
        print(f'  {p["platform"]:8} ${p["price"]:>8.2f} | {p["title"][:40]}{below}')
print(f'\nCost: ${len(MONITOR_LIST) * len(PLATFORMS) * 0.005:.3f}')

PRICE_HISTORY = 'price_alerts_history.json'

def load_price_history():
    try:
        with open(PRICE_HISTORY) as f:
            return json.load(f)
    except FileNotFoundError:
        return {}

def save_price_history(history):
    with open(PRICE_HISTORY, 'w') as f:
        json.dump(history, f, indent=2)

def check_alerts(monitor_list):
    history = load_price_history()
    alerts = []
    today = datetime.now().strftime('%Y-%m-%d')
    for product in monitor_list:
        prices = check_prices(product)
        if not prices:
            continue
        best = min(prices, key=lambda x: x['price'])
        prev_best = history.get(product['name'], {}).get('best_price')
        # Target price alert
        if best['price'] <= product['target_price']:
            alerts.append({
                'type': 'target_hit',
                'product': product['name'],
                'price': best['price'],
                'target': product['target_price'],
                'platform': best['platform'],
                'link': best['link'],
            })
        # Price drop alert
        if prev_best and best['price'] < prev_best * 0.95:
            drop_pct = (1 - best['price'] / prev_best) * 100
            alerts.append({
                'type': 'price_drop',
                'product': product['name'],
                'old_price': prev_best,
                'new_price': best['price'],
                'drop_pct': drop_pct,
                'platform': best['platform'],
            })
        history[product['name']] = {'best_price': best['price'], 'date': today, 'platform': best['platform']}
    save_price_history(history)
    return alerts

alerts = check_alerts(MONITOR_LIST)
print(f'\n=== Price Alerts ===')
print(f'  Alerts: {len(alerts)}')
for a in alerts:
    if a['type'] == 'target_hit':
        print(f'  TARGET: {a["product"]} at ${a["price"]:.2f} on {a["platform"]} (target: ${a["target"]:.2f})')
    elif a['type'] == 'price_drop':
        print(f'  DROP: {a["product"]} ${a["old_price"]:.2f} -> ${a["new_price"]:.2f} (-{a["drop_pct"]:.0f}%) on {a["platform"]}')

def daily_summary(monitor_list):
    print(f'\n{"=" * 60}')
    print(f'  CROSS-PLATFORM PRICE MONITOR - {datetime.now().strftime("%Y-%m-%d")}')
    print(f'{"=" * 60}')
    total_queries = 0
    for product in monitor_list:
        prices = check_prices(product)
        total_queries += len(PLATFORMS)
        if not prices:
            print(f'\n  {product["name"]}: No prices found')
            continue
        best = min(prices, key=lambda x: x['price'])
        worst = max(prices, key=lambda x: x['price'])
        savings = worst['price'] - best['price']
        on_target = best['price'] <= product['target_price']
        status = 'BUY' if on_target else 'WAIT'
        print(f'\n  {product["name"]} [{status}]')
        print(f'    Target: ${product["target_price"]:.2f} | Best: ${best["price"]:.2f} ({best["platform"]})')
        print(f'    Range: ${best["price"]:.2f} - ${worst["price"]:.2f} | Save ${savings:.2f} cross-platform')
    cost = total_queries * 0.005
    print(f'\n  Daily cost: ${cost:.3f}')
    print(f'  Monthly: ${cost * 30:.2f}')
    print(f'  Products: {len(monitor_list)} | Platforms: {len(PLATFORMS)} | Queries: {total_queries}')

daily_summary(MONITOR_LIST)