Quét các trang web được bảo vệ bằng CAPTCHA

Một trang thử thách trả về giữa job crawl không có nghĩa là pipeline sai. Nó chỉ thiếu một mắt xích: đọc loại CAPTCHA từ HTML, gửi tham số lên in.php, chờ kết quả ở res.php, rồi gắn token vào request kế tiếp trong cùng session. Ví dụ xuyên suốt bài là tình huống quen ở TP.HCM — team data của một nhà bán lẻ theo dõi giá catalog trên vài sàn thương mại điện tử trong nước, 5.000 trang mỗi đêm, và job cron 1 giờ sáng dừng ở trang thứ 400 vì sàn bật thử thách cho nhánh tìm kiếm.

Đối chiếu loại CAPTCHA với method CaptchaAI

Trước hết xác định trang đích dùng gì. Cột phải là giá trị method gửi lên in.php.

Hai lưu ý trước khi code:

• CaptchaFox, Friendly Captcha và Lemin đang ở bản beta — thử nghiệm được, nhưng tự đo trước khi vào production. GeeTest v4 mới ở trạng thái sắp ra mắt.
• Trang đích chạy hCaptcha hoặc FunCaptcha thì dừng ở đây: hai loại này không được hỗ trợ, bạn cần phương án khác cho riêng bước đó.

Cách 1: phát hiện trong response rồi mới giải

Mặc định nên chọn cho hầu hết dự án. Crawler chạy bình thường, mỗi response được soi dấu hiệu thử thách; chỉ khi có mới tốn một lượt giải, nên thread gần như luôn rảnh.

import requests
import time
from bs4 import BeautifulSoup

API_KEY = "YOUR_API_KEY"

class ProtectedScraper:
    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        })

    def scrape(self, url):
        resp = self.session.get(url)

        # Check for CAPTCHA
        if self._has_captcha(resp.text):
            resp = self._handle_captcha(resp.text, url)

        return resp.text

    def _has_captcha(self, html):
        indicators = ["g-recaptcha", "cf-turnstile", "h-captcha", "captcha"]
        return any(ind in html.lower() for ind in indicators)

    def _handle_captcha(self, html, url):
        soup = BeautifulSoup(html, "html.parser")

        # reCAPTCHA v2
        rc = soup.find("div", class_="g-recaptcha")
        if rc:
            token = self._solve_recaptcha(rc["data-sitekey"], url)
            return self.session.post(url, data={"g-recaptcha-response": token})

        # Cloudflare Turnstile
        ts = soup.find("div", class_="cf-turnstile")
        if ts:
            token = self._solve_turnstile(ts["data-sitekey"], url)
            return self.session.post(url, data={"cf-turnstile-response": token})

        raise Exception("Unknown CAPTCHA type")

    def _solve_recaptcha(self, site_key, page_url):
        resp = requests.get("https://ocr.captchaai.com/in.php", params={
            "key": API_KEY, "method": "userrecaptcha",
            "googlekey": site_key, "pageurl": page_url
        })
        return self._poll(resp.text.split("|")[1])

    def _solve_turnstile(self, site_key, page_url):
        resp = requests.get("https://ocr.captchaai.com/in.php", params={
            "key": API_KEY, "method": "turnstile",
            "sitekey": site_key, "pageurl": page_url
        })
        return self._poll(resp.text.split("|")[1])

    def _poll(self, task_id):
        for _ in range(60):
            time.sleep(5)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": API_KEY, "action": "get", "id": task_id
            })
            if result.text == "CAPCHA_NOT_READY": continue
            if result.text.startswith("OK|"): return result.text.split("|")[1]
            raise Exception(result.text)
        raise TimeoutError()

# Usage
scraper = ProtectedScraper()
html = scraper.scrape("https://example.com/data")

Hai chi tiết quyết định đoạn code này chạy êm hay hỏng lúc 1 giờ sáng:

• _has_captcha chỉ dò chuỗi — rẻ và đủ dùng, nhưng từ "captcha" cũng nằm trong footer hay script phân tích, nên siết chỉ báo cho đúng trang bạn thu thập.
• Sitekey luôn đọc từ data-sitekey, không hard-code: giá trị này đổi mỗi lần trang triển khai lại. _poll hỏi res.php mỗi 5 giây, tối đa 60 lần — không muốn giữ luồng chờ thì dùng callback.

Cách 2: giải sẵn cho endpoint lần nào cũng bật thử thách

Có những endpoint không bao giờ đi thẳng: form tìm kiếm nâng cao, trang tra cứu mã, bước xác nhận trước khi trả dữ liệu. Với các endpoint biết chắc, lấy token trước rồi gửi kèm ngay trong lần POST đầu, khỏi tốn một vòng request để phát hiện.

def scrape_known_captcha_page(url, site_key):
    # Solve before even loading the page
    token = solve_recaptcha(site_key, url)

    # Submit directly with token
    resp = requests.post(url, data={
        "g-recaptcha-response": token,
        "query": "search term"
    })
    return resp.text

