Thu thập dữ liệu nghiên cứu truyền thông xã hội bằng cách xử lý CAPTCHA

Crawler mạng xã hội hiếm khi chết ở bước parse HTML. Nó chết ở request thứ 200, khi nền tảng trả về trang thử thách CAPTCHA thay vì dữ liệu. Cách gọn nhất là coi CAPTCHA như một bước trong pipeline: phát hiện thử thách, trích sitekey, gửi task tới CaptchaAI, nhận token rồi gửi lại request.

Bài viết dành cho nhóm nghiên cứu thị trường, team theo dõi thương hiệu và data engineer xử lý dữ liệu công khai. Code là Python chạy được với in.php và res.php.

Bốn bước biến CAPTCHA thành một bước trong pipeline

1. Phát hiện. Tìm data-sitekey, g-recaptcha hoặc cf-turnstile trong HTML trả về.
2. Trích sitekey. Đọc data-sitekey trực tiếp từ trang, không hardcode.
3. Gửi task và polling. POST tới in.php, rồi gọi res.php mỗi 5 giây cho tới khi có kết quả.
4. Dùng token. Đặt vào g-recaptcha-response với reCAPTCHA, hoặc cf-turnstile-response với Turnstile.

Nền tảng mạng xã hội nào bật loại CAPTCHA nào

Cả bốn loại trên đều được CaptchaAI hỗ trợ chính thức.

Nhịp request an toàn theo từng nền tảng

Giãn nhịp request là cách rẻ nhất để ít gặp thử thách hơn.

Crawler mạng xã hội có sẵn bước giải CAPTCHA

Class dưới đây gói cả bốn bước, với solve_captcha dùng chung cho mọi method.

import requests
import time
import re

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("Solve timeout")

class SocialMediaResearcher:
    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 (iPhone; CPU iPhone OS 17_5 like Mac OS X) "
            "AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.5 "
            "Mobile/15E148 Safari/604.1",
            "Accept-Language": "en-US,en;q=0.9",
        })

    def authenticate(self, login_url, credentials, sitekey):
        """Login with CAPTCHA handling."""
        # Load login page
        self.session.get(login_url)

        # Solve CAPTCHA
        token = solve_captcha("userrecaptcha", sitekey, login_url)

        # Submit login
        resp = self.session.post(login_url, data={
            **credentials,
            "g-recaptcha-response": token,
        })
        return resp.status_code == 200

    def collect_profiles(self, profile_urls):
        """Collect public profile data with CAPTCHA handling."""
        profiles = []

        for url in profile_urls:
            try:
                resp = self.session.get(url, timeout=30)

                # Handle CAPTCHA if triggered
                if self._has_captcha(resp.text):
                    resp = self._handle_captcha(resp.text, url)

                profiles.append({
                    "url": url,
                    "data": self._parse_profile(resp.text),
                    "status": "success",
                })
                time.sleep(5)  # Slow down between profiles

            except Exception as e:
                profiles.append({
                    "url": url,
                    "error": str(e),
                    "status": "failed",
                })

        return profiles

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

    def _handle_captcha(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})
        else:
            token = solve_captcha("userrecaptcha", sitekey, url)
            return self.session.post(url, data={"g-recaptcha-response": token})

    def _parse_profile(self, html):
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, "html.parser")
        return {
            "name": self._safe_text(soup, "h1, .profile-name"),
            "bio": self._safe_text(soup, ".bio, .profile-bio"),
            "followers": self._safe_text(soup, "[data-followers], .followers"),
            "posts": self._safe_text(soup, "[data-posts], .posts-count"),
        }

    def _safe_text(self, soup, selector):
        el = soup.select_one(selector)
        return el.get_text(strip=True) if el else ""

collect_profiles trả về bản ghi kèm status, đủ để đo tỷ lệ hỏng theo nền tảng.

Thu thập bài đăng theo hashtag

Vòng lặp phân trang dưới đây dùng lại đúng logic phát hiện và giải ở trên.

def research_hashtag(hashtag, platform_url, pages=5):
    """Collect posts for a specific hashtag."""
    researcher = SocialMediaResearcher(
        proxy="http://user:[email protected]:5000"
    )

    all_posts = []
    for page in range(pages):
        url = f"{platform_url}/explore/tags/{hashtag}?page={page}"
        resp = researcher.session.get(url, timeout=30)

        if researcher._has_captcha(resp.text):
            resp = researcher._handle_captcha(resp.text, url)

        from bs4 import BeautifulSoup
        soup = BeautifulSoup(resp.text, "html.parser")
        posts = soup.select(".post-item, article")
        for post in posts:
            all_posts.append({
                "text": post.get_text(strip=True)[:500],
                "hashtag": hashtag,
                "page": page,
            })

        time.sleep(5)

    return all_posts

