Cas d'Usage

Agrégation d'actualités et de médias avec gestion CAPTCHA

Les médias et les plateformes médiatiques utilisent les CAPTCHA pour protéger le contenu de l'agrégation automatisée. Les observateurs des médias, les sociétés de relations publiques et les organismes de recherche doivent collecter des articles par programmation. CaptchaAI gère les défis CAPTCHA dans toutes les sources d'information.

Comment choisir une cadence de collecte réaliste ?

Type de source Cadence raisonnable Pourquoi
Grands médias généralistes Plusieurs fois par heure Les nouveaux titres tombent vite mais la pression doit rester contrôlée
Sources spécialisées ou B2B Quelques passages par jour Le rythme éditorial est souvent plus lent
Paywalls et sources sensibles Cadence prudente et ciblée Les protections se déclenchent plus vite et coûtent plus cher
Veille par mot-clé large Batchs étalés Vous évitez les pointes de trafic et les CAPTCHA inutiles

CAPTCHA sur les plateformes d'information

Type de source CAPTCHA Déclencheur Contenu
Principaux médias Cloudflare Turnstile Détection de robots Articles, titres
Services de fil (AP, Reuters) reCAPTCHA v2 Accès groupé Dernières nouvelles
Publications payantes reCAPTCHA v3 Tentatives d'accès Articles premium
Sites d'information locaux reCAPTCHA v2 Limitation du débit Actualités régionales
Agrégateurs de nouvelles Cloudflare Challenge Détection de grattage Flux agrégés
Sites de communiqués de presse Image CAPTCHA Télécharger des pages Contenu RP

Agrégateur de nouvelles

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

Surveillance des actualités basée sur des mots clés

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:pass@residential.proxy.com:5000",
)
matches = monitor.scan()

Dépannage

Problème Parce que Corriger
Cloudflare bloque toutes les requêtes Détection agressive des robots Utiliser un proxy résidentiel + une UA réaliste
Paywall au lieu de l'article Contenu derrière l'abonnement Détecter le paywall, ignorer ou gérer
CAPTCHA sur chaque page IP signalée Faites pivoter le proxy, ajoutez des délais de plus de 5 secondes
Contenu de l'article vide Contenu rendu en JS Utilisez Selenium/Puppeteer pour les sites SPA
Articles en double Même histoire à partir de plusieurs sources Dédoublonner par similarité de titre

FAQ

L'agrégation d'actualités est-elle légale ?

La collecte de titres et de métadonnées à des fins de recherche ou de suivi est une pratique courante. La reproduction d’articles entièrement protégés par le droit d’auteur sans autorisation ne l’est pas. Utilisez des extraits et créez un lien vers la source d’origine.

Comment gérer le contenu payant ?

Détectez le paywall (recherchez les classes CSS du paywall ou la longueur limitée du contenu) et signalez-le. Accédez uniquement au contenu que vous êtes autorisé à consulter.

Quel type de proxy fonctionne le mieux pour les sites d’actualités ?

Les proxys résidentiels rotatifs fonctionnent mieux. Les principaux médias utilisent Cloudflare, qui bloque de manière agressive les adresses IP des centres de données.


Guides connexes

  • Rotation des procurations résidentielles
  • Recherche sur les réseaux sociaux

Les commentaires sont désactivés pour cet article.