Monitoramento da cadeia de suprimentos: como resolver CAPTCHA nos portais

O relatório de estoque que quebra às 6h quase nunca quebra por causa do seu código. Ele quebra porque um portal passou a exibir um desafio de verificação, o script recebeu HTML de CAPTCHA em vez da tabela de SKUs, e ninguém notou até a reunião de compras. A saída é tratar o CAPTCHA como etapa previsível da coleta: detectar, resolver pela API da CaptchaAI, reenviar o formulário e seguir adiante.

Este guia monta esse monitor em Python e fecha com a política de fallback que evita que uma fonte derrube o relatório inteiro. Colete apenas de portais que você tem autorização para consultar.

Onde o CAPTCHA trava os dados da cadeia de suprimentos

Antes de codificar, mapeie onde o desafio aparece: o tipo define qual method você chama na API.

Todos estão na cobertura em GA da CaptchaAI, junto com Cloudflare Challenge e GeeTest v3. CaptchaFox, Friendly Captcha e Lemin estão em beta; hCaptcha e FunCaptcha não são suportados, e o GeeTest v4 consta apenas como "em breve".

Um caso típico: uma distribuidora de autopeças em Joinville precisa, todo dia útil às 7h, do saldo de 400 SKUs, do frete de duas transportadoras e do status de liberação alfandegária — cinco fontes e três tipos de CAPTCHA numa janela de 20 minutos.

O monitor multifornecedor em Python

São três funções de resolução (uma por tipo) e uma classe que percorre os fornecedores. Cada função envia a tarefa para in.php, guarda o ID e consulta res.php até sair de CAPCHA_NOT_READY. Troque YOUR_API_KEY pela chave do seu painel.

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:[email protected]:5000",
)
report = monitor.check_all()
print(json.dumps(report, indent=2))

Três detalhes importam: o _has_captcha evita gastar resolução em página que já veio correta; o try/except por fornecedor transforma uma fonte quebrada em linha de erro, não em exceção que aborta as outras; e o time.sleep(3) espaça as requisições, reduzindo a frequência do desafio.

Como a cobrança é por thread simultânea com resoluções ilimitadas, o que importa é quantos desafios ficam em voo ao mesmo tempo. O monitor sequencial acima cabe no BASIC (US$ 15/mês, 5 threads); paralelizando 30 fornecedores, o STANDARD (US$ 30/mês, 15 threads) passa a fazer sentido.

Consulta automatizada de tarifas de frete

A tarifa de frete é a fonte mais volátil do painel: muda por rota, peso e dia, e o formulário quase sempre está atrás de um Turnstile. O rastreador abaixo busca a página, extrai a sitekey, resolve o desafio e reenvia o formulário com o campo cf-turnstile-response.

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 []

Alertas de ruptura de estoque

Coletar sem alertar é só encher banco. O laço abaixo compara cada leitura com a anterior e avisa quando um SKU cruza o limite para baixo — é a transição que importa, senão o mesmo item em falta alerta a cada hora. Em produção, troque o print pelo canal do time de compras e persista o previous_data.

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)

Política de fallback: o que fazer quando uma fonte falha

Um monitor de suprimentos convive com falhas parciais. O que separa um painel confiável de um ignorado é o tratamento do dado que não veio.

• Classifique as fontes. Fornecedor crítico não é catálogo consultado uma vez por semana; falha no segundo caso não pode apagar o relatório.
• Carimbe a hora da coleta e exiba esse carimbo — dado velho apresentado como atual é pior que dado ausente. Falha isolada é retentativa; três ciclos seguidos sem ler um fornecedor crítico é incidente.
• Trate CAPTCHA recorrente como sinal de negócio. Desafio em toda requisição costuma indicar mudança na política de acesso. O caminho é comercial — API ou EDI —, não aumentar a frequência da coleta.

Solução de problemas

Perguntas frequentes

Qual plano atende um monitor de 30 fornecedores?

Depende de quantos desafios ficam simultâneos, já que a cobrança é por thread com resoluções ilimitadas. Sequencial, cabe no BASIC (US$ 15/mês, 5 threads); os 30 em paralelo pedem o STANDARD (US$ 30/mês, 15 threads).

E se o portal do fornecedor usar hCaptcha?

Aí a automação não é o caminho: hCaptcha e FunCaptcha não são suportados, e o GeeTest v4 consta apenas como "em breve". Resolva por acordo comercial — API do fornecedor, EDI ou exportação periódica.

Preciso de um navegador completo para essa coleta?

Quase nunca. Os exemplos usam só requests: lê o HTML, extrai a sitekey, resolve pela API e reenvia o token. Navegador só se justifica quando a tabela é montada via JavaScript após o login.

Existe questão de LGPD nesse monitoramento?

Catálogo, estoque e tarifa são dados comerciais, não pessoais. A atenção aparece se a coleta captura nomes de contatos ou dados de motoristas — nesse ponto valem as obrigações de finalidade e retenção.

Leitura relacionada

• Como funciona a rotação de proxy na resolução de CAPTCHA
• Quando manter a sessão fixa e quando fazer rotação
• Manter a sessão do navegador viva entre as coletas

Coloque o painel de suprimentos para rodar sem intervenção manual: crie sua chave de API da CaptchaAI e resolva o primeiro desafio do portal ainda hoje.