Comment construire un moniteur de produits Amazon sans scrapers

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

def search_amazon(query):
    data = requests.post('https://api.scavio.dev/api/v1/search',
        headers=SH, json={'query': query, 'platform': 'amazon', 'country_code': 'us'}, timeout=10).json()
    products = []
    for r in data.get('organic_results', []):
        products.append({
            'title': r.get('title', ''),
            'link': r.get('link', ''),
            'price': r.get('price', r.get('extracted_price', '')),
            'rating': r.get('rating', ''),
            'reviews': r.get('reviews', ''),
            'position': r.get('position', 0),
            'snippet': r.get('snippet', ''),
        })
    return products

# Monitor competitor products
MONITOR_QUERIES = [
    'wireless bluetooth earbuds',
    'mechanical keyboard gaming',
    'portable phone charger',
]

for query in MONITOR_QUERIES:
    products = search_amazon(query)
    print(f'\n{query}: {len(products)} products')
    for p in products[:3]:
        print(f'  #{p["position"]} {p["title"][:45]}')
        print(f'    Price: {p["price"]} | Rating: {p["rating"]} | Reviews: {p["reviews"]}')
print(f'\nCost: ${len(MONITOR_QUERIES) * 0.005:.3f}')

HISTORY_FILE = 'amazon_monitor_history.json'

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

def save_snapshot(query, products):
    history = load_history()
    today = datetime.now().strftime('%Y-%m-%d')
    if query not in history:
        history[query] = []
    history[query].append({
        'date': today,
        'products': [{'title': p['title'][:60], 'price': p['price'],
                      'position': p['position'], 'rating': p['rating']}
                     for p in products[:10]]
    })
    with open(HISTORY_FILE, 'w') as f:
        json.dump(history, f, indent=2)
    return history[query]

def detect_changes(query, products):
    history = load_history()
    snapshots = history.get(query, [])
    if len(snapshots) < 1:
        return []
    prev = {p['title'][:60]: p for p in snapshots[-1]['products']}
    changes = []
    for p in products[:10]:
        title_key = p['title'][:60]
        if title_key in prev:
            old = prev[title_key]
            if str(p['price']) != str(old.get('price', '')):
                changes.append(f'PRICE: {title_key[:35]} {old["price"]} -> {p["price"]}')
            if p['position'] != old.get('position'):
                changes.append(f'RANK: {title_key[:35]} #{old["position"]} -> #{p["position"]}')
    return changes

for query in MONITOR_QUERIES:
    products = search_amazon(query)
    changes = detect_changes(query, products)
    save_snapshot(query, products)
    if changes:
        print(f'\nChanges for "{query}":')
        for c in changes:
            print(f'  {c}')
    else:
        print(f'No changes for "{query}"')

def generate_alerts(monitor_queries):
    alerts = []
    for query in monitor_queries:
        products = search_amazon(query)
        changes = detect_changes(query, products)
        save_snapshot(query, products)
        for change in changes:
            if 'PRICE' in change:
                alerts.append({'type': 'price', 'query': query, 'detail': change})
            elif 'RANK' in change:
                alerts.append({'type': 'rank', 'query': query, 'detail': change})
    print(f'\n=== Amazon Monitoring Report ===')
    print(f'  Date: {datetime.now().strftime("%Y-%m-%d")}')
    print(f'  Queries monitored: {len(monitor_queries)}')
    print(f'  Alerts: {len(alerts)}')
    if alerts:
        price_alerts = [a for a in alerts if a['type'] == 'price']
        rank_alerts = [a for a in alerts if a['type'] == 'rank']
        if price_alerts:
            print(f'\n  Price Alerts ({len(price_alerts)}):')
            for a in price_alerts:
                print(f'    {a["detail"]}')
        if rank_alerts:
            print(f'\n  Rank Alerts ({len(rank_alerts)}):')
            for a in rank_alerts:
                print(f'    {a["detail"]}')
    else:
        print(f'  No changes detected.')
    print(f'\n  Cost: ${len(monitor_queries) * 0.005:.3f}/scan')
    print(f'  Daily: ${len(monitor_queries) * 0.005:.3f}')
    print(f'  Monthly: ${len(monitor_queries) * 0.005 * 30:.2f}')
    print(f'  No scrapers. No proxies. No maintenance.')

generate_alerts(MONITOR_QUERIES)