Use Cases

Агрегация новостей и СМИ с обработкой CAPTCHA

Собрать сотни заголовков в час без банов и обрывов ленты — стандартная задача для медиамониторинга, PR-агентств и исследовательских команд. Главная преграда здесь не сам парсинг HTML, а Cloudflare Turnstile, reCAPTCHA v2 и reCAPTCHA v3, которые крупные издания и новостные агрегаторы ставят перед статьями и лентами. CaptchaAI решает эти проверки через API, поэтому агрегатор новостей возобновляет сбор данных автоматически, без ручного вмешательства и без пауз на «покликать капчу руками».


Где на новостных сайтах встречается CAPTCHA

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

Тип источника Тип CAPTCHA Триггер Контент
Крупные новостные агентства Cloudflare Turnstile Обнаружение ботов Статьи, заголовки
Информагентства (AP, Reuters) reCAPTCHA v2 Массовый доступ Срочные новости
Издания с платным доступом reCAPTCHA v3 Попытки доступа Премиум-статьи
Локальные новостные сайты reCAPTCHA v2 Ограничение частоты запросов Региональные новости
Новостные агрегаторы Cloudflare Challenge Обнаружение парсинга Агрегированные ленты
Сайты пресс-релизов Графическая CAPTCHA Скачивание материалов PR-контент

Как устроен агрегатор новостей

Класс NewsAggregator ниже реализует минимальный, но рабочий конвейер: получает HTML источника, проверяет разметку на data-sitekey, g-recaptcha или cf-turnstile, при необходимости отправляет sitekey и pageurl в CaptchaAI и повторяет запрос уже с решённым токеном. Для reCAPTCHA v2 токен уходит в поле g-recaptcha-response, для Turnstile — в cf-turnstile-response; сервер CaptchaAI отвечает JSON с полем request, где сначала лежит ID задачи, а после опроса res.php — сам токен. Дальше HTML статьи разбирается через BeautifulSoup по типовым CSS-селекторам заголовков, автора, даты и тела текста.

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

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 NewsAggregator:
    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-Language": "en-US,en;q=0.9",
        })

    def collect_headlines(self, source_url, section=None):
        """Collect headlines from a news source."""
        url = f"{source_url}/{section}" if section else source_url
        resp = self.session.get(url, timeout=30)

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

        soup = BeautifulSoup(resp.text, "html.parser")
        articles = []

        for item in soup.select("article, .story, .headline-item, h2 a, h3 a"):
            link = item if item.name == "a" else item.select_one("a")
            if link:
                articles.append({
                    "title": link.get_text(strip=True),
                    "url": self._abs_url(source_url, link.get("href", "")),
                    "source": source_url,
                    "collected_at": datetime.now().isoformat(),
                })

        return articles

    def get_article(self, article_url):
        """Fetch full article content."""
        resp = self.session.get(article_url, timeout=30)

        if self._has_captcha(resp.text):
            resp = self._solve_and_retry(resp.text, article_url)

        soup = BeautifulSoup(resp.text, "html.parser")

        # Remove unwanted elements
        for tag in soup.select("script, style, nav, footer, .ad, .sidebar"):
            tag.decompose()

        content_el = soup.select_one(
            "article, .article-body, .story-body, .entry-content"
        )

        return {
            "url": article_url,
            "title": self._text(soup, "h1, .article-title"),
            "author": self._text(soup, ".author, .byline, [rel='author']"),
            "date": self._text(soup, "time, .publish-date, .article-date"),
            "content": content_el.get_text(separator="\n", strip=True) if content_el else "",
            "word_count": len(content_el.get_text().split()) if content_el else 0,
        }

    def aggregate_sources(self, sources, max_articles_per=20):
        """Aggregate headlines across multiple sources."""
        all_articles = []

        for source in sources:
            try:
                articles = self.collect_headlines(source["url"], source.get("section"))
                all_articles.extend(articles[:max_articles_per])
                print(f"{source['name']}: {len(articles)} headlines")
            except Exception as e:
                print(f"{source['name']}: Error - {e}")
            time.sleep(3)

        return all_articles

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

    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})
        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 _abs_url(self, base, href):
        if href.startswith("http"):
            return href
        return base.rstrip("/") + "/" + href.lstrip("/")


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

sources = [
    {"name": "Tech News A", "url": "https://technews-a.example.com", "section": "latest"},
    {"name": "Business B", "url": "https://business-b.example.com", "section": "tech"},
    {"name": "Industry C", "url": "https://industry-c.example.com"},
]

headlines = aggregator.aggregate_sources(sources)
print(f"Total: {len(headlines)} headlines collected")

Задержка time.sleep(3) между источниками и резидентный прокси в примере — не декоративные детали: без них Cloudflare помечает IP как подозрительный уже после первых 10–15 запросов подряд, и дальше на каждой странице будет всплывать новая проверка вместо статьи.


Мониторинг СМИ по ключевым словам

Класс NewsMonitor берёт список ключевых слов и прогоняет через них NewsAggregator по расписанию, откладывая уже увиденные URL в seen_urls, чтобы не сигнализировать дважды об одной и той же новости. Для мониторинга упоминаний бренда или отслеживания темы это часто надёжнее RSS: у многих региональных изданий ленты нет вовсе, а continuous_monitor работает поверх произвольного списка источников с любым интервалом опроса.

