Surveiller quotidiennement les ASIN Amazon pour les variations de prix, les changements de note et la disponibilité des stocks est essentiel pour la veille e-commerce. Les solutions basées sur des scrapers nécessitent une maintenance constante car Amazon modifie les mises en page et renforce la détection des robots. Ce tutoriel construit un pipeline de surveillance stable utilisant l'API Scavio qui vérifie une liste d'ASIN selon un calendrier, stocke les données historiques, détecte les changements significatifs et envoie des alertes. Chaque vérification d'ASIN coûte $0.005 avec zéro frais de maintenance.
Prérequis
- Python 3.9+ installé
- bibliothèque requests installée
- Une clé API Scavio depuis scavio.dev
- Une liste d'ASIN à surveiller
Parcours
Étape 1: Définir la liste de surveillance des ASIN et le stockage
Mettre en place la liste des ASIN à surveiller et un système de stockage basé sur JSON pour le suivi historique.
Python
import os, requests, json, time
from datetime import datetime
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
H = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'}
URL = 'https://api.scavio.dev/api/v1/search'
WATCHLIST = ['B09G9FPHY6', 'B07FZ8S74R', 'B08N5WRWNW', 'B0BSHF7WHW']
HISTORY_FILE = 'asin_history.json'
def load_history() -> dict:
if os.path.exists(HISTORY_FILE):
with open(HISTORY_FILE) as f:
return json.load(f)
return {}
def save_history(history: dict):
with open(HISTORY_FILE, 'w') as f:
json.dump(history, f, indent=2)Étape 2: Récupérer les données produit actuelles pour chaque ASIN
Interroger l'API Scavio pour chaque ASIN et extraire le prix, la note, le nombre d'avis et la disponibilité.
Python
def fetch_asin(asin: str) -> dict:
resp = requests.post(URL, headers=H,
json={'platform': 'amazon', 'query': asin, 'marketplace': 'US'})
resp.raise_for_status()
data = resp.json()
product = data.get('product', data)
price_str = product.get('price', '')
price = None
if price_str:
price = float(price_str.replace('$', '').replace(',', ''))
return {
'asin': asin,
'title': product.get('title', ''),
'price': price,
'rating': product.get('rating', ''),
'reviews': product.get('reviews_count', 0),
'availability': product.get('availability', ''),
'timestamp': datetime.now().isoformat(),
}
for asin in WATCHLIST[:2]:
data = fetch_asin(asin)
print(f'{asin}: ${data["price"]} | {data["rating"]} stars | {data["title"][:40]}')
time.sleep(0.3)Étape 3: Détecter les changements et générer des alertes
Comparer les données actuelles avec les instantanés précédents pour détecter les baisses de prix, les changements de note et les problèmes de stock.
Python
def check_changes(asin: str, current: dict, history: dict) -> list:
alerts = []
prev_records = history.get(asin, [])
if not prev_records:
return alerts
prev = prev_records[-1]
# Price change
if current['price'] and prev.get('price'):
pct_change = (current['price'] - prev['price']) / prev['price'] * 100
if abs(pct_change) > 5:
direction = 'dropped' if pct_change < 0 else 'increased'
alerts.append(f'PRICE {direction} {abs(pct_change):.1f}%: ${prev["price"]} -> ${current["price"]}')
# Rating change
if current['rating'] and prev.get('rating') and current['rating'] != prev['rating']:
alerts.append(f'RATING changed: {prev["rating"]} -> {current["rating"]}')
# Stock status
if 'out of stock' in (current.get('availability', '') or '').lower():
alerts.append('OUT OF STOCK')
return alerts
def run_monitoring():
history = load_history()
for asin in WATCHLIST:
current = fetch_asin(asin)
alerts = check_changes(asin, current, history)
if asin not in history:
history[asin] = []
history[asin].append(current)
status = f'[ALERT] {", ".join(alerts)}' if alerts else '[OK]'
print(f'{status} {asin}: ${current["price"]} | {current["title"][:35]}')
time.sleep(0.3)
save_history(history)
print(f'\nMonitored {len(WATCHLIST)} ASINs | Cost: ${len(WATCHLIST) * 0.005:.3f}')
run_monitoring()Exemple Python
Python
import os, requests, json, time
from datetime import datetime
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
H = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'}
def monitor_asins(asins):
for asin in asins:
resp = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'platform': 'amazon', 'query': asin, 'marketplace': 'US'})
p = resp.json().get('product', resp.json())
print(f'{asin}: {p.get("price", "N/A")} | {p.get("rating", "N/A")} | {p.get("title", "")[:40]}')
time.sleep(0.3)
print(f'Cost: ${len(asins) * 0.005:.3f}')
monitor_asins(['B09G9FPHY6', 'B07FZ8S74R'])Exemple JavaScript
JavaScript
const SCAVIO_KEY = process.env.SCAVIO_API_KEY;
async function monitorAsins(asins) {
for (const asin of asins) {
const resp = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ platform: 'amazon', query: asin, marketplace: 'US' })
});
const p = (await resp.json()).product || {};
console.log(`${asin}: ${p.price || 'N/A'} | ${p.rating || 'N/A'} | ${(p.title || '').slice(0, 40)}`);
}
console.log(`Cost: $${(asins.length * 0.005).toFixed(3)}`);
}
monitorAsins(['B09G9FPHY6', 'B07FZ8S74R']);Sortie attendue
JSON
[OK] B09G9FPHY6: $29.99 | Echo Dot (5th Gen) | Smart speake
[ALERT] PRICE dropped 12.3%: $28.49 -> $24.99 B07FZ8S74R: $24.99 | Fire TV Stick 4K with Alexa
[OK] B08N5WRWNW: $89.99 | Kindle Paperwhite (11th Generatio
[OK] B0BSHF7WHW: $39.99 | Fire TV Stick 4K Max (2nd Gen)
Monitored 4 ASINs | Cost: $0.020