Nếu thử thách bắt đầu xuất hiện liên tục, tăng khoảng nghỉ trước khi tăng số thread.

Theo dõi đề cập thương hiệu hằng ngày

Kịch bản sinh nhiều CAPTCHA nhất: cùng tập từ khóa, quét lại mỗi ngày, trên nhiều nền tảng.

import json
from datetime import datetime

class BrandMonitor:
    def __init__(self, brand_name, keywords, proxy=None):
        self.brand = brand_name
        self.keywords = keywords
        self.researcher = SocialMediaResearcher(proxy=proxy)

    def daily_scan(self, platform_urls):
        """Run daily brand mention scan across platforms."""
        report = {
            "brand": self.brand,
            "date": datetime.now().isoformat(),
            "platforms": {},
        }

        for name, url in platform_urls.items():
            mentions = []
            for keyword in self.keywords:
                search_url = f"{url}/search?q={keyword}"
                try:
                    resp = self.researcher.session.get(search_url, timeout=30)

                    if self.researcher._has_captcha(resp.text):
                        resp = self.researcher._handle_captcha(
                            resp.text, search_url,
                        )

                    from bs4 import BeautifulSoup
                    soup = BeautifulSoup(resp.text, "html.parser")
                    results = soup.select(".search-result, .post")
                    mentions.append({
                        "keyword": keyword,
                        "count": len(results),
                    })
                    time.sleep(5)
                except Exception as e:
                    mentions.append({
                        "keyword": keyword,
                        "error": str(e),
                    })

            report["platforms"][name] = mentions

        return report

# Usage
monitor = BrandMonitor(
    brand_name="CaptchaAI",
    keywords=["captchaai", "captcha ai", "captcha solver"],
    proxy="http://user:[email protected]:5000",
)
report = monitor.daily_scan({
    "twitter": "https://twitter-alternative.example.com",
    "reddit": "https://www.reddit.com",
})
print(json.dumps(report, indent=2))

Ví dụ: agency TP.HCM theo dõi thương hiệu FMCG

Một agency ở TP.HCM chạy báo cáo social listening hằng ngày: 6 nền tảng, 20 từ khóa tiếng Việt, vài nghìn request mỗi sáng. CAPTCHA chỉ bật rải rác, nhưng khi bật thì cả job đứng.

CaptchaAI tính tiền theo thread (luồng giải đồng thời) chứ không theo lượt giải, nên chi phí phụ thuộc số CAPTCHA phải giải cùng lúc. BASIC ($15/tháng, 5 thread) đủ cho job tuần tự; chạy song song nhiều khách trong cùng khung giờ sáng thì STANDARD ($30/tháng, 15 thread) hợp lý hơn. Giá niêm yết bằng USD.

Sự cố thường gặp và cách xử lý

Ghi log và tối thiểu hóa dữ liệu

Dữ liệu công khai vẫn có thể chứa dữ liệu cá nhân. Nếu team bạn ở Việt Nam, Nghị định 13/2023/NĐ-CP là lý do thực tế để chỉ lưu trường thật sự cần cho phân tích, ghi log job nào thu thập gì, và đặt hạn xóa dữ liệu thô — đây là thực hành kỹ thuật, không phải tư vấn pháp lý.

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

CaptchaAI có giải được hCaptcha không?

Không. CaptchaAI chưa hỗ trợ hCaptcha và FunCaptcha (Arkose Labs), còn GeeTest v4 mới ở trạng thái sắp ra mắt.

Quét 10.000 trang mỗi ngày thì tốn bao nhiêu?

Chi phí phụ thuộc số CAPTCHA phải giải đồng thời, không phụ thuộc tổng số trang: mỗi gói cho số thread cố định và không giới hạn lượt giải trong tháng.

Nên dùng polling hay callback?

Polling như trong code trên là đủ khi crawler chạy tuần tự. Chạy hàng chục job song song thì callback (CaptchaAI tự gửi kết quả về URL của bạn) giảm nhiều request rỗng tới res.php.

Nên tự parse HTML hay dùng API chính thức của nền tảng?

Nếu nền tảng có API công khai chứa dữ liệu bạn cần, dùng API trước vì ổn định hơn. Chỉ scraping kèm CaptchaAI cho phần dữ liệu công khai mà API không cung cấp.

Hướng dẫn liên quan

• Đa dạng nguồn request để tăng tỷ lệ giải
• Giữ phiên trình duyệt xuyên suốt luồng giải
• Tách profile trình duyệt khi tích hợp CaptchaAI

Đưa bước giải CAPTCHA vào crawler — lấy API key CaptchaAI và chạy lại job nghiên cứu của bạn.