Tutorials

Создайте панель анализа конкурентов с помощью CaptchaAI

Прайс-листы и карточки товаров у конкурентов почти всегда стоят за защитой от ботов — чаще всего это reCAPTCHA v2. Чтобы собирать эти данные регулярно, а не вручную раз в квартал, нужен скрипт, который решает CAPTCHA через API, сохраняет историю по каждой метрике и сам формирует отчёт на нужную дату. Ниже — рабочий пример такого дашборда на Python: SQLite вместо внешней БД, решение CAPTCHA через CaptchaAI и сравнительные отчёты одной командой.

Схема удобна и для агентств, которые ведут мониторинг сразу для нескольких клиентов: тарифы CaptchaAI фиксированы в долларах за поток, а не за решение, поэтому стоимость мониторинга не скачет вместе с числом собранных страниц. Для команды из двух-трёх человек, которая параллельно ведёт несколько проектов и выставляет клиентам счета в разных валютах, предсказуемая ставка в USD за поток считается заранее — в отличие от тарифов «за запрос», где итог зависит от того, сколько раз на прайс-листе конкурента всплыла CAPTCHA.

Дальше — по шагам: архитектура дашборда, схема хранения данных, модуль решения CAPTCHA, сам скрапер, генератор отчётов и точка входа, которую можно поставить на расписание.

Что понадобится перед стартом:

  • аккаунт CaptchaAI и API-ключ из личного кабинета;
  • Python 3.10+ и виртуальное окружение;
  • пакеты requests и beautifulsoup4 (pip install requests beautifulsoup4).

Архитектура панели анализа конкурентов

Дашборд собран из четырёх частей: скрапер обходит сайты конкурентов, модуль CaptchaAI снимает защиту там, где она стоит, SQLite хранит историю по каждой метрике, а генератор отчётов собирает срез на нужную дату.

Competitor Sites ──> CAPTCHA Solver ──> Data Extractors
                                             │
                                        SQLite Store
                                             │
                                      Dashboard Report

Модель данных для панели анализа конкурентов

Каждое значение — цена плана, число товаров в каталоге, наличие функции — хранится отдельной строкой с меткой времени. Такая плоская схема не требует миграций при добавлении новой метрики и сразу даёт историю по паре «конкурент + метрика».

# models.py
import sqlite3
from datetime import datetime
from dataclasses import dataclass
from typing import Optional


@dataclass
class CompetitorData:
    competitor: str
    metric: str
    value: str
    numeric_value: Optional[float] = None
    url: str = ""
    scraped_at: str = ""

    def __post_init__(self):
        if not self.scraped_at:
            self.scraped_at = datetime.now().isoformat()


