Les données e-commerce provenant des API de recherche peuvent contenir des prix manquants, des annonces obsolètes, des évaluations incorrectes et des produits en double. Avant d'alimenter ces données dans des moteurs de tarification, des systèmes d'inventaire ou des tableaux de bord analytiques, vous avez besoin d'une validation automatisée. Ce tutoriel construit un pipeline de qualité des données qui vérifie les données produits de l'API Scavio pour l'exhaustivité, les anomalies de prix, la fraîcheur et les doublons. Chaque recherche de produit coûte 0,005 $ et la logique de validation s'exécute localement sans coût supplémentaire.
Prérequis
- Python 3.9+ installé
- bibliothèque requests installée
- Une clé API Scavio depuis scavio.dev
- Compréhension de base des structures de données produits
Parcours
Étape 1: Récupérer les données produit depuis l'API
Recherchez des produits sur Amazon via l'API Scavio. La réponse inclut le titre, le prix, l'évaluation, le nombre d'avis et la disponibilité.
Python
import requests, os
API_KEY = os.environ['SCAVIO_API_KEY']
def fetch_products(query: str) -> list:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
json={'platform': 'amazon', 'query': query, 'marketplace': 'US'})
resp.raise_for_status()
return resp.json().get('products', [])
products = fetch_products('wireless mouse')
print(f'Fetched {len(products)} products')Étape 2: Vérifier les champs obligatoires manquants
Validez que chaque produit possède les champs essentiels : titre, prix, évaluation. Signalez les enregistrements incomplets afin qu'ils puissent être exclus ou enrichis.
Python
REQUIRED_FIELDS = ['title', 'price', 'rating', 'link']
def check_completeness(products: list) -> dict:
complete = []
incomplete = []
for p in products:
missing = [f for f in REQUIRED_FIELDS if not p.get(f)]
if missing:
incomplete.append({'product': p.get('title', 'unknown'), 'missing': missing})
else:
complete.append(p)
return {
'complete': len(complete),
'incomplete': len(incomplete),
'details': incomplete,
'completeness_rate': len(complete) / max(len(products), 1) * 100
}
report = check_completeness(products)
print(f'Completeness: {report["completeness_rate"]:.0f}% ({report["complete"]}/{report["complete"] + report["incomplete"]})')
for d in report['details']:
print(f' Missing {d["missing"]} for: {d["product"][:50]}')Étape 3: Détecter les anomalies de prix
Signalez les produits dont les prix sont suspectement bas ou élevés par rapport à la médiane. Cela permet de détecter les erreurs de liste et les surfacturations des tiers.
Python
import statistics
def detect_price_anomalies(products: list, threshold: float = 2.0) -> list:
priced = [p for p in products if p.get('price')]
if len(priced) < 3:
return []
prices = []
for p in priced:
price_str = str(p['price']).replace('$', '').replace(',', '')
try:
prices.append(float(price_str))
except ValueError:
continue
if not prices:
return []
median = statistics.median(prices)
stdev = statistics.stdev(prices) if len(prices) > 1 else median * 0.5
anomalies = []
for p, price in zip(priced, prices):
deviation = abs(price - median) / max(stdev, 0.01)
if deviation > threshold:
anomalies.append({
'product': p.get('title', '')[:50],
'price': price,
'median': median,
'deviation': round(deviation, 1)
})
return anomalies
anomalies = detect_price_anomalies(products)
for a in anomalies:
print(f'ANOMALY: ${a["price"]} (median ${a["median"]}) - {a["product"]}')Étape 4: Vérifier les annonces en double
Détectez les produits en double ou quasi-doubles en comparant les titres normalisés. Cela évite le double comptage dans l'analyse des prix.
Python
from difflib import SequenceMatcher
def find_duplicates(products: list, similarity_threshold: float = 0.85) -> list:
dupes = []
titles = [(i, p.get('title', '').lower().strip()) for i, p in enumerate(products)]
for i in range(len(titles)):
for j in range(i + 1, len(titles)):
ratio = SequenceMatcher(None, titles[i][1], titles[j][1]).ratio()
if ratio >= similarity_threshold:
dupes.append({
'product_a': products[titles[i][0]].get('title', '')[:50],
'product_b': products[titles[j][0]].get('title', '')[:50],
'similarity': round(ratio * 100, 1)
})
return dupes
dupes = find_duplicates(products)
for d in dupes:
print(f'DUPLICATE ({d["similarity"]}%): {d["product_a"]} <-> {d["product_b"]}')Étape 5: Générer un tableau de bord de qualité
Combinez toutes les vérifications en un seul score de qualité. Utilisez-le pour décider si vous pouvez faire confiance aux données ou les récupérer à nouveau.
Python
def quality_scorecard(products: list) -> dict:
completeness = check_completeness(products)
anomalies = detect_price_anomalies(products)
duplicates = find_duplicates(products)
total = len(products)
score = 100
score -= (completeness['incomplete'] / max(total, 1)) * 30 # up to -30 for missing fields
score -= len(anomalies) * 5 # -5 per anomaly
score -= len(duplicates) * 3 # -3 per duplicate pair
return {
'total_products': total,
'quality_score': max(round(score, 1), 0),
'completeness_rate': round(completeness['completeness_rate'], 1),
'anomaly_count': len(anomalies),
'duplicate_pairs': len(duplicates),
'verdict': 'PASS' if score >= 70 else 'REVIEW' if score >= 50 else 'FAIL'
}
card = quality_scorecard(products)
for k, v in card.items():
print(f'{k}: {v}')Exemple Python
Python
import os, requests, statistics
API_KEY = os.environ['SCAVIO_API_KEY']
def fetch_products(query: str) -> list:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
json={'platform': 'amazon', 'query': query, 'marketplace': 'US'})
return resp.json().get('products', [])
def validate(products: list) -> dict:
total = len(products)
complete = sum(1 for p in products if all(p.get(f) for f in ['title', 'price', 'rating']))
return {'total': total, 'complete': complete, 'rate': f'{complete/max(total,1)*100:.0f}%'}
products = fetch_products('wireless mouse')
print(validate(products))Exemple JavaScript
JavaScript
const API_KEY = process.env.SCAVIO_API_KEY;
async function fetchProducts(query) {
const resp = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': API_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ platform: 'amazon', query, marketplace: 'US' })
});
return (await resp.json()).products || [];
}
async function main() {
const products = await fetchProducts('wireless mouse');
const complete = products.filter(p => p.title && p.price && p.rating).length;
console.log(`${complete}/${products.length} complete (${(complete/products.length*100).toFixed(0)}%)`);
}
main().catch(console.error);Sortie attendue
JSON
Fetched 15 products
Completeness: 87% (13/15)
Missing ['price'] for: Generic USB Mouse Adapter Cable...
Missing ['rating'] for: Bulk Pack Wireless Mouse 10-Unit...
ANOMALY: $149.99 (median $24.99) - Premium Ergonomic Wireless Mouse Gold
DUPLICATE (91.2%): Logitech M720 Triathlon Multi-Device <-> Logitech M720 Triathlon Wireless
total_products: 15
quality_score: 78.0
completeness_rate: 86.7
anomaly_count: 1
duplicate_pairs: 1
verdict: PASS