Cách 3: khi Cloudflare chặn nguyên trang

Khác hẳn hai cách trên: trang trả về HTML chặn toàn màn hình chứ không phải widget trong form. Dùng method=cloudflare_challenge, và thứ nhận về là cặp cookie phiên cùng user agent chứ không phải token.

def get_cloudflare_clearance(url, proxy):
    resp = requests.get("https://ocr.captchaai.com/in.php", params={
        "key": API_KEY,
        "method": "cloudflare_challenge",
        "pageurl": url,
        "proxy": proxy,
        "proxytype": "HTTP"
    })
    task_id = resp.text.split("|")[1]

    for _ in range(60):
        time.sleep(5)
        result = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY, "action": "get", "id": task_id
        })
        if result.text == "CAPCHA_NOT_READY": continue
        if "qa_session_cookie" in result.text:
            # Parse qa_session_cookie and user_agent from response
            return result.text
    raise TimeoutError()

Điều kiện bắt buộc: giữ nguyên proxy và user agent cho mọi request trong phiên — tham số proxy lúc giải phải trùng proxy mà crawler dùng khi đọc dữ liệu.

Ghép vào vòng lặp phân trang

Cả ba cách trên chỉ là một bước trong vòng lặp lớn hơn. Mẫu dưới bắt lỗi từng trang, để một trang hỏng không kéo chết cả job.

def scrape_multiple_pages(base_url, site_key, pages):
    scraper = ProtectedScraper()
    results = []

    for page in pages:
        url = f"{base_url}?page={page}"
        try:
            html = scraper.scrape(url)
            soup = BeautifulSoup(html, "html.parser")
            items = soup.find_all("div", class_="item")
            results.extend([item.text.strip() for item in items])
            print(f"Page {page}: {len(items)} items")
        except Exception as e:
            print(f"Page {page} failed: {e}")

        time.sleep(random.uniform(2, 5))

    return results

Khoảng nghỉ ngẫu nhiên 2–5 giây không phải mẹo vặt: nhịp request đều tăm tắp là lý do phổ biến nhất khiến trang bật thử thách ở mọi lượt. Nếu dữ liệu có thông tin cá nhân, Nghị định 13/2023/NĐ-CP là lý do đủ tốt để ghi log nguồn và chỉ lưu trường thật sự cần.

Chọn gói theo đỉnh tải, không theo tổng số trang

CaptchaAI tính tiền theo thread — số lượt giải chạy đồng thời — chứ không theo từng lượt, nên con số cần đếm là đỉnh tải chứ không phải tổng số trang.

Quay lại crawler 5.000 trang mỗi đêm: 3% trang bật thử thách là chừng 150 lượt giải rải trong vài giờ, hiếm khi quá vài lượt cùng lúc, nên BASIC ($15/tháng, 5 thread) thừa sức. Lên mười worker song song mà worker nào cũng có thể đang chờ token thì STANDARD ($30/tháng, 15 thread) hợp lý; job nhiều tiến trình thì nhìn sang ADVANCE ($90/tháng, 50 thread). Giá niêm yết bằng USD.

Bảng tra khi crawler vẫn hỏng

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

Crawler chạy bao nhiêu worker thì đủ 5 thread?

Đếm số lượt giải chạy chồng nhau ở đỉnh tải, không đếm tổng số trang. Crawler tuần tự thì một worker hiếm khi giữ quá một thread, nên 5 thread của gói BASIC ($15/tháng) đủ cho vài worker song song.

Phát hiện CAPTCHA bằng cách dò chuỗi HTML có đủ tin cậy không?

Đủ cho phần lớn trang nhưng dễ báo nhầm. Thay chỉ báo chung bằng đúng class mà trang đích render, và log lại mỗi lần nhận diện để rà soát.

Giải một lần rồi tái sử dụng token cho nhiều trang được không?

Không. Mỗi token gắn với một lần gửi form, hết hạn sau khoảng 120 giây, nên phải giải riêng từng lượt.

Trang render bằng JavaScript thì lấy sitekey ở đâu?

Render DOM bằng Selenium, Puppeteer hoặc Playwright rồi mới đọc data-sitekey, vì tham số CAPTCHA chỉ xuất hiện sau khi script chạy. Chi tiết ở xử lý CAPTCHA với Selenium và Python.

Trang thương mại điện tử chặn theo phiên đăng nhập thì sao?

Đăng nhập trước, giải CAPTCHA ngay ở bước đó, giữ cookie phiên rồi mới thu thập trang phía trong. Chỉ áp dụng với hệ thống bạn có quyền truy cập hợp lệ, ví dụ cổng bán hàng của công ty bạn.

Đọc thêm

• Xử lý CAPTCHA trong luồng scraping
• Vì sao headless browser hay dính CAPTCHA
• Cơ chế phát hiện CAPTCHA khi crawl