class CompetitorDB:
    def __init__(self, path="competitor_data.db"):
        self.conn = sqlite3.connect(path)
        self._init()

    def _init(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS metrics (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                competitor TEXT,
                metric TEXT,
                value TEXT,
                numeric_value REAL,
                url TEXT,
                scraped_at TEXT
            )
        """)
        self.conn.commit()

    def save(self, data: CompetitorData):
        self.conn.execute(
            """INSERT INTO metrics
               (competitor, metric, value, numeric_value, url, scraped_at)
               VALUES (?, ?, ?, ?, ?, ?)""",
            (data.competitor, data.metric, data.value,
             data.numeric_value, data.url, data.scraped_at),
        )
        self.conn.commit()

    def get_history(self, competitor, metric, limit=30):
        cursor = self.conn.execute(
            """SELECT value, numeric_value, scraped_at
               FROM metrics
               WHERE competitor = ? AND metric = ?
               ORDER BY scraped_at DESC LIMIT ?""",
            (competitor, metric, limit),
        )
        return cursor.fetchall()

    def latest_comparison(self, metric):
        cursor = self.conn.execute(
            """SELECT competitor, value, numeric_value, MAX(scraped_at) as latest
               FROM metrics WHERE metric = ?
               GROUP BY competitor ORDER BY numeric_value""",
            (metric,),
        )
        return cursor.fetchall()

Модуль решения CAPTCHA для сбора данных

Класс ниже проверяет HTML на признак data-sitekey, передаёт sitekey и адрес страницы в in.php, затем опрашивает res.php, пока CaptchaAI не вернёт токен g-recaptcha-response. Пауза 15 секунд перед первым опросом и до 24 повторов с интервалом 5 секунд соответствуют обычному времени решения reCAPTCHA v2. Если конкурент со временем поставит на прайс-лист hCaptcha или GeeTest v4, этот модуль не подойдёт — CaptchaAI их пока не решает, и сбор данных с такой страницы потребует другого сценария.

# solver.py
import requests
import time
import re
import os


class CaptchaSolver:
    def __init__(self):
        self.api_key = os.environ["CAPTCHAAI_API_KEY"]

    def solve_if_needed(self, session, url, html):
        if "data-sitekey" not in html:
            return html

        match = re.search(r'data-sitekey="([^"]+)"', html)
        if not match:
            return html

        sitekey = match.group(1)
        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key,
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": url,
            "json": 1,
        }, timeout=30)
        task_id = resp.json()["request"]

        time.sleep(15)
        for _ in range(24):
            resp = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get",
                "id": task_id, "json": 1,
            }, timeout=15)
            data = resp.json()
            if data.get("status") == 1:
                post_resp = session.post(url, data={
                    "g-recaptcha-response": data["request"],
                }, timeout=30)
                return post_resp.text
            if data["request"] != "CAPCHA_NOT_READY":
                raise RuntimeError(data["request"])
            time.sleep(5)

        raise TimeoutError("CAPTCHA solve timeout")

Скрапер конкурентов на Python

CompetitorScraper обходит страницы с прайсами, списками функций и счётчиками товаров, а перед разбором каждой страницы прогоняет HTML через решатель — если находит data-sitekey, скрипт дождётся токена и только потом продолжит парсинг. User-Agent актуального Chrome снижает долю ложных срабатываний антибот-фильтров на обычных, не защищённых CAPTCHA страницах.

# scraper.py
import requests
import re
from bs4 import BeautifulSoup
from solver import CaptchaSolver
from models import CompetitorData


class CompetitorScraper:
    def __init__(self):
        self.solver = CaptchaSolver()
        self.session = requests.Session()
        self.session.headers["User-Agent"] = (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 Chrome/125.0.0.0 Safari/537.36"
        )

    def scrape_pricing(self, competitor_name, url, plan_selector, price_selector):
        html = self._fetch(url)
        soup = BeautifulSoup(html, "html.parser")
        plans = soup.select(plan_selector)
        data = []

        for plan in plans:
            name_el = plan.select_one("h3, h2, .plan-name")
            price_el = plan.select_one(price_selector)

            if not name_el or not price_el:
                continue

            price_text = price_el.get_text(strip=True)
            match = re.search(r'[\d,.]+', price_text)
            numeric = float(match.group().replace(",", "")) if match else None

            data.append(CompetitorData(
                competitor=competitor_name,
                metric=f"price_{name_el.get_text(strip=True).lower().replace(' ', '_')}",
                value=price_text,
                numeric_value=numeric,
                url=url,
            ))

        return data

    def scrape_features(self, competitor_name, url, feature_list_selector):
        html = self._fetch(url)
        soup = BeautifulSoup(html, "html.parser")
        features = soup.select(f"{feature_list_selector} li")

        return [
            CompetitorData(
                competitor=competitor_name,
                metric="feature",
                value=f.get_text(strip=True),
                url=url,
            )
            for f in features if f.get_text(strip=True)
        ]

    def scrape_product_count(self, competitor_name, url, count_selector):
        html = self._fetch(url)
        soup = BeautifulSoup(html, "html.parser")
        el = soup.select_one(count_selector)

        if el:
            text = el.get_text(strip=True)
            match = re.search(r'[\d,]+', text)
            if match:
                count = int(match.group().replace(",", ""))
                return CompetitorData(
                    competitor=competitor_name,
                    metric="product_count",
                    value=text,
                    numeric_value=count,
                    url=url,
                )
        return None

    def _fetch(self, url):
        resp = self.session.get(url, timeout=20)
        return self.solver.solve_if_needed(self.session, url, resp.text)

Генератор сравнительных отчётов

generate_report берёт последнее значение каждой метрики по всем конкурентам и подписывает изменение в долларах или в штуках — для еженедельной сводки этого достаточно без отдельного BI-инструмента. generate_trend разворачивает историю одной метрики, чтобы увидеть, как цена двигалась за последние периоды.

# report.py
from models import CompetitorDB


def generate_report(db: CompetitorDB, metrics):
    lines = ["=" * 60, "Competitor Analysis Report", "=" * 60, ""]

    for metric in metrics:
        results = db.latest_comparison(metric)
        if not results:
            continue

        lines.append(f"--- {metric.replace('_', ' ').title()} ---")
        for comp, value, numeric, ts in results:
            marker = ""
            if numeric is not None:
                marker = f" (${numeric:,.2f})" if "price" in metric else f" ({numeric:,.0f})"
            lines.append(f"  {comp}: {value}{marker}")
        lines.append("")

    return "\n".join(lines)


def generate_trend(db: CompetitorDB, competitor, metric, periods=10):
    history = db.get_history(competitor, metric, limit=periods)
    if not history:
        return f"No data for {competitor} — {metric}"

    lines = [f"Trend: {competitor} — {metric}", "-" * 40]
    for value, numeric, ts in reversed(history):
        date = ts[:10]
        lines.append(f"  {date}: {value}")

    return "\n".join(lines)

Точка входа и расписание запуска

Список COMPETITORS задаёт CSS-селекторы для каждого сайта — настройка нового конкурента почти всегда сводится к их подбору в DevTools. Запускайте main.py по расписанию (cron, systemd-таймер или облачная задача), чтобы данные копились ежедневно, а не только в момент ручного запуска.

# main.py
import time
from models import CompetitorDB
from scraper import CompetitorScraper
from report import generate_report

COMPETITORS = [
    {
        "name": "Competitor A",
        "pricing_url": "https://competitor-a.example.com/pricing",
        "plan_selector": ".pricing-plan",
        "price_selector": ".price",
    },
    {
        "name": "Competitor B",
        "pricing_url": "https://competitor-b.example.com/pricing",
        "plan_selector": ".plan-card",
        "price_selector": ".plan-price",
    },
]


def main():
    db = CompetitorDB()
    scraper = CompetitorScraper()

    for comp in COMPETITORS:
        print(f"Scraping {comp['name']}...")

        try:
            pricing = scraper.scrape_pricing(
                comp["name"], comp["pricing_url"],
                comp["plan_selector"], comp["price_selector"],
            )
            for p in pricing:
                db.save(p)
                print(f"  {p.metric}: {p.value}")
        except Exception as e:
            print(f"  Error: {e}")

        time.sleep(5)

    # Generate report
    metrics = ["price_basic", "price_pro", "price_enterprise", "product_count"]
    report = generate_report(db, metrics)
    print(report)

    with open("competitor_report.txt", "w") as f:
        f.write(report)


if __name__ == "__main__":
    main()

Решение проблем панели анализа конкурентов

Три частые причины, по которым дашборд перестаёт собирать данные: конкурент поменял вёрстку страницы, история ещё не набралась после первого запуска, или на странице внезапно появилась CAPTCHA там, где её раньше не было.

Проблема Причина Решение
Цены не извлекаются Селектор больше не подходит Откройте HTML страницы в DevTools и обновите селекторы для конкретного конкурента
Нет исторических данных Это первый запуск Данные накапливаются постепенно; запускайте скрипт ежедневно, чтобы увидеть тренд
На странице с ценами появилась CAPTCHA Сайт включил защиту от ботов Добавьте паузы между запросами и переиспользуйте cookies сессии
В отчёте видны устаревшие значения Одна и та же запись вставлена повторно Используйте latest_comparison — она группирует по MAX-дате

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


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

Сколько потоков CaptchaAI нужно для мониторинга конкурентов?

Для 5–10 конкурентов с ежедневным сбором данных хватает тарифа BASIC ($15/мес, 5 потоков) — решения идут по очереди по мере обхода страниц. Если добавляете параллельный сбор или десятки конкурентов, переходите на STANDARD ($30/мес, 15 потоков) или выше.

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

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

Как визуализировать динамику цен?

Экспортируйте данные из SQLite и постройте график в matplotlib, либо отдайте CSV-выгрузку в Google Sheets — там графики уже встроены.

Как настроить оповещения об изменении цен?

Сравните сегодняшние значения из latest_comparison со вчерашними и отправляйте уведомление в Slack или на email, когда разница превышает выбранный порог.


Похожие материалы


Отслеживайте конкурентов в промышленном масштабе — начните с CaptchaAI.

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