Use Cases

Мониторинг цепочки поставок с обработкой CAPTCHA

Python reCAPTCHA v2 Cloudflare Turnstile Image OCR Мониторинг
К Русская команда CaptchaAI
Опубликовано на Jul 21, 2026
Мониторинг цепочки поставок с обработкой CAPTCHA
Избранное изображение недоступно

Монитор цепочки поставок ломается не на разборе HTML, а на проверке доступа: портал поставщика отдаёт reCAPTCHA v2 вместо таблицы остатков, сайт перевозчика закрывает калькулятор тарифов Cloudflare Turnstile, а каталог производителя показывает капчу-картинку. Вывод из этого один — решение CAPTCHA должно жить внутри того же цикла опроса, что и HTTP-запрос, а не разбираться руками по утрам. CaptchaAI закрывает эту часть через API: вы отправляете задачу на in.php, забираете токен с res.php и продолжаете обычный сценарий сбора данных.

Ниже — карта источников, рабочий монитор на Python, расчёт потоков под частоту опроса и разбор типовых сбоев.

Сколько потоков нужно монитору поставщиков

Тариф CaptchaAI считается по числу одновременных задач, то есть потоков, а не по количеству решений. Внутри тарифа число решений в месяц не ограничено, поэтому планировать нужно пик параллелизма, а не суммарный объём.

Практический ориентир для отдела закупок, который ведёт 60 поставщиков с ежедневным срезом и 8 перевозчиков с ежечасным опросом: за сутки набегает около 250 проверок, но одновременно в работе висит не больше 5–10 задач, если опрос разнесён по времени. На старте этого хватает тарифа BASIC ($15/мес, 5 потоков), а когда добавляются ночные окна и повторы — STANDARD ($30/мес, 15 потоков). Расширение до сотен источников с параллельными воркерами — уже ADVANCE ($90/мес, 50 потоков).

Отсюда практическое правило: сначала растяните расписание, потом покупайте потоки. Монитор, который бьёт по всем поставщикам ровно в 09:00, требует втрое больше потоков, чем тот же монитор с окном сбора в четыре часа — при одинаковом наборе данных. Для команд, которые считают бюджет в волатильной локальной валюте, предсказуемая месячная цена в USD за фиксированное число потоков планируется проще, чем оплата за каждое решение.

Где в цепочке поставок стоит CAPTCHA

Тип источника Тип CAPTCHA Какие данные Частота опроса
Порталы поставщиков reCAPTCHA v2 Остатки, цены, сроки поставки Раз в сутки
Сайты перевозчиков Cloudflare Turnstile Трекинг, тарифы, расчётная дата доставки Раз в час
Каталоги производителей Капча-картинка (OCR) Характеристики, минимальная партия Раз в неделю
Таможенные порталы reCAPTCHA v2 Ставки пошлин, коды ТН ВЭД Раз в сутки
Портовые администрации Капча-картинка (OCR) Расписание судов, загрузка терминала Каждые 6 часов
Товарные биржи reCAPTCHA v3 Спот-цены, фьючерсы Близко к реальному времени

Три семейства покрывают почти весь список: reCAPTCHA v2/v3, Cloudflare Turnstile и обычная капча-картинка. Все три поддерживаются, поэтому монитору хватает одной интеграции с тремя ветками по типу проверки. Отдельно стоит проверить исключения: hCaptcha и FunCaptcha CaptchaAI не решает, а GeeTest v4 значится как «скоро». Если такой источник попал в список, заложите для него ручной или альтернативный маршрут ещё на этапе проектирования.

Монитор нескольких поставщиков на Python

Класс ниже обходит список поставщиков, определяет наличие проверки по разметке страницы и вызывает нужный решатель. Каждый источник описан словарём с URL, типом капчи и, если он известен заранее, значением sitekey — это экономит один лишний запрос.

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))

Обратите внимание на три детали. Первая: _has_captcha смотрит на разметку, а не на код ответа: порталы поставщиков часто отдают проверку со статусом 200. Вторая: пауза в 3 с между поставщиками равномерно размазывает нагрузку и снижает вероятность того, что портал начнёт показывать проверку на каждый заход. Третья: ошибка одного источника не рушит отчёт целиком — она попадает в поле status и разбирается отдельно.

Отслеживание тарифов на доставку

Калькуляторы перевозчиков — самый частый случай Turnstile в этой связке. Форма уходит методом POST, и токен нужно положить в поле 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 []

