Agregasi Berita dan Media dengan Penanganan CAPTCHA

Tim monitoring media dan agensi PR yang mengumpulkan artikel dari puluhan sumber sekaligus biasanya berhenti di titik yang sama: begitu volume request naik, Cloudflare Turnstile atau reCAPTCHA langsung muncul di halaman yang tadinya bisa diakses bebas. CaptchaAI menyelesaikan tantangan itu lewat API, jadi pipeline agregasi berita Anda tetap jalan otomatis tanpa harus menyelesaikan CAPTCHA satu per satu secara manual — cocok untuk pemantauan media, riset kompetitor, maupun crawler berita internal.

CAPTCHA yang Muncul Saat Agregasi Berita

Pola di atas cukup konsisten: makin besar skala publikasi dan makin sering IP yang sama menghantam server-nya, makin ketat pula jenis CAPTCHA yang dipasang.

Membangun Agregator Berita dengan Python

Kelas NewsAggregator di bawah menjalankan alur inti: ambil halaman, cek apakah ada widget CAPTCHA, dan kalau ada, kirim ke CaptchaAI sebelum mengambil ulang kontennya. Fungsi solve_captcha memakai pola empat langkah yang sama di semua integrasi CaptchaAI: kirim task ke in.php, simpan task_id dari respons, polling res.php setiap beberapa detik, lalu pakai token begitu status berubah dari CAPCHA_NOT_READY.

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

Method _solve_and_retry memilih token yang tepat berdasarkan widget yang ditemukan di HTML — cf-turnstile-response untuk Cloudflare Turnstile, g-recaptcha-response untuk reCAPTCHA — lalu mengirim ulang request dengan token itu. Untuk situs dengan banyak sumber, aggregate_sources memanggil collect_headlines per sumber sambil menjaga jeda beberapa detik di antara request supaya pola trafiknya tidak terlihat seperti serangan crawl massal.

Monitoring Berita Berdasarkan Kata Kunci

NewsMonitor dibangun di atas NewsAggregator yang sama, tapi menambahkan filter kata kunci dan pelacakan URL yang sudah pernah dilihat (seen_urls) supaya artikel yang sama tidak diproses dua kali. Cocok untuk tim yang memantau topik spesifik — sebutan brand, isu regulasi, atau kompetitor — di banyak portal berita sekaligus.

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

Method continuous_monitor menjalankan scan() dalam loop dengan interval yang bisa diatur. Untuk pemantauan produksi, jalankan ini sebagai systemd service atau container terjadwal, bukan proses while True yang menempel di satu terminal — supaya prosesnya tetap hidup setelah restart server dan mudah dipantau lewat log terpusat.

Menghitung Kebutuhan Thread untuk Pipeline Agregasi Berita

CaptchaAI menagih per thread yang berjalan bersamaan, bukan per solve — biaya bulanan Anda tetap sama berapa pun jumlah artikel yang diambil, selama jumlah request paralel tidak berubah. Ini relevan buat tim monitoring media atau agensi PR di Indonesia yang biasanya cost-sensitive: pipeline kecil dengan 3-5 sumber yang dipoll bergantian biasanya cukup dengan BASIC ($15/bulan, 5 thread), sementara agregator dengan puluhan sumber paralel — misalnya yang di-deploy di region AWS ap-southeast-1 (Singapura) atau ap-southeast-3 (Jakarta) supaya latensinya rendah — lebih pas di STANDARD ($30/bulan, 15 thread) atau ADVANCE ($90/bulan, 50 thread).

Kalau pipeline ini menyentuh data yang berkaitan dengan pengguna (komentar, profil penulis, dan semacamnya), ingat UU Pelindungan Data Pribadi (UU 27/2022) dan UU ITE tetap berlaku. Batasi pengumpulan pada data publik yang memang berhak Anda proses, dan hindari menyimpan data pribadi tanpa dasar yang jelas.

Rencana Anggaran Crawl per Domain

• Tetapkan anggaran crawl per sumber berdasarkan kebutuhan kesegaran konten, seberapa sering CAPTCHA muncul, dan nilai artikel yang dihasilkan dari sumber itu.
• Perlambat domain yang sensitif lebih dulu — jangan terapkan pola retry yang sama untuk semua penerbit, karena tiap situs punya ambang deteksi yang berbeda.
• Lacak jumlah solve, error, dan ritme akses per domain supaya anggaran bisa disesuaikan berdasarkan data nyata, bukan asumsi di awal proyek.

Mengatasi Masalah yang Sering Muncul

Pertanyaan Umum

Berapa banyak sumber berita yang bisa dipantau sekaligus lewat satu pipeline?

Jumlahnya dibatasi oleh thread CaptchaAI yang Anda punya, bukan oleh jumlah domain. Paket BASIC (5 thread) cukup untuk memantau 3-5 sumber secara bergantian; kalau Anda menambah puluhan sumber dan ingin semuanya dipoll paralel, naikkan ke STANDARD (15 thread) atau ADVANCE (50 thread) supaya antrean tidak menumpuk.

Apakah mengumpulkan headline berita untuk riset atau monitoring melanggar hukum di Indonesia?

Ini bukan nasihat hukum, tapi secara umum mengambil headline dan metadata untuk riset atau pemantauan adalah praktik yang wajar, selama Anda tidak mereproduksi artikel berhak cipta secara penuh tanpa izin. Gunakan cuplikan, tautkan kembali ke sumber asli, dan perhatikan UU Pelindungan Data Pribadi kalau kontennya menyentuh data pengguna.

Bagaimana cara mendeteksi dan menangani artikel yang ada di balik paywall?

Cek class CSS yang umum dipakai untuk paywall, atau perhatikan panjang konten yang tiba-tiba terpotong, lalu tandai artikel itu alih-alih memaksakan ekstraksi. Hanya proses konten yang memang berhak Anda akses.

Berapa lama biasanya Cloudflare Turnstile di situs berita selesai diselesaikan?

Turnstile biasanya selesai dalam waktu kurang dari 10 detik dengan tingkat keberhasilan tinggi pada tipe yang didukung; reCAPTCHA v2 di layanan wire bisa memakan waktu sampai di bawah 60 detik karena butuh lebih banyak validasi. Sesuaikan timeout polling di kode Anda dengan angka ini supaya tidak terputus sebelum token benar-benar siap.

Apa yang harus dilakukan kalau satu domain terus mengembalikan CAPTCHA di setiap request?

Itu biasanya tanda IP Anda sudah ditandai oleh sistem deteksi situs tersebut. Perlambat interval request ke domain itu, tambahkan jeda 5 detik atau lebih di antara permintaan, dan pertimbangkan merotasi ke egress jaringan yang diotorisasi dari penyedia tepercaya — bukan cuma menambah retry di layer aplikasi.

Panduan Terkait

• Egress jaringan terotorisasi untuk solve CAPTCHA
• Riset media sosial dengan penanganan CAPTCHA

Kumpulkan berita dari sumber mana pun — ambil API key CaptchaAI dan otomatiskan pipeline agregasi Anda.