Мониторинг розничных товарных запасов с обработкой CAPTCHA

Мониторинг остатков ломается не на разборе HTML, а на третьем десятке запросов: магазин видит серию однотипных обращений и отдаёт страницу проверки вместо карточки товара. Лечится это одним звеном в цикле — детектор проверки, отправка задачи в CaptchaAI, повтор запроса с токеном. Ниже — схема на Python: как распознавать reCAPTCHA v2/v3 и Cloudflare Turnstile на страницах товаров, как выбрать частоту опроса и сколько потоков заложить в тариф.

Почему карточка товара отдаёт проверку вместо цены

Проверку запускает совокупность сигналов: частота запросов с одного IP, отсутствие привычной последовательности переходов, репутация адреса, обращение к цене корзины. Чем ближе запрос к деньгам, тем агрессивнее площадка защищает ответ.

Картина по крупным западным площадкам ориентировочно такая — тип проверки со временем меняется, поэтому определяйте его по разметке страницы, а не по таблице:

CaptchaAI решает все типы из этой таблицы: reCAPTCHA v2 (включая Invisible, Callback и Enterprise), reCAPTCHA v3, Cloudflare Turnstile и Challenge, GeeTest v3, изображения и grid-капчи, BLS. CaptchaFox, Friendly Captcha и Lemin — в статусе beta. hCaptcha и FunCaptcha не поддерживаются, GeeTest v4 заявлен как «скоро».

Куда встроить решение CAPTCHA в цикле опроса

Парсер переписывать не нужно — достаточно вставить проверку между ответом и разбором разметки:

1. Запрос страницы. GET через сессию с постоянными заголовками.
2. Детектор. Ищите маркеры data-sitekey, g-recaptcha, cf-turnstile; если их нет — сразу к парсингу.
3. Отправка задачи. POST на in.php с методом (userrecaptcha или turnstile), sitekey и pageurl; в ответе приходит ID задачи.
4. Опрос результата. Опрашивайте res.php до токена; CAPCHA_NOT_READY означает «ещё считаем».
5. Повтор с токеном. Повторите исходный запрос, положив токен в g-recaptcha-response или cf-turnstile-response.

Для расчёта стоимости важно: поток занят только на шагах 3–4. Страницы, которые отдались без проверки, потоки не расходуют вовсе.

Монитор товара на Python

Класс ниже держит одну сессию, сам ловит проверку, решает её, повторяет запрос и сообщает только об изменениях.

import requests
import time
import re
import json
from datetime import datetime
from bs4 import BeautifulSoup

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("Timeout")


