Collecte de données de recherche sur les réseaux sociaux avec gestion CAPTCHA

Un pipeline de veille sociale ne casse presque jamais sur le parsing HTML : il casse à la troisième requête, quand la plateforme remplace la page par un défi CAPTCHA. La parade tient en quatre gestes : repérer le défi dans la réponse, envoyer le sitekey à l'API CaptchaAI, réinjecter le token, puis ralentir assez pour ne pas le rappeler aussitôt.

Ce que vous collectez, et ce que vous n'enregistrez pas

Tranchez le périmètre avant la première requête : contenus publics uniquement, et seulement les champs utiles à votre analyse. La minimisation RGPD vaut aussi pour un corpus de recherche. Un laboratoire de sciences sociales à Lyon qui étudie la circulation d'un hashtag n'a que faire des noms d'utilisateur : un identifiant pseudonymisé et un horodatage suffisent. Une agence de veille à Bruxelles garde, elle, l'URL publique du message pour le citer.

Défis et cadence par plateforme

Le type de défi et le rythme tenable changent d'une plateforme à l'autre : plus l'intervalle est respecté, moins vous déclenchez de défis facturés.

Les intervalles ci-dessous reposent sur des mesures observées et des retours d'utilisateurs ; ils varient selon l'environnement, le volume et l'heure de la journée.

Un collecteur Python qui repère le défi et le résout

La classe ci-dessous garde un User-Agent mobile cohérent, cherche dans chaque réponse un marqueur de défi (data-sitekey, cf-turnstile, challenge-platform), puis rejoue la requête avec le token. Le champ dépend du type : g-recaptcha-response pour reCAPTCHA, cf-turnstile-response pour Turnstile. Le polling sur res.php attend que le statut quitte CAPCHA_NOT_READY.

import requests
import time
import re

CAPTCHAAI_KEY = "YOUR_API_KEY"
CAPTCHAAI_URL = "https://ocr.captchaai.com"


def solve_captcha(method, sitekey, pageurl, **kwargs):
    data = {
        "key": CAPTCHAAI_KEY,
        "method": method,
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": 1,
    }
    data.update(kwargs)
    resp = requests.post(f"{CAPTCHAAI_URL}/in.php", data=data)
    task_id = resp.json()["request"]

    for _ in range(60):
        time.sleep(5)
        result = requests.get(f"{CAPTCHAAI_URL}/res.php", params={
            "key": CAPTCHAAI_KEY, "action": "get",
            "id": task_id, "json": 1,
        })
        r = result.json()
        if r["request"] != "CAPCHA_NOT_READY":
            return r["request"]
    raise TimeoutError("Solve timeout")


class SocialMediaResearcher:
    def __init__(self, proxy=None):
        self.session = requests.Session()
        if proxy:
            self.session.proxies = {"http": proxy, "https": proxy}
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 17_5 like Mac OS X) "
            "AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.5 "
            "Mobile/15E148 Safari/604.1",
            "Accept-Language": "en-US,en;q=0.9",
        })

    def authenticate(self, login_url, credentials, sitekey):
        """Login with CAPTCHA handling."""
        # Load login page
        self.session.get(login_url)

        # Solve CAPTCHA
        token = solve_captcha("userrecaptcha", sitekey, login_url)

        # Submit login
        resp = self.session.post(login_url, data={
            **credentials,
            "g-recaptcha-response": token,
        })
        return resp.status_code == 200

    def collect_profiles(self, profile_urls):
        """Collect public profile data with CAPTCHA handling."""
        profiles = []

        for url in profile_urls:
            try:
                resp = self.session.get(url, timeout=30)

                # Handle CAPTCHA if triggered
                if self._has_captcha(resp.text):
                    resp = self._handle_captcha(resp.text, url)

                profiles.append({
                    "url": url,
                    "data": self._parse_profile(resp.text),
                    "status": "success",
                })
                time.sleep(5)  # Slow down between profiles

            except Exception as e:
                profiles.append({
                    "url": url,
                    "error": str(e),
                    "status": "failed",
                })

        return profiles

    def _has_captcha(self, html):
        return any(tag in html.lower() for tag in [
            'data-sitekey', 'g-recaptcha', 'cf-turnstile',
            'challenge-platform', 'captcha',
        ])

    def _handle_captcha(self, html, url):
        match = re.search(r'data-sitekey="([^"]+)"', html)
        if not match:
            return self.session.get(url)

        sitekey = match.group(1)

        if 'cf-turnstile' in html:
            token = solve_captcha("turnstile", sitekey, url)
            return self.session.post(url, data={"cf-turnstile-response": token})
        else:
            token = solve_captcha("userrecaptcha", sitekey, url)
            return self.session.post(url, data={"g-recaptcha-response": token})

    def _parse_profile(self, html):
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, "html.parser")
        return {
            "name": self._safe_text(soup, "h1, .profile-name"),
            "bio": self._safe_text(soup, ".bio, .profile-bio"),
            "followers": self._safe_text(soup, "[data-followers], .followers"),
            "posts": self._safe_text(soup, "[data-posts], .posts-count"),
        }

    def _safe_text(self, soup, selector):
        el = soup.select_one(selector)
        return el.get_text(strip=True) if el else ""