Тарифы меняются чаще остатков, поэтому перевозчиков имеет смысл опрашивать раз в час, а не раз в сутки. Когда маршрутов много, разложите их в очередь и держите параллелизм на уровне купленных потоков — иначе задачи начнут ждать свободного потока, и час опроса растянется.

Оповещения об изменении остатков

Мониторинг без порогов превращается в поток бесполезных отчётов. Здесь предыдущий снимок сравнивается с текущим, и сигнал срабатывает только на переходе через порог — когда остаток падает ниже критичной отметки.

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)

Порог задаётся по SKU: для дефицитной позиции 10 штук — уже повод для реакции, для расходников — шум. Хранить previous_data лучше во внешнем хранилище, чтобы перезапуск процесса не обнулял историю сравнений.

Политика отказов по поставщикам

  • Разделите поставщиков на критичных и опциональных: сбой одного опционального источника не должен обнулять весь отчёт мониторинга.
  • Зафиксируйте заранее, когда устаревшие данные приемлемы, а когда пропуск поставщика должен заводить инцидент.
  • Считайте постоянное появление CAPTCHA у конкретного поставщика сигналом риска в закупках, а не только инженерной ошибкой: обычно за этим стоит слишком частый опрос или смена защиты на портале.
  • Собирайте только те данные, которые вы вправе обрабатывать. Если в срез попадают контакты сотрудников поставщика, действуют требования 152-ФЗ «О персональных данных», а для трансграничных источников — соответствующая проверка на стороне юристов. Это ремарка о вашей собственной осмотрительности, а не о свойствах CaptchaAI, и не юридическая консультация.

Разбор типовых сбоев

Симптом Причина Что сделать
Селекторы перестали находить таблицу Редизайн портала Обновить CSS-селекторы, добавить тест на структуру
CAPTCHA появляется при каждом заходе Слишком частый опрос Увеличить интервал, развести источники по расписанию
Сессия истекает в середине проверки Тайм-аут портала Держать сессию закреплённой, сокращать время прохода
Не приходят данные по тарифам Требуется авторизация Добавить шаг входа перед запросом калькулятора
Цены отличаются от ожидаемых Ценообразование по региону Сопоставить регион выхода прокси с целевым рынком
Задача висит в CAPCHA_NOT_READY Слишком частый опрос res.php или неверный method Проверить метод отправки, опрашивать раз в 3–5 секунд

Часто задаваемые вопросы

Сколько потоков CaptchaAI нужно на 60 поставщиков?

Считайте не суточный объём, а пик одновременных задач. При 60 поставщиках с суточным срезом и разнесённым расписанием обычно достаточно 5 потоков (BASIC, $15/мес); когда добавляется ежечасный опрос перевозчиков и повторные попытки — 15 потоков (STANDARD, $30/мес).

Что делать, если портал поставщика показывает капчу на каждый заход?

Почти всегда это следствие частоты, а не блокировки. Увеличьте интервал, разнесите источники по расписанию и уберите параллельные заходы на один домен. Если проверка остаётся, зафиксируйте её как риск по источнику и добавьте повторную попытку с экспоненциальной задержкой.

Можно ли обрабатывать капчу-картинку и reCAPTCHA в одном пайплайне?

Да. Различается только метод отправки: post / base64 для картинки и userrecaptcha для reCAPTCHA. Цикл опроса res.php общий, поэтому ветвление сводится к выбору решателя и имени поля в форме.

Что делать, если на портале стоит тип CAPTCHA, который CaptchaAI не решает?

Сначала проверьте список по документации: hCaptcha и FunCaptcha не поддерживаются, GeeTest v4 заявлен как «скоро». Поддерживаются reCAPTCHA v2/v3, Cloudflare Turnstile и Cloudflare Challenge, GeeTest v3, капча-картинка, grid и BLS; CaptchaFox, Friendly Captcha и Lemin — в статусе beta.

Нужны ли прокси для мониторинга поставщиков?

Прокси здесь — обычная инфраструктура: распределение нагрузки и соответствие региону. Если цены на портале зависят от рынка, регион выхода должен совпадать с целевым, иначе в отчёт попадут не те цифры.

Связанные руководства

Соберите монитор один раз и перестаньте терять срезы на проверках — получите API-ключ CaptchaAI и подключите решение CAPTCHA к пайплайну сбора данных.

Комментарии для этой статьи отключены.