class RetailMonitor:
    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",
            "Accept": "text/html,application/xhtml+xml,*/*;q=0.8",
            "Accept-Language": "en-US,en;q=0.9",
        })

    def check_product(self, url):
        """Check single product's price and availability."""
        resp = self.session.get(url, timeout=30)

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

        soup = BeautifulSoup(resp.text, "html.parser")
        return {
            "url": url,
            "title": self._text(soup, "h1, .product-title, #productTitle"),
            "price": self._text(soup, ".price, .a-price .a-offscreen, .prod-price"),
            "availability": self._text(soup, "#availability, .stock-status, .fulfillment"),
            "in_stock": self._check_stock(soup),
            "timestamp": datetime.now().isoformat(),
        }

    def monitor_products(self, product_urls, interval_sec=1800):
        """Continuously monitor products for changes."""
        history = {}

        while True:
            for url in product_urls:
                try:
                    current = self.check_product(url)

                    # Check for changes
                    prev = history.get(url)
                    if prev:
                        changes = self._detect_changes(prev, current)
                        if changes:
                            self._alert(current["title"], changes)

                    history[url] = current
                    time.sleep(3)

                except Exception as e:
                    print(f"Error checking {url}: {e}")

            print(f"Cycle complete: {len(product_urls)} products checked")
            time.sleep(interval_sec)

    def track_prices(self, product_urls, output_file="prices.json"):
        """Single price check across all products."""
        results = []
        for url in product_urls:
            try:
                data = self.check_product(url)
                results.append(data)
                time.sleep(3)
            except Exception as e:
                results.append({"url": url, "error": str(e)})

        with open(output_file, "w") as f:
            json.dump(results, f, indent=2)
        print(f"Tracked {len(results)} products → {output_file}")
        return results

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

    def _solve_and_retry(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 _text(self, soup, selector):
        el = soup.select_one(selector)
        return el.get_text(strip=True) if el else ""

    def _check_stock(self, soup):
        stock_el = soup.select_one("#availability, .stock-status")
        if stock_el:
            text = stock_el.get_text(strip=True).lower()
            return "in stock" in text or "available" in text
        return None

    def _detect_changes(self, prev, current):
        changes = []
        if prev["price"] != current["price"]:
            changes.append(f"Price: {prev['price']} → {current['price']}")
        if prev["in_stock"] != current["in_stock"]:
            status = "In Stock" if current["in_stock"] else "Out of Stock"
            changes.append(f"Stock: → {status}")
        return changes

    def _alert(self, title, changes):
        print(f"ALERT [{title}]: {', '.join(changes)}")


# Usage
monitor = RetailMonitor(
    proxy="http://user:[email protected]:5000"
)

products = [
    "https://store.example.com/product/abc123",
    "https://store.example.com/product/def456",
    "https://store.example.com/product/ghi789",
]

# One-time price check
results = monitor.track_prices(products)

# Or continuous monitoring (every 30 min)
# monitor.monitor_products(products, interval_sec=1800)

Два места стоят внимания. time.sleep(3) между товарами — самый дешёвый способ снизить долю страниц с проверкой. Цикл на 60 итераций по 5 с — разумный запас для reCAPTCHA v2; на Turnstile он почти не выбирается.

Сканирование категории постранично

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

def scan_category(base_url, category, max_pages=20):
    """Scan an entire product category for stock status."""
    monitor = RetailMonitor(
        proxy="http://user:[email protected]:5000"
    )

    all_products = []
    for page in range(1, max_pages + 1):
        url = f"{base_url}/{category}?page={page}"
        resp = monitor.session.get(url, timeout=30)

        if monitor._has_captcha(resp.text):
            resp = monitor._solve_and_retry(resp.text, url)

        soup = BeautifulSoup(resp.text, "html.parser")
        items = soup.select(".product-card, .s-result-item")

        if not items:
            break

        for item in items:
            all_products.append({
                "name": monitor._text(item, ".product-name, .a-text-normal"),
                "price": monitor._text(item, ".price, .a-price"),
                "stock": monitor._text(item, ".stock, .a-color-success"),
                "url": item.select_one("a")["href"] if item.select_one("a") else "",
            })

        time.sleep(3)

    return all_products

Сравнение цен между магазинами

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

def compare_product_across_stores(product_name, stores):
    """Compare prices across retailers for the same product."""
    results = []

    for store in stores:
        monitor = RetailMonitor(proxy=store.get("proxy"))
        search_url = f"{store['base_url']}/search?q={product_name}"

        try:
            resp = monitor.session.get(search_url, timeout=30)
            if monitor._has_captcha(resp.text):
                resp = monitor._solve_and_retry(resp.text, search_url)

            soup = BeautifulSoup(resp.text, "html.parser")
            first_result = soup.select_one(".product-card, .s-result-item")

            if first_result:
                results.append({
                    "store": store["name"],
                    "price": monitor._text(first_result, ".price"),
                    "in_stock": "in stock" in first_result.get_text().lower(),
                })
        except Exception as e:
            results.append({"store": store["name"], "error": str(e)})

        time.sleep(5)

    results.sort(key=lambda x: x.get("price", "zzzz"))
    return results

Складывайте результат с отметкой времени: без истории видна цена, но не динамика.

Частота опроса и сетевая инфраструктура

Частота — главный рычаг. Практичные интервалы, от которых можно отталкиваться:

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

Сколько потоков заложить в тариф

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

Пример: каталог на 1000 SKU, цикл каждые 30 минут, задержка 3 с между товарами. Цикл занимает около 50 минут, проверка возникает примерно на 5 % страниц — это ~50 задач за цикл, редко больше двух-трёх одновременно. Профиль закрывается тарифом BASIC ($15/мес, 5 потоков).

При параллельном сканировании в 10–20 воркеров задач больше — смотрите STANDARD ($30/мес, 15 потоков); для десятков клиентских каталогов — ADVANCE ($90/мес, 50 потоков). Выше идут PREMIUM ($170/мес, 100 потоков), CORPORATE ($240/мес, 150 потоков) и ENTERPRISE ($300/мес, 200 потоков). Правило: удвойте ожидаемое число одновременных задач — поток освобождается сразу после решения. Фиксированная цена в USD удобна тем, кто считает бюджет в волатильной валюте.

Локальный контекст: маркетплейсы и персональные данные

Команды в России, Беларуси и Казахстане чаще мониторят локальные маркетплейсы, чем Amazon. Схема та же, меняются два параметра. Первый — регион сетевого выхода: цена и наличие зависят от склада и города, поэтому выход в «неправильном» регионе даёт формально корректный, но бесполезный ответ. Второй — тип проверки: определяйте его по разметке площадки.

Отдельный сюжет — сами данные. Публичные цены и остатки персональными данными не являются, а карточки с отзывами и именами авторов — уже пограничная зона. Собирайте только то, что вправе обрабатывать, помня о требованиях 152-ФЗ «О персональных данных» (для трансграничных проектов — сопоставимой практике GDPR).

Диагностика типичных сбоев

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

Какие типы CAPTCHA CaptchaAI решает на страницах товаров?

reCAPTCHA v2 (включая Invisible, Callback, Enterprise), reCAPTCHA v3, Turnstile, Cloudflare Challenge, GeeTest v3, изображения, grid-капчи, BLS. CaptchaFox, Friendly Captcha и Lemin — в статусе beta. hCaptcha, FunCaptcha и GeeTest v4 не поддерживаются.

Сколько потоков нужно, если проверки появляются редко?

Считайте пиковую одновременность, а не общее число решений. Одному последовательному воркеру хватает BASIC ($15/мес, 5 потоков): поток занят только пока задача решается.

Что делать, если цена подгружается через JavaScript?

requests вернёт разметку без цены. Переведите такие страницы на Selenium или Puppeteer; решение проверки не меняется — токен так же подставляется в поле формы.

Почему один и тот же товар показывает разное наличие с разных IP?

Наличие привязано к региону и складу. Зафиксируйте регион выхода за конкретным заданием, иначе история смешает несопоставимые данные.

Законно ли собирать цены конкурентов?

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

Связанные материалы

• Ротация резидентных прокси при решении CAPTCHA
• Закреплённые и ротируемые сессии: что выбрать
• Мониторинг цепочки поставок с обработкой CAPTCHA

Отслеживайте остатки и цены без пауз на проверку — получите API-ключ CaptchaAI и подключите решение CAPTCHA к своему циклу мониторинга.