Scarnox · Blog

Scrapy captcha middleware: solve captcha ใน spider ของคุณ (2026)

Scrapy ถูกสร้างขึ้นรอบ request และ callback ดังนั้น captcha จึงไม่พอดีกับมันนัก วิธีที่สะอาดที่สุดคือใช้ downloader middleware มันเห็นทุกการตอบกลับ สังเกตเห็นความท้าทาย รับ token และให้ spider ส่งฟอร์มพร้อม token นั้น คู่มือนี้สร้าง middleware นั้นสำหรับหน้า PopularCaptcha ซึ่ง Scarnox solve ได้ในวันนี้แบบ proxyless

เริ่มต้นด้วยเครดิตทดลองใช้ฟรี ไม่ต้องใช้บัตรเครดิต และลองโค้ดกับหน้าทดสอบก่อนนำไปใช้กับสิ่งอื่น

middleware ทำงานอย่างไร

middleware ทำงานกับทุกการตอบกลับ:

  1. หากการตอบกลับไม่มี data-sitekey ก็ส่งการตอบกลับผ่านไปตามปกติ
  2. หากมี middleware จะ solve captcha และคืนสำเนาของ request เดิมโดยเก็บ token ไว้ใน meta
  3. Scrapy ดึง request นั้นอีกครั้ง middleware เห็น token ใน meta และปล่อยการตอบกลับผ่าน
  4. spider อ่าน token จาก meta และส่งฟอร์มด้วย FormRequest.from_response

การเรียก solve ทำงานในเธรดผ่าน deferToThread จึงไม่บล็อก reactor ของ Scrapy request อื่นยังคงเดินหน้าต่อขณะที่กำลัง solve captcha

การตั้งค่าโปรเจกต์

ใส่ key ของคุณในตัวแปรสภาพแวดล้อม ไม่ใช่ในโค้ด:

# 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

download delay และการทำงานพร้อมกันหนึ่งเดียวต่อโดเมนถูกตั้งไว้โดยเจตนา เพื่อให้ภาระต่อเว็บไซต์ต่ำและลดจำนวน captcha ที่คุณเห็น

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,  # ทำงานในเธรดของ worker ไม่ใช่ใน 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):
        # แก้ไปแล้ว หรือเป็นการส่งฟอร์มที่เราสร้างเอง: ปล่อยผ่าน
        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:
            # การคืน Request จะกำหนดเวลาให้ทำงานใหม่ dont_filter ช่วยข้าม dupe filter
            return request.replace(
                dont_filter=True,
                meta={**request.meta, "captcha_token": token},
            )

        return deferToThread(self._solve, sitekey, response.url).addCallback(retry_with_token)

Scrapy รองรับการคืน Deferred จาก process_response callback จะเปลี่ยน token ให้เป็น request ที่ถูก retry และ Scrapy จะกำหนดเวลาให้เหมือน request อื่น

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 กรอกฟิลด์ของฟอร์มเอง เราเพิ่มเฉพาะ 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(),
            }

selector เป็นตัวอย่าง ให้แทนที่ด้วย selector จริงจากเว็บไซต์ที่คุณ scrape

flag skip_captcha คือสิ่งที่หยุดไม่ให้ middleware solve การส่งฟอร์มซ้ำอีก หากไม่มี flag นี้ ทุกการส่งจะเสียค่า solve อีกครั้ง

รัน

export SCARNOX_API_KEY=scarnox_YOURAPIKEY
scrapy crawl search -O results.json

ดู log ไว้ บรรทัดเช่น PopularCaptcha on ..., solving หมายความว่าพบความท้าทายและมีการขอ token หากคุณเห็นบรรทัดนั้นทุกหน้า ให้ช้าลง ความท้าทายที่ถี่แสดงว่าเว็บไซต์สังเกตเห็นการ crawl แล้ว

เวลาของ token

token ของ PopularCaptcha หมดอายุภายในประมาณสองนาที middleware ขอให้ retry ทันทีที่ token มาถึง และ spider ส่งฟอร์มในห่วงโซ่ callback เดียวกัน ทำให้ห่วงโซ่สั้น อย่าเพิ่มการดาวน์โหลดที่ช้าหรือการประมวลผลที่ยาวระหว่าง token กับการส่งฟอร์ม สำหรับช่วงเวลานี้ดู การหมดอายุของ captcha token

Error

การ solve ที่ล้มเหลวจะ raise RuntimeError ในเธรด Scrapy จะ log ว่าเป็น request ที่ล้มเหลว การ solve ที่ล้มเหลวไม่ถูกเรียกเก็บเงิน ดังนั้นการ retry จึงไม่มีค่าใช้จ่าย สำหรับนโยบาย retry ที่มีขีดจำกัด ให้ใช้ RetryMiddleware ในตัวของ Scrapy และห่อ _solve ด้วยลูปสั้น ๆ คู่มือ การจัดการ error ของ captcha solver แสดงรูปแบบที่ใช้ได้จริงในโปรดักชัน

Scrape อย่างมีความรับผิดชอบ

สำหรับพื้นฐานทางกฎหมาย ดู web scraping ถูกกฎหมายหรือไม่

ราคา

Scarnox คิดเงินต่อการ solve ที่สำเร็จ และการ solve ที่ล้มเหลวไม่มีค่าใช้จ่าย ดู ราคาปัจจุบัน สำหรับอัตราต่าง ๆ เริ่มต้นด้วยเครดิตทดลองใช้ฟรี ไม่ต้องใช้บัตรเครดิต

ขั้นตอนถัดไป