Surveillance de la chaîne d'approvisionnement avec la gestion des CAPTCHA

Une rupture fournisseur coûte cher surtout parce que vous l'apprenez trois jours trop tard. La cause technique est presque toujours la même : le collecteur s'arrête sur un défi CAPTCHA en ouvrant la page d'inventaire, et personne ne relit les logs avant lundi matin. CaptchaAI résout les types qui protègent ces portails — reCAPTCHA v2, reCAPTCHA v3, Cloudflare Turnstile, CAPTCHA image/OCR — et renvoie un token que votre script injecte dans le formulaire.

Voici comment câbler cette résolution dans un pipeline de surveillance multifournisseurs en Python : où les défis apparaissent, quel code écrire, comment dimensionner les threads et quoi vérifier quand une source décroche.

Où les CAPTCHA interrompent la collecte fournisseurs

Les défis ne sont pas répartis au hasard : chaque famille de source a son type dominant.

Trois familles couvrent l'essentiel, et ce sont celles que gère le code ci-dessous. Si un portail affiche hCaptcha ou FunCaptcha, ces types ne sont pas pris en charge par CaptchaAI : prévoyez un flux EDI ou un export négocié avec le fournisseur.

Quelles sources surveiller, et à quelle cadence

Une cadence trop agressive fait apparaître davantage de défis sans apporter d'information nouvelle. La règle : la cadence suit le délai de réapprovisionnement, pas la curiosité.

Le moniteur multifournisseurs en Python

Le script regroupe les trois résolveurs derrière une seule classe. Chaque fournisseur est décrit par une entrée de configuration — URL, type de défi, sitekey éventuel — et le moniteur choisit le chemin de résolution à la volée. Les erreurs sont capturées source par source : un portail en panne n'interrompt pas la tournée.

import requests
import time
import re
import json
import base64
from datetime import datetime

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


def solve_recaptcha(sitekey, pageurl):
    resp = requests.post(f"{CAPTCHAAI_URL}/in.php", data={
        "key": CAPTCHAAI_KEY, "method": "userrecaptcha",
        "googlekey": sitekey, "pageurl": pageurl, "json": 1,
    })
    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,
        })
        data = result.json()
        if data["request"] != "CAPCHA_NOT_READY":
            return data["request"]
    raise TimeoutError("Timeout")


def solve_turnstile(sitekey, pageurl):
    resp = requests.post(f"{CAPTCHAAI_URL}/in.php", data={
        "key": CAPTCHAAI_KEY, "method": "turnstile",
        "sitekey": sitekey, "pageurl": pageurl, "json": 1,
    })
    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,
        })
        data = result.json()
        if data["request"] != "CAPCHA_NOT_READY":
            return data["request"]
    raise TimeoutError("Timeout")


def solve_image(image_bytes):
    img_b64 = base64.b64encode(image_bytes).decode()
    resp = requests.post(f"{CAPTCHAAI_URL}/in.php", data={
        "key": CAPTCHAAI_KEY, "method": "base64",
        "body": img_b64, "json": 1,
    })
    task_id = resp.json()["request"]
    for _ in range(20):
        time.sleep(3)
        result = requests.get(f"{CAPTCHAAI_URL}/res.php", params={
            "key": CAPTCHAAI_KEY, "action": "get",
            "id": task_id, "json": 1,
        })
        data = result.json()
        if data["request"] != "CAPCHA_NOT_READY":
            return data["request"]
    raise TimeoutError("Timeout")


