Get past captcha pages in Scrapy crawls

import os
import time

import requests

API_URL = "https://bypass.so"
CLIENT_KEY = os.environ.get("BYPASS_API_KEY", "")


class BypassError(Exception):
    def __init__(self, code, description="", task_id=None):
        super().__init__(f"{code}: {description}")
        self.code = code
        self.description = description
        self.task_id = task_id


def _read(response):
    try:
        data = response.json()
    except ValueError:
        raise BypassError("HTTP_ERROR", f"HTTP {response.status_code} without a JSON body")
    if data.get("errorId"):
        raise BypassError(
            data.get("errorCode", "ERROR_UNKNOWN"),
            data.get("errorDescription", ""),
            data.get("taskId"),
        )
    return data


def solve(task_type, params, timeout=120, poll_every=2.0):
    if not CLIENT_KEY:
        raise BypassError("ERROR_KEY_INVALID", "set the BYPASS_API_KEY environment variable")
    body = {"clientKey": CLIENT_KEY, "type": task_type, **params}
    data = _read(requests.post(f"{API_URL}/createTask", json=body, timeout=30))
    task_id = data["taskId"]
    deadline = time.monotonic() + timeout
    while data.get("status") != "ready":
        if time.monotonic() > deadline:
            raise TimeoutError(f"task {task_id} was not ready after {timeout}s")
        time.sleep(poll_every)
        data = _read(requests.get(
            f"{API_URL}/getTaskResult/{task_id}",
            params={"clientKey": CLIENT_KEY},
            timeout=30,
        ))
    solution = dict(data.get("solution") or {})
    solution["taskId"] = task_id
    return solution


def token_of(solution):
    return solution.get("gRecaptchaResponse") or solution.get("token") or solution.get("text")


def balance():
    headers = {"Authorization": f"Bearer {CLIENT_KEY}"}
    return _read(requests.get(f"{API_URL}/balance", headers=headers, timeout=30))


def report_bad(task_id):
    headers = {"Authorization": f"Bearer {CLIENT_KEY}"}
    return _read(requests.post(
        f"{API_URL}/reportBad",
        params={"task_id": task_id},
        headers=headers,
        timeout=30,
    ))

import logging

from form2request import form2request
from scrapy.utils.defer import maybe_deferred_to_future
from twisted.internet.threads import deferToThread

from myproject.bypass_client import BypassError, solve, token_of

logger = logging.getLogger(__name__)

# task type, CSS selector of the widget, form field that receives the token
WIDGETS = [
    ("TurnstileTask", ".cf-turnstile[data-sitekey]", "cf-turnstile-response"),
    ("RecaptchaV2Task", ".g-recaptcha[data-sitekey]", "g-recaptcha-response"),
]


class BypassCaptchaMiddleware:
    """Detects captcha pages, solves them with bypass.so and resubmits the form."""

    def __init__(self, settings):
        self.max_attempts = settings.getint("BYPASS_MAX_ATTEMPTS", 2)
        self.math_selector = settings.get("BYPASS_MATH_SELECTOR", "#captcha-question::text")
        self.image_selector = settings.get("BYPASS_IMAGE_SELECTOR", "img.captcha::attr(src)")
        self.answer_field = settings.get("BYPASS_ANSWER_FIELD", "captcha")

    @classmethod
    def from_crawler(cls, crawler):
        return cls(crawler.settings)

    def find_captcha(self, response):
        """Return (task_type, params, field, form_xpath) for a captcha page, else None."""
        if not hasattr(response, "css"):  # images, PDFs and other binary bodies
            return None
        for task_type, selector, field in WIDGETS:
            sitekey = response.css(selector).attrib.get("data-sitekey")
            if sitekey:
                params = {"websiteURL": response.url, "websiteKey": sitekey}
                return task_type, params, field, "//form[.//*[@data-sitekey]]"
        answer_form = f'//form[.//*[@name="{self.answer_field}"]]'
        question = response.css(self.math_selector).get()
        if question:
            return "MathCaptchaTask", {"captchaText": question.strip()}, self.answer_field, answer_form
        src = response.css(self.image_selector).get()
        if src and src.startswith("data:image/"):
            image_b64 = src.split(",", 1)[1]
            return "ImageToTextTask", {"imageBase64": image_b64}, self.answer_field, answer_form
        return None

    async def process_response(self, request, response, spider=None):
        found = self.find_captcha(response)
        if found is None:
            return response
        attempt = request.meta.get("bypass_attempt", 0) + 1
        if attempt > self.max_attempts:
            logger.warning("captcha still shown after %d attempts: %s", self.max_attempts, response.url)
            return response

        task_type, params, field, form_xpath = found
        try:
            # solve() blocks while it polls, so run it on the reactor thread pool.
            # Awaiting the Deferred pauses only this response; the crawl keeps going.
            solution = await maybe_deferred_to_future(deferToThread(solve, task_type, params))
        except (BypassError, TimeoutError) as err:
            logger.error("could not solve %s on %s: %s", task_type, response.url, err)
            return response
        logger.info("solved %s on %s (task %s)", task_type, response.url, solution["taskId"])

        form = response.xpath(form_xpath)
        if not form:
            logger.error("no form to submit the answer on %s", response.url)
            return response
        # Fill the answer into the page's own form and send it like a browser would.
        return form2request(form, {field: token_of(solution)}).to_scrapy(
            callback=request.callback,
            errback=request.errback,
            cb_kwargs=request.cb_kwargs,
            meta={**request.meta, "bypass_attempt": attempt, "bypass_task_id": solution["taskId"]},
            priority=request.priority + 1,
            dont_filter=True,
        )

"""Local test site: / shows a math captcha until the right answer is posted."""
import sys
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import parse_qs

CAPTCHA = b"""<!doctype html><html><body><h1>Are you human?</h1>
<form method="post" action="/"><label>Solve: <span id="captcha-question">3+4</span></label>
<input name="captcha"><input type="hidden" name="next" value="/"><button>Continue</button></form></body></html>"""
PRODUCTS = b"""<!doctype html><html><body>
<div class="product"><h2>Blue mug</h2><span class="price">9.00</span></div>
<div class="product"><h2>Red mug</h2><span class="price">11.00</span></div></body></html>"""


class H(BaseHTTPRequestHandler):
    def send(self, code, body=b"", headers=()):
        self.send_response(code)
        for k, v in headers:
            self.send_header(k, v)
        self.send_header("Content-Type", "text/html")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path == "/robots.txt":
            return self.send(404)
        ok = "human=1" in (self.headers.get("Cookie") or "")
        print("SERVER GET", self.path, "->", "products" if ok else "captcha page", flush=True)
        self.send(200, PRODUCTS if ok else CAPTCHA)

    def do_POST(self):
        form = parse_qs(self.rfile.read(int(self.headers["Content-Length"])).decode())
        answer = form.get("captcha", [""])[0]
        print("SERVER POST / captcha=%r" % answer, flush=True)
        if answer == "7":
            self.send(302, headers=[("Location", "/"), ("Set-Cookie", "human=1; Path=/")])
        else:
            self.send(200, CAPTCHA)

    def log_message(self, *a):
        pass


HTTPServer(("127.0.0.1", int(sys.argv[1])), H).serve_forever()