Tổng hợp tin tức và phương tiện truyền thông bằng cách xử lý CAPTCHA

Nhiều agency PR và đội media monitoring cần thu thập hàng trăm bài báo mỗi ngày từ nhiều nguồn để theo dõi nhắc đến khách hàng. Vấn đề: phần lớn trang tin lớn chặn crawler bằng Cloudflare Turnstile hoặc reCAPTCHA khi phát hiện lượng truy cập bất thường. CaptchaAI giải các thử thách đó qua API để bộ tổng hợp tiếp tục chạy thay vì dừng lại ở mỗi trang bị chặn.

Một agency PR tại TP.HCM theo dõi 40 đầu báo quốc tế mỗi ngày cho khách hàng — gần một nửa số trang đó chặn bằng Cloudflare Turnstile hoặc reCAPTCHA v2/v3.

CAPTCHA xuất hiện ở đâu trên các trang tin tức

Câu hỏi thường gặp

Tổng hợp tin tức tự động có vi phạm bản quyền không?

Thu thập tiêu đề, metadata để nghiên cứu là thông lệ phổ biến; sao chép toàn văn bài có bản quyền thì không — dùng đoạn trích và liên kết ngược nguồn gốc.

Nên dùng polling hay webhook khi giám sát nhiều nguồn tin?

Polling đơn giản hơn và đủ cho hầu hết khối lượng theo dõi. Webhook đáng cân nhắc khi giám sát hàng trăm nguồn và cần độ trễ thấp hơn.

Giám sát 50 nguồn tin mỗi ngày cần bao nhiêu thread CaptchaAI?

Gói STANDARD ($30/tháng, 15 thread) thường đủ cho vài chục nguồn chạy theo lô. Chạy nhiều batch song song thì cân nhắc ADVANCE ($90/tháng, 50 thread).

Làm sao phát hiện và bỏ qua nội dung sau tường phí?

Tìm lớp CSS đặc trưng của tường phí hoặc độ dài nội dung bất thường ngắn, rồi gắn cờ bài đó thay vì cố lấy toàn văn.

Ba bước xử lý CAPTCHA khi tổng hợp tin

1. Phát hiện CAPTCHA trên trang qua dấu hiệu data-sitekey, g-recaptcha, cf-turnstile.
2. Gửi task tới CaptchaAI qua in.php, polling res.php cho đến khi có token.
3. Gửi lại request kèm token vừa nhận rồi tiếp tục thu thập nội dung.

Xây bộ tổng hợp tin tức bằng Python

Class NewsAggregator hiện thực ba bước trên cho nhiều nguồn cùng lúc.

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

Giám sát tin tức theo từ khóa với cảnh báo tự động

NewsMonitor lọc kết quả của NewsAggregator theo từ khóa và bỏ qua URL đã thấy.

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

Khắc phục sự cố thường gặp

Lên ngân sách thu thập theo từng miền

• Chia ngân sách crawl theo tần suất cập nhật, mức độ CAPTCHA và giá trị nội dung từng nguồn.
• Giãn tốc độ request cho miền nhạy cảm trước, không áp cùng một kiểu retry cho mọi nhà xuất bản.
• Theo dõi số lần giải, tỷ lệ bỏ sót và nhịp tìm nạp theo miền để điều chỉnh ngân sách bằng số liệu thực tế.

Hướng dẫn liên quan

• Đa dạng nguồn yêu cầu khi thu thập dữ liệu quy mô lớn
• Thu thập dữ liệu mạng xã hội để nghiên cứu

Tổng hợp tin tức từ mọi nguồn — lấy API key CaptchaAI và tự động hóa việc thu thập nội dung.