Hệ thống phần mềm trung gian của Scrapy cho phép bạn chặn các yêu cầu và phản hồi. Dưới đây là cách xây dựng phần mềm trung gian của trình tải xuống tự động giải CAPTCHA bằng CaptchaAI.
Phần mềm trung gian của trình tải xuống
# middlewares.py
import re
import time
import logging
import requests as http_requests
from scrapy import signals
from scrapy.http import HtmlResponse
logger = logging.getLogger(__name__)
class CaptchaAIMiddleware:
"""Scrapy downloader middleware for automatic CAPTCHA solving."""
CAPTCHA_PATTERNS = [
(r'data-sitekey="([^"]+)"', "recaptcha"),
(r"cf-turnstile.*?data-sitekey=\"([^\"]+)\"", "turnstile"),
]
def __init__(self, api_key, max_retries=2):
self.api_key = api_key
self.max_retries = max_retries
self.stats = {"detected": 0, "solved": 0, "failed": 0}
@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")
middleware = cls(
api_key=api_key,
max_retries=crawler.settings.getint("CAPTCHAAI_MAX_RETRIES", 2),
)
crawler.signals.connect(middleware.spider_closed, signal=signals.spider_closed)
return middleware
def process_response(self, request, response, spider):
"""Check response for CAPTCHA and solve if found."""
if not isinstance(response, HtmlResponse):
return response
body = response.text
for pattern, captcha_type in self.CAPTCHA_PATTERNS:
match = re.search(pattern, body)
if match:
sitekey = match.group(1)
self.stats["detected"] += 1
logger.info(
f"CAPTCHA ({captcha_type}) on {response.url}, solving..."
)
retries = request.meta.get("captcha_retries", 0)
if retries >= self.max_retries:
self.stats["failed"] += 1
logger.error(f"Max CAPTCHA retries on {response.url}")
return response
token = self._solve(captcha_type, sitekey, response.url)
if token:
self.stats["solved"] += 1
# Re-request with token
new_request = request.copy()
new_request.meta["captcha_retries"] = retries + 1
new_request.meta["captcha_token"] = token
new_request.method = "POST"
new_request.body = f"g-recaptcha-response={token}"
new_request.headers[b"Content-Type"] = b"application/x-www-form-urlencoded"
new_request.dont_filter = True
return new_request
else:
self.stats["failed"] += 1
return response
def _solve(self, captcha_type, sitekey, pageurl):
"""Solve CAPTCHA via CaptchaAI."""
method_map = {
"recaptcha": {"method": "userrecaptcha", "googlekey": sitekey},
"turnstile": {"method": "turnstile", "sitekey": sitekey},
}
params = method_map.get(captcha_type)
if not params:
return None
try:
resp = http_requests.post("https://ocr.captchaai.com/in.php", data={
"key": self.api_key,
"pageurl": pageurl,
"json": 1,
**params,
}, timeout=30)
result = resp.json()
if result.get("status") != 1:
logger.error(f"Submit error: {result.get('request')}")
return None
task_id = result["request"]
time.sleep(10)
for _ in range(24):
resp = http_requests.get("https://ocr.captchaai.com/res.php", params={
"key": self.api_key, "action": "get",
"id": task_id, "json": 1,
}, timeout=15)
data = resp.json()
if data.get("status") == 1:
return data["request"]
if data["request"] != "CAPCHA_NOT_READY":
return None
time.sleep(5)
except Exception as e:
logger.error(f"Solve exception: {e}")
return None
def spider_closed(self, spider):
"""Log CAPTCHA statistics on spider close."""
logger.info(
f"CAPTCHA Stats — Detected: {self.stats['detected']}, "
f"Solved: {self.stats['solved']}, "
f"Failed: {self.stats['failed']}"
)
Cài đặt vụn vặt
# settings.py
# CaptchaAI configuration
CAPTCHAAI_API_KEY = "YOUR_API_KEY" # Better: use env variable
CAPTCHAAI_MAX_RETRIES = 2
# Enable the middleware
DOWNLOADER_MIDDLEWARES = {
"myproject.middlewares.CaptchaAIMiddleware": 600,
}
# Increase timeouts for CAPTCHA solving
DOWNLOAD_TIMEOUT = 180
# Rate limiting
DOWNLOAD_DELAY = 3
RANDOMIZE_DOWNLOAD_DELAY = True
CONCURRENT_REQUESTS = 4
CONCURRENT_REQUESTS_PER_DOMAIN = 2
Ví dụ về nhện
# spiders/product_spider.py
import scrapy
class ProductSpider(scrapy.Spider):
name = "products"
start_urls = ["https://example.com/products"]
def parse(self, response):
"""Parse product listing page."""
# The middleware handles CAPTCHAs automatically
# This method only deals with parsing
for product in response.css("div.product-card"):
yield {
"name": product.css(".name::text").get("").strip(),
"price": product.css(".price::text").get("").strip(),
"url": response.urljoin(product.css("a::attr(href)").get("")),
}
# Follow pagination
next_page = response.css("a.next-page::attr(href)").get()
if next_page:
yield scrapy.Request(response.urljoin(next_page))
def parse_product(self, response):
"""Parse individual product page."""
# Access CAPTCHA token if middleware solved one
token = response.meta.get("captcha_token")
if token:
self.logger.info(f"Page accessed after CAPTCHA solve: {response.url}")
yield {
"title": response.css("h1::text").get("").strip(),
"description": response.css(".description::text").get("").strip(),
"price": response.css(".price::text").get("").strip(),
}
Phần mềm trung gian truyền mã thông báo
Đối với các trình thu thập thông tin cần mã thông báo trong phương thức phân tích cú pháp:
class CaptchaTokenSpiderMiddleware:
"""Pass CAPTCHA tokens to spider callbacks."""
def process_spider_input(self, response, spider):
"""Add CAPTCHA token to response meta if available."""
token = response.meta.get("captcha_token")
if token:
spider.logger.debug(f"CAPTCHA token available for {response.url}")
return None
def process_spider_output(self, response, result, spider):
"""Forward token to new requests from this response."""
token = response.meta.get("captcha_token")
for item_or_request in result:
if isinstance(item_or_request, scrapy.Request) and token:
item_or_request.meta.setdefault("parent_captcha_token", token)
yield item_or_request
Câu hỏi thường gặp
Phần mềm trung gian có chặn vòng lặp sự kiện của Scrapy không?
Có, các lệnh gọi HTTP đồng bộ tới khối API của CaptchaAI trong quá trình bỏ phiếu. Đối với các trình thu thập thông tin có tính đồng thời cao, thay vào đó hãy cân nhắc sử dụng scrapy-playwright với tính năng giải CAPTCHA không đồng bộ.
Tôi có thể sử dụng phần mềm trung gian này với Scrapy-Splash không?
Vâng. Phần mềm trung gian chặn các phản hồi bất kể chúng được hiển thị như thế nào. Nó hoạt động với các phản hồi Splash, Playwright và HTTP tiêu chuẩn.
Làm cách nào để kiểm tra phần mềm trung gian?
Sử dụng trình trợ giúp fake_response của Scrapy trong các bài kiểm tra đơn vị. Mô phỏng phản hồi API CaptchaAI để kiểm tra logic phát hiện CAPTCHA và thử lại mà không thực hiện lệnh gọi API thực.
Hướng dẫn liên quan
Thêm CaptchaAI vào Scrapy —lấy khóa API của bạn.
