Solve captchas from Python with one function

python -m pip install requests

export BYPASS_API_KEY="your-api-key"

$env:BYPASS_API_KEY = "your-api-key"

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 time

import requests

from bypass_client import BypassError, solve

RETRYABLE = {"ERROR_CAPTCHA_UNSOLVABLE", "ERROR_INTERNAL", "HTTP_ERROR"}


def solve_with_retry(task_type, params, attempts=3):
    for attempt in range(1, attempts + 1):
        try:
            return solve(task_type, params)
        except BypassError as err:
            if err.code not in RETRYABLE or attempt == attempts:
                raise
            print(f"attempt {attempt}: {err.code}, retrying")
        except (TimeoutError, requests.RequestException) as err:
            if attempt == attempts:
                raise
            print(f"attempt {attempt}: {err!r}, retrying")
        time.sleep(2 * attempt)


if __name__ == "__main__":
    print(solve_with_retry("MathCaptchaTask", {"captchaText": "12 + 30 = ?"}))

from bypass_client import balance, report_bad


def flag_wrong_answer(solution):
    result = report_bad(solution["taskId"])
    print("refunded units:", result["refunded"], "balance now:", result["balanceUsd"])


if __name__ == "__main__":
    info = balance()
    print(f"balance: ${info['balanceUsd']:.2f}")

import base64
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

from bypass_client import BypassError, solve


def read_one(path):
    image_b64 = base64.b64encode(path.read_bytes()).decode()
    return path.name, solve("ImageToTextTask", {"imageBase64": image_b64})["text"]


files = sorted(Path("captchas").glob("*.png"))
with ThreadPoolExecutor(8) as pool:
    jobs = [pool.submit(read_one, p) for p in files]
    for done in as_completed(jobs):
        try:
            name, text = done.result()
            print(f"{name}: {text}")
        except BypassError as err:
            print("failed:", err)