class NewsMonitor:
    def __init__(self, keywords, sources, proxy=None):
        self.keywords = [kw.lower() for kw in keywords]
        self.aggregator = NewsAggregator(proxy=proxy)
        self.sources = sources
        self.seen_urls = set()

    def scan(self):
        """Scan for articles matching keywords."""
        headlines = self.aggregator.aggregate_sources(self.sources)
        matches = []

        for article in headlines:
            if article["url"] in self.seen_urls:
                continue

            title_lower = article["title"].lower()
            matched_kws = [kw for kw in self.keywords if kw in title_lower]

            if matched_kws:
                article["matched_keywords"] = matched_kws
                matches.append(article)
                self.seen_urls.add(article["url"])

        return matches

    def continuous_monitor(self, interval_min=30):
        """Run continuous monitoring with alerts."""
        while True:
            matches = self.scan()
            if matches:
                print(f"\n=== {len(matches)} new matches found ===")
                for m in matches:
                    print(f"  [{', '.join(m['matched_keywords'])}] {m['title']}")
                    print(f"    {m['url']}")
            else:
                print(f"No new matches at {datetime.now().strftime('%H:%M')}")

            time.sleep(interval_min * 60)


# Monitor for specific topics
monitor = NewsMonitor(
    keywords=["captcha", "bot detection", "web scraping", "automation"],
    sources=sources,
    proxy="http://user:[email protected]:5000",
)
matches = monitor.scan()

Типичные проблемы при сборе новостей и их решения

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

Проблема Причина Решение
Cloudflare блокирует все запросы Агрессивное обнаружение ботов Резидентный прокси + реалистичный User-Agent
Paywall вместо статьи Контент доступен только по подписке Определяйте paywall по CSS-классам или по резко укороченной длине текста и обрабатывайте отдельно
CAPTCHA всплывает на каждой странице IP помечен как подозрительный Ротация прокси, задержка от 5 секунд между запросами
Тело статьи пустое Контент рендерится через JS Selenium или Puppeteer для SPA-сайтов
Дублирующиеся статьи Одна новость с нескольких источников Дедупликация по схожести заголовков

Пример: мониторинг упоминаний бренда для PR-агентства

PR-агентство в Алматы или Москве отслеживает упоминания клиента на 40 региональных и федеральных изданиях. Запуск continuous_monitor с интервалом 30 минут по 40 источникам — это до 80 обращений в час, и почти каждое рано или поздно встретит Cloudflare Turnstile или reCAPTCHA v2. Плана BASIC ($15/мес, 5 потоков) достаточно для пилота на 5–10 источниках; при масштабировании до полусотни изданий и более частом опросе разумнее перейти на STANDARD ($30/мес, 15 потоков) или ADVANCE ($90/мес, 50 потоков) — тарификация идёт по потокам, а не по числу решённых CAPTCHA, поэтому стоимость не растёт линейно вместе с объёмом трафика.


План бюджета домена

  • Назначайте каждому источнику бюджет сканирования исходя из того, как часто там выходят новости, из давления CAPTCHA и из ценности потока контента.
  • Сначала снижайте частоту запросов к чувствительным доменам, а не применяйте один и тот же шаблон повторов ко всем изданиям.
  • Считайте решения, промахи и частоту срабатывания CAPTCHA по каждому домену — так бюджет можно калибровать по реальным цифрам, а не на глаз.

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

Сколько источников можно опрашивать одновременно на плане BASIC?

BASIC даёт 5 потоков — это до 5 CAPTCHA, решаемых параллельно, а не 5 источников в буквальном смысле: пока одна проверка решается, поток занят, как только пришёл ответ — берёт следующую задачу. Для 5–10 источников с интервалом опроса 20–30 минут этого обычно хватает; при более частых обращениях или большем числе изданий переходите на STANDARD (15 потоков) или ADVANCE (50 потоков).

Как отличить блокировку Cloudflare от обычной сетевой ошибки?

Сетевая ошибка — это таймаут, 5xx-код или разрыв соединения, и она не зависит от числа предыдущих запросов. Блокировка Cloudflare почти всегда приходит как HTML-страница проверки с cf-turnstile в разметке или как код 403 после нескольких запросов подряд с одного IP. Если _has_captcha из примера выше возвращает True при статусе 200 — это Cloudflare Turnstile, а не сбой сети.

Как поступать с контентом за paywall?

Определяйте paywall по CSS-классам ограниченного доступа или по резко укороченной длине текста и помечайте такие статьи отдельно. Забирайте только тот контент, на доступ к которому у вас есть право: полный текст платной статьи для перепечатки без лицензии агрегировать нельзя, а заголовок и первый абзац для мониторинга — обычная практика.

Нужно ли уведомлять издание о сборе его материалов?

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


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

  • Ротация резидентных прокси
  • Исследования в социальных сетях

Собирайте новости из любого источника — получите API-ключ CaptchaAI и автоматизируйте сбор контента.

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