class SupplyChainMonitor:
    def __init__(self, suppliers, proxy=None):
        self.suppliers = suppliers
        self.session = requests.Session()
        if proxy:
            self.session.proxies = {"http": proxy, "https": proxy}
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 Chrome/126.0.0.0 Safari/537.36",
        })

    def check_all(self):
        """Check inventory and pricing across all suppliers."""
        report = {
            "timestamp": datetime.now().isoformat(),
            "suppliers": {},
        }

        for supplier in self.suppliers:
            try:
                data = self._check_supplier(supplier)
                report["suppliers"][supplier["name"]] = {
                    "status": "success",
                    "data": data,
                }
            except Exception as e:
                report["suppliers"][supplier["name"]] = {
                    "status": "error",
                    "error": str(e),
                }
            time.sleep(3)

        return report

    def _check_supplier(self, supplier):
        url = supplier["url"]
        resp = self.session.get(url, timeout=30)

        # Handle CAPTCHA based on type
        captcha_type = supplier.get("captcha_type")
        if captcha_type and self._has_captcha(resp.text):
            resp = self._solve_captcha(resp, url, supplier)

        from bs4 import BeautifulSoup
        soup = BeautifulSoup(resp.text, "html.parser")

        return {
            "products": self._extract_inventory(soup),
            "last_updated": self._extract_date(soup),
        }

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

    def _solve_captcha(self, resp, url, supplier):
        captcha_type = supplier.get("captcha_type", "recaptcha")
        sitekey = supplier.get("sitekey", "")

        if not sitekey:
            match = re.search(r'data-sitekey="([^"]+)"', resp.text)
            sitekey = match.group(1) if match else ""

        if captcha_type == "turnstile":
            token = solve_turnstile(sitekey, url)
            return self.session.post(url, data={"cf-turnstile-response": token})
        elif captcha_type == "image":
            match = re.search(r'src="(/captcha[^"]+)"', resp.text)
            if match:
                img_resp = self.session.get(url.rstrip("/") + match.group(1))
                answer = solve_image(img_resp.content)
                return self.session.post(url, data={"captcha": answer})
        else:
            token = solve_recaptcha(sitekey, url)
            return self.session.post(url, data={"g-recaptcha-response": token})

        return resp

    def _extract_inventory(self, soup):
        items = []
        for row in soup.select("table.inventory tr, .product-row"):
            cols = row.select("td, .col")
            if len(cols) >= 3:
                items.append({
                    "sku": cols[0].get_text(strip=True),
                    "stock": cols[1].get_text(strip=True),
                    "price": cols[2].get_text(strip=True),
                })
        return items

    def _extract_date(self, soup):
        date_el = soup.select_one(".last-updated, .update-time")
        return date_el.get_text(strip=True) if date_el else ""


# Configure suppliers
suppliers = [
    {
        "name": "Supplier A",
        "url": "https://supplier-a.example.com/inventory",
        "captcha_type": "recaptcha",
        "sitekey": "6Lc_xxxxxxx",
    },
    {
        "name": "Carrier B",
        "url": "https://carrier-b.example.com/rates",
        "captcha_type": "turnstile",
        "sitekey": "0x4AAAAAAA_xxx",
    },
    {
        "name": "Manufacturer C",
        "url": "https://manufacturer-c.example.com/catalog",
        "captcha_type": "image",
    },
]

monitor = SupplyChainMonitor(
    suppliers=suppliers,
    proxy="http://user:pass@residential.proxy.com:5000",
)
report = monitor.check_all()
print(json.dumps(report, indent=2))

Deux détails comptent : _has_captcha inspecte le HTML retourné plutôt que de supposer un défi permanent — beaucoup de portails ne challengent qu'une visite sur cinq — et le délai de 3 secondes entre deux fournisseurs lisse la charge.

Suivre les tarifs des transporteurs

Les grilles tarifaires sont la deuxième source la plus utile après l'inventaire, et la plus souvent protégée par Cloudflare Turnstile. Le token se glisse dans le champ cf-turnstile-response du formulaire.

class ShippingRateTracker:
    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 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 Chrome/126.0.0.0 Safari/537.36",
        })

    def get_rates(self, carrier_url, origin, destination, weight):
        """Fetch shipping rates, handling Turnstile CAPTCHA."""
        resp = self.session.get(carrier_url, timeout=30)

        sitekey_match = re.search(r'data-sitekey="([^"]+)"', resp.text)
        if sitekey_match:
            token = solve_turnstile(sitekey_match.group(1), carrier_url)
            resp = self.session.post(carrier_url, data={
                "origin": origin,
                "destination": destination,
                "weight": weight,
                "cf-turnstile-response": token,
            })

        if resp.status_code == 200:
            return resp.json().get("rates", [])
        return []

Turnstile se résout généralement en moins de 10 secondes, avec un taux de réussite élevé sur les types pris en charge. Prévoyez malgré tout un retry : un token expiré ou réutilisé est refusé par le portail, pas par l'API.

Déclencher des alertes sur les ruptures de stock

