Tích Hợp

Hướng dẫn tích hợp Scrapy + CaptchaAI

Scrapy là khung thu thập dữ liệu Python phổ biến nhất. Hướng dẫn này chỉ ra cách thêm giải pháp CaptchaAI CAPTCHA vào trình thu thập thông tin của bạn bằng cách sử dụng phần mềm trung gian tùy chỉnh.

Yêu cầu

Yêu cầu Chi tiết
Python 3,8+
vụn vặt 2,5+
yêu cầu Đối với lệnh gọi API CaptchaAI
Khóa API CaptchaAI Nhận một cái ở đây
pip install scrapy requests

Mô-đun bộ giải CaptchaAI

Tạo captcha_solver.py trong thư mục gốc dự án Scrapy của bạn:

import requests
import time

class CaptchaAISolver:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"

    def solve_recaptcha(self, site_key, page_url, timeout=300):
        resp = requests.get(f"{self.base_url}/in.php", params={
            "key": self.api_key,
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": page_url,
        })

        if not resp.text.startswith("OK|"):
            raise Exception(f"Submit failed: {resp.text}")

        task_id = resp.text.split("|")[1]
        deadline = time.time() + timeout

        while time.time() < deadline:
            time.sleep(5)
            result = requests.get(f"{self.base_url}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
            })

            if result.text == "CAPCHA_NOT_READY":
                continue
            if result.text.startswith("OK|"):
                return result.text.split("|", 1)[1]
            raise Exception(f"Solve failed: {result.text}")

        raise TimeoutError(f"Task {task_id} timed out")

    def solve_image(self, image_base64, timeout=120):
        resp = requests.get(f"{self.base_url}/in.php", params={
            "key": self.api_key,
            "method": "base64",
            "body": image_base64,
        })

        if not resp.text.startswith("OK|"):
            raise Exception(f"Submit failed: {resp.text}")

        task_id = resp.text.split("|")[1]
        deadline = time.time() + timeout

        while time.time() < deadline:
            time.sleep(5)
            result = requests.get(f"{self.base_url}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
            })

            if result.text == "CAPCHA_NOT_READY":
                continue
            if result.text.startswith("OK|"):
                return result.text.split("|", 1)[1]
            raise Exception(f"Solve failed: {result.text}")

        raise TimeoutError(f"Task {task_id} timed out")

Phần mềm trung gian phế liệu

Tạo middlewares.py:

import base64
import re
from scrapy import signals
from scrapy.http import HtmlResponse
from captcha_solver import CaptchaAISolver

class CaptchaAIMiddleware:
    """Scrapy downloader middleware that detects and solves CAPTCHAs."""

    def __init__(self, api_key):
        self.solver = CaptchaAISolver(api_key)

    @classmethod
    def from_crawler(cls, crawler):
        api_key = crawler.settings.get("CAPTCHAAI_API_KEY")
        if not api_key:
            raise ValueError("CAPTCHAAI_API_KEY setting is required")
        return cls(api_key)

    def process_response(self, request, response, spider):
        # Check for reCAPTCHA on the page
        site_key = self._find_recaptcha_key(response.text)
        if site_key:
            spider.logger.info(f"reCAPTCHA detected on {response.url}")
            token = self.solver.solve_recaptcha(site_key, response.url)
            request.meta["captcha_token"] = token
            spider.logger.info("CAPTCHA solved successfully")

        # Check for image CAPTCHA
        captcha_img = self._find_image_captcha(response)
        if captcha_img:
            spider.logger.info(f"Image CAPTCHA detected on {response.url}")
            text = self.solver.solve_image(captcha_img)
            request.meta["captcha_text"] = text
            spider.logger.info(f"Image CAPTCHA solved: {text}")

        return response

    def _find_recaptcha_key(self, html):
        match = re.search(
            r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', html
        )
        return match.group(1) if match else None

    def _find_image_captcha(self, response):
        img = response.css("img#captcha-image::attr(src)").get()
        if img and img.startswith("data:image"):
            return img.split(",", 1)[1]
        return None

