Middleware captcha cho Scrapy: giải captcha trong spider (2026)
Scrapy được xây quanh request và callback, nên captcha không khớp gọn gàng vào nó. Cách sạch để xử lý là một downloader middleware: nó thấy mọi phản hồi, nhận ra một thử thách, lấy token, và để spider gửi form với token đó. Hướng dẫn này xây dựng middleware cho các trang PopularCaptcha, loại mà Scarnox giải hiện nay theo kiểu proxyless.
Bắt đầu với tín dụng dùng thử miễn phí, không cần thẻ và thử code trên một trang thử nghiệm trước khi áp dụng vào bất kỳ thứ gì khác.
Middleware hoạt động thế nào
Middleware chạy trên mọi phản hồi:
- Nếu phản hồi không có
data-sitekey, nó cho phản hồi đi qua. - Nếu có, middleware giải captcha và trả về một bản sao của cùng request với token được lưu trong
meta. - Scrapy tải lại request đó. Middleware thấy token trong
metavà cho phản hồi đi qua. - Spider đọc token từ
metavà gửi form bằngFormRequest.from_response.
Lệnh gọi giải chạy trong một thread qua deferToThread, nên nó không chặn reactor của Scrapy. Các request khác vẫn tiếp tục trong khi captcha đang được giải.
Cấu hình dự án
Đặt khóa của bạn trong biến môi trường, không trong code:
# settings.py
import os
SCARNOX_API_KEY = os.environ["SCARNOX_API_KEY"]
SCARNOX_TIMEOUT = 185
DOWNLOADER_MIDDLEWARES = {
"myproject.middlewares.PopularCaptchaMiddleware": 543,
}
ROBOTSTXT_OBEY = True
DOWNLOAD_DELAY = 3
CONCURRENT_REQUESTS_PER_DOMAIN = 1
Độ trễ tải và đồng thời một request mỗi domain được đặt có chủ đích. Chúng giữ tải lên trang thấp và giảm số captcha bạn gặp.
Middleware
# myproject/middlewares.py
import requests
from scrapy import Request
from scrapy.http import Response
from twisted.internet.threads import deferToThread
API_URL = "https://api.scarnox.com/api/tasks/create"
class PopularCaptchaMiddleware:
def __init__(self, api_key: str, timeout: int):
self.api_key = api_key
self.timeout = timeout
@classmethod
def from_crawler(cls, crawler):
return cls(
api_key=crawler.settings.get("SCARNOX_API_KEY"),
timeout=crawler.settings.getint("SCARNOX_TIMEOUT", 185),
)
def _solve(self, sitekey: str, page_url: str) -> str:
r = requests.post(
API_URL,
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"captcha_type": "PopularCaptchaTaskProxyless",
"site_key": sitekey,
"site_url": page_url,
},
timeout=self.timeout, # chạy trong worker thread, không phải reactor
)
data = r.json()
if data.get("errorId") != 0:
raise RuntimeError(f"{data.get('errorCode')}: {data.get('errorDescription')}")
return data["solution"]["token"]
def process_response(self, request: Request, response: Response, spider):
# Đã giải, hoặc là form submit do chúng ta tự tạo: cho đi qua.
if request.meta.get("captcha_token") or request.meta.get("skip_captcha"):
return response
sitekey = response.css("[data-sitekey]::attr(data-sitekey)").get()
if not sitekey:
return response
spider.logger.info("PopularCaptcha on %s, solving", response.url)
def retry_with_token(token: str) -> Request:
# Trả về Request sẽ lên lịch lại nó. dont_filter tránh bộ lọc trùng lặp.
return request.replace(
dont_filter=True,
meta={**request.meta, "captcha_token": token},
)
return deferToThread(self._solve, sitekey, response.url).addCallback(retry_with_token)
Scrapy hỗ trợ trả về một Deferred từ process_response. Callback biến token thành một request được retry, và Scrapy lên lịch nó như mọi request khác.
Spider
# myproject/spiders/search.py
import scrapy
from scrapy.http import FormRequest
class SearchSpider(scrapy.Spider):
name = "search"
start_urls = ["https://search.example.com/search?q=web+scraping"]
def parse(self, response):
token = response.meta.get("captcha_token")
if token and response.css("[data-sitekey]"):
# Scrapy điền các trường riêng của form; chúng ta thêm token.
yield FormRequest.from_response(
response,
formdata={
"h-captcha-response": token,
"g-recaptcha-response": token,
},
callback=self.parse_results,
meta={"skip_captcha": True},
dont_filter=True,
)
return
yield from self.parse_results(response)
def parse_results(self, response):
for item in response.css("div.result"):
yield {
"title": item.css("h3::text").get(),
"url": item.css("a::attr(href)").get(),
}
Các selector chỉ là chỗ giữ chỗ. Thay chúng bằng selector thật từ trang bạn đang scrape.
Cờ skip_captcha là thứ ngăn middleware giải lại việc gửi form. Nếu không có nó, mỗi lần gửi sẽ tốn thêm một lần giải.
Chạy
export SCARNOX_API_KEY=scarnox_YOURAPIKEY
scrapy crawl search -O results.json
Theo dõi log. Một dòng như PopularCaptcha on ..., solving nghĩa là đã tìm thấy thử thách và token đã được yêu cầu. Nếu bạn thấy dòng đó ở mọi trang, hãy chậm lại. Captcha xuất hiện thường xuyên là dấu hiệu trang đã nhận ra crawl của bạn.
Thời gian của token
Token PopularCaptcha hết hạn trong khoảng hai phút. Middleware yêu cầu retry ngay khi token về, và spider gửi trong cùng chuỗi callback. Hãy giữ chuỗi đó ngắn. Đừng thêm tải xuống chậm hoặc xử lý dài giữa lúc có token và lúc gửi form. Để biết thêm về khoảng thời gian này, xem token captcha hết hạn.
Lỗi
Một lần giải thất bại sẽ ném RuntimeError trong thread. Scrapy sẽ ghi log nó như một request thất bại. Lần giải thất bại không bị tính phí, nên retry rất rẻ. Với chính sách retry có giới hạn, hãy dùng RetryMiddleware tích hợp của Scrapy và bọc _solve trong một vòng lặp ngắn. Bài viết về xử lý lỗi khi giải captcha liệt kê các mẫu đã được kiểm chứng trong môi trường production.
Scrape có trách nhiệm
- Tuân thủ robots.txt.
ROBOTSTXT_OBEY = Trueở trên làm điều này. Hãy giữ nó bật. - Giữ độ trễ cao. Một thử thách là dấu hiệu trang muốn bạn chậm lại. Đừng lách độ trễ để xong nhanh hơn.
- Đọc điều khoản dịch vụ trước khi crawl. Một số trang cấm truy cập tự động hoàn toàn.
- Xử lý dữ liệu cá nhân cẩn thận. Chỉ lưu các trường bạn cần, và bảo vệ những gì bạn giữ lại.
Về nền tảng pháp lý, xem web scraping có hợp pháp không.
Giá
Scarnox tính phí theo mỗi lần giải thành công, và lần giải thất bại thì miễn phí. Xem giá hiện tại để biết mức phí. Bắt đầu với tín dụng dùng thử miễn phí, không cần thẻ.