La veille tarifaire bute sur le même obstacle : les pages de prix et de catalogue se ferment derrière un reCAPTCHA v2 dès qu'un script les visite. Ce tableau de bord collecte ces données, les historise dans SQLite et produit des rapports comparatifs, en résolvant le défi CAPTCHA à la volée avec l'API CaptchaAI.
Le pipeline tient en cinq fichiers Python que vous assemblez de bas en haut : un modèle de données SQLite, un solveur CAPTCHA, un scraper, un générateur de rapports et un point d'entrée qui orchestre le tout. À l'arrivée, une seule commande collecte les prix et catalogues de chaque concurrent, les range dans une base locale et produit un rapport comparatif daté que vous pouvez versionner ou envoyer à votre équipe.
Prérequis
- Python 3.10+ avec les paquets
requestsetbeautifulsoup4. - Une clé API CaptchaAI exportée dans la variable
CAPTCHAAI_API_KEY. - Des cibles publiques uniquement : grilles tarifaires, fiches produits, pages de fonctionnalités déjà publiées.
Ne collectez jamais de données derrière un compte que vous ne possédez pas. Une page de tarifs publique ne contient en principe pas de données personnelles ; dès que votre collecte en touche, appliquez les principes de minimisation du RGPD.
Architecture
Competitor Sites ──> CAPTCHA Solver ──> Data Extractors
│
SQLite Store
│
Dashboard Report
Le flux est linéaire et sans état partagé, ce qui le rend facile à déboguer. Le scraper récupère chaque page ; si un reCAPTCHA v2 s'y trouve, il passe la main au solveur, qui renvoie le HTML une fois le défi résolu. Les extracteurs transforment ce HTML en mesures normalisées, rangées dans SQLite avec leur horodatage. Le générateur relit ensuite la base pour produire le comparatif du jour et les tendances. Chaque étage étant indépendant, vous remplacez le stockage ou ajoutez un extracteur sans toucher au reste.
Modèles de données
Chaque mesure devient une ligne horodatée dans la table SQLite metrics, ce qui permet de comparer l'état du jour et de suivre les tendances. Le choix de SQLite est délibéré : un seul fichier, aucun serveur à installer, et du SQL suffisant pour agréger et comparer. La classe CompetitorData normalise chaque relevé — concurrent, métrique, valeur brute, valeur numérique, URL, horodatage — pour que le rapport n'ait rien à deviner. Conservez la valeur brute et la valeur numérique : l'une pour l'affichage, l'autre pour les comparaisons et les seuils d'alerte.
# models.py
import sqlite3
from datetime import datetime
from dataclasses import dataclass
from typing import Optional
@dataclass
class CompetitorData:
competitor: str
metric: str
value: str
numeric_value: Optional[float] = None
url: str = ""
scraped_at: str = ""
def __post_init__(self):
if not self.scraped_at:
self.scraped_at = datetime.now().isoformat()
class CompetitorDB:
def __init__(self, path="competitor_data.db"):
self.conn = sqlite3.connect(path)
self._init()
def _init(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS metrics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
competitor TEXT,
metric TEXT,
value TEXT,
numeric_value REAL,
url TEXT,
scraped_at TEXT
)
""")
self.conn.commit()
def save(self, data: CompetitorData):
self.conn.execute(
"""INSERT INTO metrics
(competitor, metric, value, numeric_value, url, scraped_at)
VALUES (?, ?, ?, ?, ?, ?)""",
(data.competitor, data.metric, data.value,
data.numeric_value, data.url, data.scraped_at),
)
self.conn.commit()
def get_history(self, competitor, metric, limit=30):
cursor = self.conn.execute(
"""SELECT value, numeric_value, scraped_at
FROM metrics
WHERE competitor = ? AND metric = ?
ORDER BY scraped_at DESC LIMIT ?""",
(competitor, metric, limit),
)
return cursor.fetchall()
def latest_comparison(self, metric):
cursor = self.conn.execute(
"""SELECT competitor, value, numeric_value, MAX(scraped_at) as latest
FROM metrics WHERE metric = ?
GROUP BY competitor ORDER BY numeric_value""",
(metric,),
)
return cursor.fetchall()
Solveur CAPTCHA
Le solveur n'intervient que si la page renvoie un reCAPTCHA v2 (présence de data-sitekey). L'appel se fait en deux temps sur les endpoints in.php et res.php : vous soumettez le sitekey et l'URL, récupérez un identifiant de tâche, puis interrogez le résultat toutes les 5 secondes jusqu'à obtenir le token g-recaptcha-response. La boucle plafonne à deux minutes pour qu'un site cassé ne bloque pas la collecte. Toute réponse autre que CAPCHA_NOT_READY lève une exception, attrapée par concurrent : un site en échec n'interrompt pas les autres.
# solver.py
import requests
import time
import re
import os
class CaptchaSolver:
def __init__(self):
self.api_key = os.environ["CAPTCHAAI_API_KEY"]
def solve_if_needed(self, session, url, html):
if "data-sitekey" not in html:
return html
match = re.search(r'data-sitekey="([^"]+)"', html)
if not match:
return html
sitekey = match.group(1)
resp = requests.post("https://ocr.captchaai.com/in.php", data={
"key": self.api_key,
"method": "userrecaptcha",
"googlekey": sitekey,
"pageurl": url,
"json": 1,
}, timeout=30)
task_id = resp.json()["request"]
time.sleep(15)
for _ in range(24):
resp = requests.get("https://ocr.captchaai.com/res.php", params={
"key": self.api_key, "action": "get",
"id": task_id, "json": 1,
}, timeout=15)
data = resp.json()
if data.get("status") == 1:
post_resp = session.post(url, data={
"g-recaptcha-response": data["request"],
}, timeout=30)
return post_resp.text
if data["request"] != "CAPCHA_NOT_READY":
raise RuntimeError(data["request"])
time.sleep(5)
raise TimeoutError("CAPTCHA solve timeout")
Le scraper des concurrents
Le scraper enveloppe chaque requête dans _fetch, qui délègue au solveur dès qu'un défi apparaît. Un User-Agent de navigateur récent et une Session réutilisée limitent les déclenchements de CAPTCHA en conservant les cookies entre les requêtes. Trois extracteurs couvrent les cas courants : scrape_pricing lit chaque bloc de plan, scrape_features liste les fonctionnalités et scrape_product_count récupère la taille du catalogue. Chaque concurrent a ses propres sélecteurs CSS, passés en configuration — c'est le seul endroit à ajuster quand un site change de gabarit.
# scraper.py
import requests
import re
from bs4 import BeautifulSoup
from solver import CaptchaSolver
from models import CompetitorData
class CompetitorScraper:
def __init__(self):
self.solver = CaptchaSolver()
self.session = requests.Session()
self.session.headers["User-Agent"] = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 Chrome/125.0.0.0 Safari/537.36"
)
def scrape_pricing(self, competitor_name, url, plan_selector, price_selector):
html = self._fetch(url)
soup = BeautifulSoup(html, "html.parser")
plans = soup.select(plan_selector)
data = []
for plan in plans:
name_el = plan.select_one("h3, h2, .plan-name")
price_el = plan.select_one(price_selector)
if not name_el or not price_el:
continue
price_text = price_el.get_text(strip=True)
match = re.search(r'[\d,.]+', price_text)
numeric = float(match.group().replace(",", "")) if match else None
data.append(CompetitorData(
competitor=competitor_name,
metric=f"price_{name_el.get_text(strip=True).lower().replace(' ', '_')}",
value=price_text,
numeric_value=numeric,
url=url,
))
return data
def scrape_features(self, competitor_name, url, feature_list_selector):
html = self._fetch(url)
soup = BeautifulSoup(html, "html.parser")
features = soup.select(f"{feature_list_selector} li")
return [
CompetitorData(
competitor=competitor_name,
metric="feature",
value=f.get_text(strip=True),
url=url,
)
for f in features if f.get_text(strip=True)
]
def scrape_product_count(self, competitor_name, url, count_selector):
html = self._fetch(url)
soup = BeautifulSoup(html, "html.parser")
el = soup.select_one(count_selector)
if el:
text = el.get_text(strip=True)
match = re.search(r'[\d,]+', text)
if match:
count = int(match.group().replace(",", ""))
return CompetitorData(
competitor=competitor_name,
metric="product_count",
value=text,
numeric_value=count,
url=url,
)
return None
def _fetch(self, url):
resp = self.session.get(url, timeout=20)
return self.solver.solve_if_needed(self.session, url, resp.text)
Générateur de rapports
latest_comparison fournit le comparatif du jour, tandis que get_history rejoue la série d'une métrique pour en dessiner la tendance. Le rapport reste volontairement en texte brut : facile à versionner dans Git, à comparer d'un jour à l'autre et à coller dans un canal d'équipe. Pour un rendu visuel, exportez la table metrics en CSV et branchez-la sur l'outil graphique de votre choix.
# report.py
from models import CompetitorDB
def generate_report(db: CompetitorDB, metrics):
lines = ["=" * 60, "Competitor Analysis Report", "=" * 60, ""]
for metric in metrics:
results = db.latest_comparison(metric)
if not results:
continue
lines.append(f"--- {metric.replace('_', ' ').title()} ---")
for comp, value, numeric, ts in results:
marker = ""
if numeric is not None:
marker = f" (${numeric:,.2f})" if "price" in metric else f" ({numeric:,.0f})"
lines.append(f" {comp}: {value}{marker}")
lines.append("")
return "\n".join(lines)
def generate_trend(db: CompetitorDB, competitor, metric, periods=10):
history = db.get_history(competitor, metric, limit=periods)
if not history:
return f"No data for {competitor} — {metric}"
lines = [f"Trend: {competitor} — {metric}", "-" * 40]
for value, numeric, ts in reversed(history):
date = ts[:10]
lines.append(f" {date}: {value}")
return "\n".join(lines)
Point d'entrée
Le point d'entrée parcourt les concurrents, espace les requêtes et écrit le rapport dans un fichier. La liste COMPETITORS centralise les URL et les sélecteurs : ajouter un concurrent revient à y ajouter une entrée. Le time.sleep entre chaque site espace les requêtes et réduit la charge visible côté serveur distant.
# main.py
import time
from models import CompetitorDB
from scraper import CompetitorScraper
from report import generate_report
COMPETITORS = [
{
"name": "Competitor A",
"pricing_url": "https://competitor-a.example.com/pricing",
"plan_selector": ".pricing-plan",
"price_selector": ".price",
},
{
"name": "Competitor B",
"pricing_url": "https://competitor-b.example.com/pricing",
"plan_selector": ".plan-card",
"price_selector": ".plan-price",
},
]
def main():
db = CompetitorDB()
scraper = CompetitorScraper()
for comp in COMPETITORS:
print(f"Scraping {comp['name']}...")
try:
pricing = scraper.scrape_pricing(
comp["name"], comp["pricing_url"],
comp["plan_selector"], comp["price_selector"],
)
for p in pricing:
db.save(p)
print(f" {p.metric}: {p.value}")
except Exception as e:
print(f" Error: {e}")
time.sleep(5)
# Generate report
metrics = ["price_basic", "price_pro", "price_enterprise", "product_count"]
report = generate_report(db, metrics)
print(report)
with open("competitor_report.txt", "w") as f:
f.write(report)
if __name__ == "__main__":
main()
Planifier et fiabiliser la collecte
Un tableau de bord de veille n'a de valeur que s'il tourne régulièrement et sans surveillance.
Automatiser l'exécution
Planifiez main.py une fois par jour avec cron sous Linux, ou avec une tâche planifiée sur un petit serveur OVHcloud ou Scaleway — une machine à quelques dollars par mois suffit, la charge est minime. Une collecte quotidienne sur quelques concurrents reste largement dans les limites du plan BASIC ($15/mois, 5 threads) ; passez à STANDARD ($30/mois, 15 threads) si vous parallélisez les scrapers.
Garde-fous à prévoir
- Nouvelles tentatives. Enveloppez chaque collecte dans un backoff exponentiel : une page qui échoue une fois réussit souvent au second essai.
- Alertes. Comparez la valeur du jour à la dernière valeur stockée et prévenez votre équipe (Slack, e-mail) quand l'écart dépasse un seuil.
- Proxys. Sur des volumes plus élevés, un proxy résidentiel réduit les blocages ; réservez-le aux sites qui le justifient vraiment.
- Journalisation. Conservez les logs de chaque exécution pour distinguer un sélecteur cassé d'un vrai changement de prix.
Dépannage
| Problème | Cause | Correctif |
|---|---|---|
| Prix non extraits | Sélecteur CSS inadapté | Inspectez le HTML de la page et ajustez les sélecteurs par concurrent |
| Aucune tendance disponible | Première exécution | Les données s'accumulent ; lancez la collecte chaque jour pour voir les tendances |
| CAPTCHA sur la page de tarification | Détection du trafic automatisé | Espacez les requêtes et réutilisez les cookies de session |
| Le rapport affiche des valeurs périmées | Même entrée réinsérée | Utilisez latest_comparison, qui regroupe par date la plus récente |
FAQ
Cette collecte est-elle conforme au RGPD ?
Une page de tarifs publique ne contient en général pas de données personnelles, donc peu de risques RGPD. Dès qu'elle touche des données nominatives, minimisez ce que vous conservez et documentez la finalité.
Quel plan CaptchaAI choisir pour surveiller plusieurs concurrents ?
Pour une collecte quotidienne sur quelques sites, le plan BASIC ($15/mois, 5 threads) suffit. Passez à STANDARD ($30/mois, 15 threads) si vous parallélisez les scrapers.
À quelle fréquence lancer le scraping sans être bloqué ?
Une exécution par jour et par concurrent suffit pour la plupart des besoins de veille. Espacez les requêtes et évitez les rafales, qui déclenchent la détection.
Que faire quand un concurrent change la structure de sa page ?
Les sélecteurs CSS finissent par casser. Isolez-les par concurrent dans la liste COMPETITORS, surveillez les extractions vides comme signal d'alerte et corrigez uniquement le sélecteur concerné plutôt que de réécrire tout le scraper.
Puis-je suivre d'autres indicateurs que le prix ?
Oui. scrape_features récupère les listes de fonctionnalités et scrape_product_count la taille du catalogue ; ajoutez vos propres extracteurs pour tout indicateur exposé publiquement, comme les délais de livraison ou la disponibilité.
Guides connexes
Suivez vos concurrents à grande échelle — commencez avec CaptchaAI.