Cấu hình cài đặt

Thêm vào settings.py:

import os

CAPTCHAAI_API_KEY = os.environ.get("CAPTCHAAI_API_KEY")

DOWNLOADER_MIDDLEWARES = {
    "myproject.middlewares.CaptchaAIMiddleware": 560,
}

Ví dụ về nhện

import scrapy

class ProductSpider(scrapy.Spider):
    name = "products"
    start_urls = ["https://example.com/products"]

    def parse(self, response):
        # If CAPTCHA was solved, the token is in meta
        token = response.meta.get("captcha_token")
        if token:
            # Resubmit the page with the token
            yield scrapy.FormRequest(
                url=response.url,
                formdata={"g-recaptcha-response": token},
                callback=self.parse_products,
            )
        else:
            yield from self.parse_products(response)

    def parse_products(self, response):
        for product in response.css(".product-item"):
            yield {
                "name": product.css("h2::text").get(),
                "price": product.css(".price::text").get(),
                "url": response.urljoin(
                    product.css("a::attr(href)").get()
                ),
            }

        next_page = response.css("a.next-page::attr(href)").get()
        if next_page:
            yield scrapy.Request(response.urljoin(next_page))

Thử lại trên các trang CAPTCHA

Thêm tính năng tự động thử lại khi CAPTCHA xuất hiện:

class CaptchaRetryMiddleware:
    """Retry requests that return CAPTCHA challenge pages."""

    max_retries = 3

    def process_response(self, request, response, spider):
        if self._is_captcha_page(response):
            retries = request.meta.get("captcha_retries", 0)
            if retries < self.max_retries:
                request.meta["captcha_retries"] = retries + 1
                spider.logger.info(
                    f"CAPTCHA page detected, retry {retries + 1}"
                )
                return request.copy()

        return response

    def _is_captcha_page(self, response):
        indicators = [
            "g-recaptcha",
            "cf-turnstile",
            "captcha-image",
            "Please verify you are human",
        ]
        return any(ind in response.text for ind in indicators)

Chạy nhện

export CAPTCHAAI_API_KEY="YOUR_API_KEY"
scrapy crawl products -o products.json

Khắc phục sự cố

Vấn đề Nguyên nhân Cách xử lý
ValueError: CAPTCHAAI_API_KEY setting is required Thiếu biến env Đặt CAPTCHAAI_API_KEY
CAPTCHA không được phát hiện Cấu trúc HTML khác nhau Cập nhật mẫu biểu thức chính quy trong phần mềm trung gian
TimeoutError khi giải quyết Giải quyết chậm hoặc mạng Tăng thời gian chờ trong bộ giải
Nhện bị chặn sau khi giải quyết Chặn dựa trên IP Thêm phần mềm trung gian xoay proxy

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

Tôi có thể sử dụng cái này với Scrapy-Splash hoặc Scrapy-Playwright không?

Vâng. Đối với các trang được hiển thị bằng JavaScript, phần mềm trung gian hoạt động theo cách tương tự - nó kiểm tra phản hồi HTML cuối cùng cho các phần tử CAPTCHA.

Phần mềm trung gian có làm chậm quá trình thu thập thông tin không?

Việc giải CAPTCHA mất 5-15 giây mỗi trang. Sử dụng CONCURRENT_REQUESTS để thu thập dữ liệu các trang khác trong khi chờ đợi. Chỉ những trang có CAPTCHA mới gây ra sự chậm trễ.

Làm cách nào để xử lý các loại CAPTCHA khác nhau trên mỗi trang?

Mở rộng phương thức process_response của phần mềm trung gian để kiểm tra Turnstile, GeeTest hoặc các loại khác và gọi phương thức giải thích hợp.

Hướng dẫn liên quan

Os comentários estão desativados para este artigo.