Suivre un hashtag page par page

Même boucle, appliquée à la pagination d'un flux thématique. Le time.sleep(5) entre deux pages n'est pas décoratif : c'est lui qui évite un défi à chaque appel.

def research_hashtag(hashtag, platform_url, pages=5):
    """Collect posts for a specific hashtag."""
    researcher = SocialMediaResearcher(
        proxy="http://user:pass@mobile.proxy.com:5000"
    )

    all_posts = []
    for page in range(pages):
        url = f"{platform_url}/explore/tags/{hashtag}?page={page}"
        resp = researcher.session.get(url, timeout=30)

        if researcher._has_captcha(resp.text):
            resp = researcher._handle_captcha(resp.text, url)

        from bs4 import BeautifulSoup
        soup = BeautifulSoup(resp.text, "html.parser")
        posts = soup.select(".post-item, article")
        for post in posts:
            all_posts.append({
                "text": post.get_text(strip=True)[:500],
                "hashtag": hashtag,
                "page": page,
            })

        time.sleep(5)

    return all_posts

Veille de marque : le balayage quotidien

Un suivi de mentions devient un job planifié : mots-clés, plateformes, rapport JSON horodaté. Les erreurs restent dans le rapport plutôt que de faire tomber le balayage.

import json
from datetime import datetime


class BrandMonitor:
    def __init__(self, brand_name, keywords, proxy=None):
        self.brand = brand_name
        self.keywords = keywords
        self.researcher = SocialMediaResearcher(proxy=proxy)

    def daily_scan(self, platform_urls):
        """Run daily brand mention scan across platforms."""
        report = {
            "brand": self.brand,
            "date": datetime.now().isoformat(),
            "platforms": {},
        }

        for name, url in platform_urls.items():
            mentions = []
            for keyword in self.keywords:
                search_url = f"{url}/search?q={keyword}"
                try:
                    resp = self.researcher.session.get(search_url, timeout=30)

                    if self.researcher._has_captcha(resp.text):
                        resp = self.researcher._handle_captcha(
                            resp.text, search_url,
                        )

                    from bs4 import BeautifulSoup
                    soup = BeautifulSoup(resp.text, "html.parser")
                    results = soup.select(".search-result, .post")
                    mentions.append({
                        "keyword": keyword,
                        "count": len(results),
                    })
                    time.sleep(5)
                except Exception as e:
                    mentions.append({
                        "keyword": keyword,
                        "error": str(e),
                    })

            report["platforms"][name] = mentions

        return report


# Usage
monitor = BrandMonitor(
    brand_name="CaptchaAI",
    keywords=["captchaai", "captcha ai", "captcha solver"],
    proxy="http://user:pass@mobile.proxy.com:5000",
)
report = monitor.daily_scan({
    "twitter": "https://twitter-alternative.example.com",
    "reddit": "https://www.reddit.com",
})
print(json.dumps(report, indent=2))

Quel proxy pour quelle plateforme

Le proxy pèse autant que la cadence. Instagram et TikTok attendent du trafic mobile (4G) ; Facebook et Twitter/X filtrent durement le datacenter et réclament du résidentiel ; LinkedIn s'attend à des IP d'entreprise, donc du résidentiel FAI ; Reddit limite le débit par IP, donc du résidentiel rotatif.

Dimensionner les threads et le budget

Un thread correspond à une résolution en cours : à cette cadence, un collecteur attend plus qu'il ne résout. Quelques centaines de profils par jour tiennent sur BASIC ($15/mois, 5 threads) ; un suivi de marque multiplateforme respire mieux sur STANDARD ($30/mois, 15 threads), une collecte continue sur ADVANCE ($90/mois, 50 threads). Dimensionnez sur le pic de parallélisme, pas sur le volume mensuel.

Dépannage

FAQ

Quels types de CAPTCHA sont pris en charge ici ?

reCAPTCHA v2 et v3, Turnstile, Cloudflare Challenge, GeeTest v3 et les CAPTCHA image ou texte. hCaptcha et FunCaptcha (Arkose Labs) ne sont pas pris en charge ; GeeTest v4 est à venir. CaptchaFox, Friendly Captcha et Lemin restent en bêta.

Combien de threads faut-il pour un balayage quotidien ?

Comptez les collecteurs simultanés, pas les profils. Cinq threads (BASIC, $15/mois) couvrent trois à quatre workers espacés ; au-delà de dix en parallèle, passez à STANDARD ($30/mois, 15 threads).

Le token est accepté mais la page reste bloquée : que vérifier ?

Trois causes reviennent : le token part d'une autre session que celle qui a chargé la page, le pageurl transmis diffère de l'URL réelle, ou le token a expiré. Renvoyez-le immédiatement, depuis la même session.

Le RGPD s'applique-t-il à des profils publics ?

Oui : « public » ne signifie pas « hors du champ du RGPD » dès qu'une personne reste identifiable. Limitez les champs collectés, pseudonymisez et fixez une durée de conservation.

Guides connexes

• La rotation de proxys sous CAPTCHA
• Conserver l'état de session entre workers
• Isoler les profils de navigateur avec CaptchaAI

Fiabilisez votre collecte de données sociales : créez votre clé API CaptchaAI et laissez la résolution du défi se faire dans le flux.