La boucle ci-dessous compare chaque relevé au précédent et ne signale que le franchissement d'un seuil, au lieu de notifier chaque variation d'une unité.

def monitor_with_alerts(monitor, alert_thresholds, check_interval=3600):
    """Continuously monitor and alert on inventory changes."""
    previous_data = {}

    while True:
        report = monitor.check_all()

        for supplier, info in report["suppliers"].items():
            if info["status"] != "success":
                continue

            for product in info["data"].get("products", []):
                sku = product["sku"]
                stock = product.get("stock", "")

                # Parse stock level
                try:
                    stock_qty = int(re.sub(r'\D', '', stock))
                except ValueError:
                    continue

                key = f"{supplier}:{sku}"
                prev_qty = previous_data.get(key, stock_qty)

                threshold = alert_thresholds.get(sku, 10)
                if stock_qty < threshold and prev_qty >= threshold:
                    print(f"ALERT: {supplier} - {sku} dropped to {stock_qty}")

                previous_data[key] = stock_qty

        time.sleep(check_interval)

Branchez print sur ce que votre équipe lit : Slack, Teams ou une file d'attente.

Dimensionner les threads et le budget

Exemple : une PME industrielle de Lille surveille 40 portails fournisseurs européens et 6 transporteurs depuis un worker déployé chez OVHcloud ou Scaleway. Chaque tournée horaire déclenche une vingtaine de défis.

La facturation CaptchaAI se fait au thread simultané, pas à la résolution : chaque plan inclut un nombre de résolutions illimité par thread. Un thread correspond à un défi en cours ; dès qu'il se termine, il reprend le suivant. Pour ce profil, ADVANCE ($90/mois, 50 threads) laisse une marge confortable, tandis que STANDARD ($30/mois, 15 threads) suffit à une tournée quotidienne sur une trentaine de sources. BASIC ($15/mois, 5 threads) sert à valider l'intégration. La facturation reste en dollars US.

Placez le worker dans la zone de vos fournisseurs : une instance à Paris (eu-west-3) réduit la latence.

Périmètre de collecte et RGPD

Les portails fournisseurs contiennent des données contractuelles, mais aussi des contacts nominatifs. Deux réflexes avant la mise en production :

• Ne collectez que les champs utiles — références, quantités, prix, délais. Les noms et e-mails ne font qu'alourdir vos obligations RGPD.
• Restez dans le périmètre autorisé : identifiants remis par le fournisseur, conditions d'utilisation respectées, cadence documentée.

Fixez enfin une durée de conservation : 90 jours d'historique suffisent à repérer une tendance.

Dépannage

FAQ

Combien de threads faut-il pour surveiller une cinquantaine de portails ?

Comptez les défis simultanés, pas les fournisseurs. Une tournée séquentielle sur 50 sources n'en déclenche que quelques-uns en parallèle : STANDARD ($30/mois, 15 threads) suffit, et ADVANCE ($90/mois, 50 threads) devient utile quand plusieurs tournées se chevauchent.

CaptchaAI prend-il en charge hCaptcha sur les portails logistiques ?

Non — hCaptcha n'est pas pris en charge, pas plus que FunCaptcha (Arkose Labs). GeeTest v4 est annoncé comme à venir. Pour ces sources, négociez un export périodique avec le fournisseur.

Pourquoi mon token Turnstile est-il refusé alors que l'API l'a renvoyé ?

Parce qu'il a été soumis trop tard ou depuis une autre session. Demandez la résolution juste avant l'envoi du formulaire, dans la même session requests, et ne le mettez jamais en cache.

Faut-il un proxy résidentiel pour surveiller des fournisseurs ?

Pas systématiquement : un accès authentifié à cadence raisonnable passe souvent depuis une IP datacenter. Les proxys résidentiels servent surtout aux tournées fréquentes ou aux portails qui affichent des tarifs par pays.

Guides connexes

• Faire tourner vos proxys résidentiels sans interrompre la résolution
• Sessions collantes ou rotatives : que choisir pour un portail authentifié
• Conserver l'état de session entre workers distribués

Gardez vos stocks et vos délais sous les yeux : récupérez votre clé API CaptchaAI et relancez vos tournées de collecte dès aujourd'hui.