Сценарии использования

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

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


CAPTCHA на платформах электронной коммерции

Платформа Тип капчи Курок Данные
Амазонка reCAPTCHA v2 Большой объем доступа Цена, наличие, отзывы
Уолмарт reCAPTCHA v3 + Cloudflare Обнаружение ботов Инвентарь, цены
подходящий покупка reCAPTCHA v2 Проверка добавления в корзину Акции, цены
Цель Cloudflare Turnstile Автоматизированный доступ Доступность
Магазины Shopify Cloudflare Turnstile Ограничение скорости Данные о продукте
eBay reCAPTCHA v2 Поиск + доступ к листингу Объявления, цены

Монитор продукта

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:pass@residential.proxy.com: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)

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

def scan_category(base_url, category, max_pages=20):
    """Scan an entire product category for stock status."""
    monitor = RetailMonitor(
        proxy="http://user:pass@residential.proxy.com: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

График мониторинга

Тип продукта Проверить частоту Тип прокси
Электроника Каждые 30 минут Вращающийся жилой дом
Продукты питания Каждые 4 часа Вращающийся жилой дом
Мода Каждые 2 часа Жилой
Ограниченные выпуски Каждую 1 минуту авторизованный сетевой выход
Товары для дома Каждые 6 часов Жилой
Товарные товары Каждые 12 часов Центр обработки данных (часто достаточно)

Поиск неисправностей

Проблема Причина Исправить
CAPTCHA на каждой странице товара IP помечен или скорость превышена Увеличение задержки, ротация прокси
Неверная цена удалена Динамическое ценообразование, отображаемое JS Вместо этого используйте Selenium/Puppeteer.
Страница «Проверка роботом» Обнаружение ботов в стиле Amazon Используйте реалистичные заголовки + авторизованный сетевой выход
На складе указано «недоступно». Доступность с географическими ограничениями Сопоставить прокси с целевым регионом
Отсутствуют данные о продукте Структура страницы изменена Обновить селекторы CSS

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

Как часто я могу проверять страницы продуктов?

Каждые 30–60 минут на каждый товар безопасны для большинства розничных продавцов. Более частый мониторинг (1–5 минут) работает, но требует большего количества прокси и запускает CAPTCHA.

Следует ли мне использовать чередующиеся или фиксированные сеансы?

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

Могу ли я контролировать тысячи продуктов?

Да. Распределяйте запросы по нескольким IP-адресам прокси и чередуйте проверки. При 1000 товарах с задержкой в ​​3 секунды полный цикл занимает ~50 минут.


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

  • Ротация авторизованный сетевой выход
  • Прикрепленные и чередующиеся сеансы
  • Мониторинг цепочки поставок

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

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