Cómo crear un rastreador automatizado de precios de la competencia

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'}

PRODUCTS = ['wireless earbuds', 'mechanical keyboard', 'USB-C hub']

def get_prices(product, platform=None):
    body = {'query': product, 'country_code': 'us'}
    if platform:
        body['platform'] = platform
    data = requests.post('https://api.scavio.dev/api/v1/search',
        headers=SH, json=body).json()
    results = data.get('shopping_results', data.get('organic_results', []))
    prices = []
    for r in results[:5]:
        price = r.get('price', r.get('extracted_price', ''))
        prices.append({
            'title': r.get('title', '')[:60],
            'price': price,
            'source': r.get('source', r.get('displayed_link', '')),
            'link': r.get('link', ''),
            'platform': platform or 'google'
        })
    return prices

for product in PRODUCTS:
    print(f'\n{product}:')
    for platform in [None, 'amazon', 'walmart']:
        prices = get_prices(product, platform)
        label = platform or 'google'
        if prices:
            lowest = min(prices, key=lambda p: float(str(p['price']).replace('$', '').replace(',', '') or '999'))
            print(f'  [{label:8}] {len(prices)} listings, lowest: {lowest["price"]} - {lowest["title"][:35]}')
        else:
            print(f'  [{label:8}] No results')
print(f'\nCost: ${len(PRODUCTS) * 3 * 0.005:.3f}')

def track_price_changes(products, history_file='price_history.json'):
    try:
        with open(history_file) as f:
            history = json.load(f)
    except FileNotFoundError:
        history = {}
    today = datetime.now().strftime('%Y-%m-%d')
    changes = []
    for product in products:
        prices = get_prices(product, 'amazon')
        if not prices:
            continue
        top_price = prices[0]
        price_val = str(top_price.get('price', '')).replace('$', '').replace(',', '')
        try:
            price_num = float(price_val)
        except ValueError:
            continue
        key = f'{product}_amazon'
        if key in history:
            prev = history[key]['price']
            if abs(price_num - prev) > 0.01:
                pct = (price_num - prev) / prev * 100
                direction = 'UP' if price_num > prev else 'DOWN'
                changes.append({
                    'product': product, 'prev': prev, 'current': price_num,
                    'change_pct': pct, 'direction': direction
                })
        history[key] = {'price': price_num, 'date': today, 'title': top_price['title']}
    with open(history_file, 'w') as f:
        json.dump(history, f, indent=2)
    if changes:
        print(f'\n=== Price Changes Detected ===')
        for c in changes:
            print(f'  [{c["direction"]:4}] {c["product"]}: ${c["prev"]:.2f} -> ${c["current"]:.2f} ({c["change_pct"]:+.1f}%)')
    else:
        print(f'\nNo price changes detected (or first scan).')
    return changes

track_price_changes(PRODUCTS)

def price_comparison_report(products):
    print(f'\n=== Cross-Platform Price Report - {datetime.now().strftime("%Y-%m-%d")} ===')
    for product in products:
        print(f'\n  {product.upper()}')
        all_prices = []
        for platform in [None, 'amazon', 'walmart']:
            prices = get_prices(product, platform)
            all_prices.extend(prices)
        # Sort by price
        for p in all_prices:
            try:
                p['price_num'] = float(str(p['price']).replace('$', '').replace(',', ''))
            except (ValueError, TypeError):
                p['price_num'] = 999
        all_prices.sort(key=lambda x: x['price_num'])
        for p in all_prices[:5]:
            print(f'    ${p["price_num"]:8.2f} | {p["platform"]:8} | {p["title"][:35]}')
        if len(all_prices) >= 2:
            spread = all_prices[-1]['price_num'] - all_prices[0]['price_num']
            print(f'    Price spread: ${spread:.2f}')
    total_queries = len(products) * 3
    print(f'\n  Total cost: ${total_queries * 0.005:.3f} ({total_queries} queries)')

price_comparison_report(PRODUCTS)