AWS Lambda + CaptchaAI：无服务器验证码解决

AWS Lambda 无需管理服务器即可处理验证码解决。仅在求解、自动扩展以及与 API Gateway、SQS 或 Step Functions 集成时付费。

拉姆达处理器

# lambda_function.py
import json
import os
import time
import urllib.request
import urllib.parse


def lambda_handler(event, context):
    """AWS Lambda handler for CaptchaAI solving."""
    api_key = os.environ["CAPTCHAAI_KEY"]

    # Parse input
    body = json.loads(event.get("body", "{}")) if isinstance(event.get("body"), str) else event

    method = body.get("method", "userrecaptcha")
    params = body.get("params", {})

    try:
        token = solve_captcha(api_key, method, params)
        return {
            "statusCode": 200,
            "body": json.dumps({"token": token}),
        }
    except Exception as e:
        return {
            "statusCode": 500,
            "body": json.dumps({"error": str(e)}),
        }


def solve_captcha(api_key, method, params, timeout=90):
    """Solve CAPTCHA using CaptchaAI API."""
    # Submit task
    submit_data = urllib.parse.urlencode({
        "key": api_key,
        "method": method,
        "json": 1,
        **params,
    }).encode()

    req = urllib.request.Request(
        "https://ocr.captchaai.com/in.php",
        data=submit_data,
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        result = json.loads(resp.read())

    if result.get("status") != 1:
        raise RuntimeError(f"Submit error: {result.get('request')}")

    task_id = result["request"]

    # Poll for result
    start = time.time()
    while time.time() - start < timeout:
        time.sleep(5)
        poll_url = (
            f"https://ocr.captchaai.com/res.php"
            f"?key={api_key}&action=get&id={task_id}&json=1"
        )
        with urllib.request.urlopen(poll_url, timeout=15) as resp:
            data = json.loads(resp.read())

        if data["request"] != "CAPCHA_NOT_READY":
            if data.get("status") == 1:
                return data["request"]
            raise RuntimeError(f"Solve error: {data['request']}")

    raise TimeoutError("Solve timeout")

使用 Secrets Manager 保护 API 密钥

import json
import boto3


def get_api_key():
    """Retrieve CaptchaAI key from AWS Secrets Manager."""
    client = boto3.client("secretsmanager")
    response = client.get_secret_value(SecretId="captchaai/api-key")
    secret = json.loads(response["SecretString"])
    return secret["api_key"]

存储秘密：

aws secretsmanager create-secret \
  --name captchaai/api-key \
  --secret-string '{"api_key":"YOUR_API_KEY"}'

SAM 模板（基础设施即代码）

# template.yaml
AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31

Globals:
  Function:
    Timeout: 120
    MemorySize: 256
    Runtime: python3.11

Resources:
  CaptchaSolverFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: lambda_function.lambda_handler
      Environment:
        Variables:
          CAPTCHAAI_KEY: !Sub "{{resolve:secretsmanager:captchaai/api-key:SecretString:api_key}}"
      Events:
        SolveApi:
          Type: Api
          Properties:
            Path: /solve
            Method: post
      Policies:

        - AWSSecretsManagerGetSecretValuePolicy:
            SecretArn: !Sub "arn:aws:secretsmanager:${AWS::Region}:${AWS::AccountId}:secret:captchaai/api-key-*"

Outputs:
  SolveApiUrl:
    Value: !Sub "https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/solve"

部署

# Build and deploy
sam build
sam deploy --guided

# Test
curl -X POST https://YOUR_API_ID.execute-api.us-east-1.amazonaws.com/Prod/solve \
  -H "Content-Type: application/json" \
  -d '{
    "method": "userrecaptcha",
    "params": {
      "googlekey": "SITE_KEY",
      "pageurl": "https://example.com"
    }
  }'

SQS 触发的批处理

处理来自 SQS 队列的 CAPTCHA 任务：

import json
import os
import time
import urllib.request
import urllib.parse


def sqs_handler(event, context):
    """Process CAPTCHA tasks from SQS queue."""
    api_key = os.environ["CAPTCHAAI_KEY"]
    results = []

    for record in event["Records"]:
        task = json.loads(record["body"])
        try:
            token = solve_captcha(
                api_key,
                task["method"],
                task["params"],
            )
            results.append({
                "task_id": task.get("id"),
                "status": "success",
                "token": token[:50],
            })
        except Exception as e:
            results.append({
                "task_id": task.get("id"),
                "status": "error",
                "error": str(e),
            })

    return {"results": results}

拉姆达注意事项

故障排除

常问问题

Lambda 对于验证码解决具有成本效益吗？

是的。每次调用（256MB，60 秒）的费用约为 0.0001 美元，除了 CaptchaAI API 费用之外，Lambda 增加的成本可以忽略不计。您可以避免空闲期间的服务器成本。

Lambda 的 15 分钟超时怎么样？

大多数验证码可在 10-60 秒内解决。将 Lambda 超时设置为 120 秒。对于像 reCAPTCHA Enterprise 这样的复杂类型，请使用 180 秒。

我可以将 Lambda 层用于请求库吗？

是的，但是 urllib.request（内置）对于 CaptchaAI 的简单 HTTP API 来说工作得很好。这完全避免了层管理。

相关指南

• 谷歌云函数 + CaptchaAI
• Azure 函数 + CaptchaAI

走向无服务器——获取您的 CaptchaAI 密钥今天。