跳转到内容
YesCaptchaYesCaptcha 文档

Python

Markdown 版在 ChatGPT 中打开在 Claude 中打开

运行环境：Python 3.8 以上，pip install requests。运行前设置环境变量：

终端窗口
export YESCAPTCHA_CLIENT_KEY=你的clientKey

查询余额

Section titled “查询余额”

不扣点，用来确认 clientKey 正确：

get_balance.py
"""Check the balance. getBalance is free and is a quick way to verify your clientKey."""
import os


import requests


CLIENT_KEY = os.environ["YESCAPTCHA_CLIENT_KEY"]


r = requests.post(
    "https://api.yescaptcha.com/getBalance",
    json={"clientKey": CLIENT_KEY},
    timeout=30,
)
data = r.json()
if data.get("errorId"):
    # e.g. ERROR_KEY_DOES_NOT_EXIST when the clientKey is wrong
    print("error:", data.get("errorCode"), data.get("errorDescription"))
else:
    print("balance (points):", data["balance"])

识别一张图片验证码

Section titled “识别一张图片验证码”

同步类型 ImageToTextTaskOcrBase，一次请求拿到结果。示例图片：captcha.jpg。

quickstart.py
"""First request: read an alphanumeric image captcha with ImageToTextTaskOcrBase
(synchronous, 2 points per image).


Usage:
    export YESCAPTCHA_CLIENT_KEY=your_client_key
    python quickstart.py captcha.png
"""
import base64
import os
import sys


import requests


CLIENT_KEY = os.environ["YESCAPTCHA_CLIENT_KEY"]
image_path = sys.argv[1] if len(sys.argv) > 1 else "captcha.png"


with open(image_path, "rb") as f:
    body = base64.b64encode(f.read()).decode()  # no "data:" prefix


r = requests.post(
    "https://api.yescaptcha.com/createTask",
    json={
        "clientKey": CLIENT_KEY,
        "task": {"type": "ImageToTextTaskOcrBase", "body": body},
    },
    timeout=30,
)
data = r.json()
if data.get("errorId"):
    raise SystemExit(f"{data.get('errorCode')}: {data.get('errorDescription')}")
print("text:", data["solution"]["text"])

可复用的客户端

Section titled “可复用的客户端”

把下面的文件复制进你的项目，同步和异步类型都能用：

yescaptcha.py
"""A small client you can copy into your project: create tasks, poll results, check the balance.


    from yescaptcha import YesCaptcha
    client = YesCaptcha(os.environ["YESCAPTCHA_CLIENT_KEY"])
    solution = client.solve({
        "type": "NoCaptchaTaskProxyless",
        "websiteURL": "https://www.google.com/recaptcha/api2/demo",
        "websiteKey": "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
    })
    print(solution["gRecaptchaResponse"])
"""
import os
import time


import requests




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




class YesCaptcha:
    def __init__(self, client_key, endpoint="https://api.yescaptcha.com", poll_interval=5, timeout=180):
        self.client_key = client_key
        self.endpoint = endpoint.rstrip("/")
        self.poll_interval = poll_interval  # 5 s: polling the same taskId more often only returns "processing"
        self.timeout = timeout
        self.session = requests.Session()


    def call(self, method, payload=None):
        body = {"clientKey": self.client_key, **(payload or {})}
        data = self.session.post(f"{self.endpoint}/{method}", json=body, timeout=30).json()
        if data.get("errorId"):
            raise YesCaptchaError(data.get("errorCode"), data.get("errorDescription"))
        return data


    def balance(self):
        return self.call("getBalance")["balance"]


    def solve(self, task):
        created = self.call("createTask", {"task": task})
        if created.get("status") == "ready":  # synchronous types return the solution directly
            return created["solution"]
        task_id = created["taskId"]
        deadline = time.time() + self.timeout
        while time.time() < deadline:
            time.sleep(self.poll_interval)
            result = self.call("getTaskResult", {"taskId": task_id})
            if result.get("status") == "ready":
                return result["solution"]
        raise TimeoutError(f"task {task_id} not ready after {self.timeout}s")




if __name__ == "__main__":
    client = YesCaptcha(os.environ["YESCAPTCHA_CLIENT_KEY"])
    print("balance:", client.balance())
from yescaptcha import YesCaptcha


client = YesCaptcha(os.environ["YESCAPTCHA_CLIENT_KEY"])
solution = client.solve({
    "type": "HCaptchaTaskProxyless",
    "websiteURL": "https://accounts.hcaptcha.com/demo",
    "websiteKey": "a5f74b19-9e45-40e0-b45d-47ff91b7a6c2",
})
print(solution["gRecaptchaResponse"])

出错时抛出 YesCaptchaError，e.code 是错误码，含义见 错误码。

控制并发

Section titled “控制并发”

批量发任务时限制同时在处理中的请求数，收到 ERROR_THREADS_MAXIMUM 稍等再试（需要 pip install aiohttp），说明见 并发限制与自动暂停：

concurrency.py
"""Limit concurrency: keep the number of requests waiting for a result below your limit, and back off
on ERROR_THREADS_MAXIMUM (rejected requests are not charged). Requires: pip install aiohttp"""
import asyncio
import os


import aiohttp


CLIENT_KEY = os.environ["YESCAPTCHA_CLIENT_KEY"]
LIMIT = 15  # below your concurrency limit
sem = asyncio.Semaphore(LIMIT)




async def solve(session, task):
    async with sem:  # at most LIMIT requests in progress at a time
        result = {}
        for attempt in range(5):
            async with session.post(
                "https://api.yescaptcha.com/createTask",
                json={"clientKey": CLIENT_KEY, "task": task},
            ) as resp:
                result = await resp.json()
            if result.get("errorCode") != "ERROR_THREADS_MAXIMUM":
                return result
            await asyncio.sleep(1 + attempt)  # over the limit: wait and retry
        return result




async def main(tasks):
    async with aiohttp.ClientSession() as session:
        return await asyncio.gather(*(solve(session, t) for t in tasks))




if __name__ == "__main__":
    print(asyncio.run(main([])))

提交 reCAPTCHA token

Section titled “提交 reCAPTCHA token”