DevOps & Scaling

Azure Functions + CaptchaAI：云集成

自动化 Python 所有 CAPTCHA 类型 DevOps Azure
作者 CaptchaAI 中文团队
发布时间 Apr 23, 2026
Azure Functions + CaptchaAI：云集成
特色图片不可用

要不要为验证码识别单独维护一台常驻服务器？大多数团队并不需要——用 Azure Functions 按次触发就够了。本文演示如何把 CaptchaAI 接入 Azure 的无服务器体系：HTTP 触发函数处理实时请求，Key Vault 托管 API Key，Queue Storage 做批量分发。函数跑在 Azure 全球区域（示例用 West US 2），云端直接和 CaptchaAI、reCAPTCHA 等目标站点通信，不依赖本机网络能否访问 Google 服务，这对国内开发者尤其友好。

HTTP 触发函数：验证码识别的入口

先建一个 HTTP 触发的 Function，接收 POST 请求里的验证码类型（method）和参数（params），转发给 CaptchaAI 的 in.php 提交、res.php 轮询，拿到 token 后原样返回。转发逻辑封装在 solve() 里，默认超时 90 秒，覆盖大多数 reCAPTCHA v2 / Turnstile 场景。

# function_app.py
import json
import time
import os
import logging
import urllib.request
import urllib.parse
import azure.functions as func

app = func.FunctionApp()


@app.route(route="solve", methods=["POST"])
def solve_captcha(req: func.HttpRequest) -> func.HttpResponse:
    """HTTP trigger for CAPTCHA solving."""
    try:
        body = req.get_json()
    except ValueError:
        return func.HttpResponse(
            json.dumps({"error": "JSON body required"}),
            status_code=400,
            mimetype="application/json",
        )

    method = body.get("method", "userrecaptcha")
    params = body.get("params", {})
    api_key = os.environ["CAPTCHAAI_KEY"]

    try:
        token = solve(api_key, method, params)
        return func.HttpResponse(
            json.dumps({"token": token}),
            mimetype="application/json",
        )
    except Exception as e:
        logging.error(f"Solve failed: {e}")
        return func.HttpResponse(
            json.dumps({"error": str(e)}),
            status_code=500,
            mimetype="application/json",
        )


def solve(api_key, method, params, timeout=90):
    """Solve CAPTCHA via CaptchaAI API."""
    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"]

    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")

solve() 只依赖标准库 urllib，打包进消费计划（Consumption Plan）不需要绑定虚拟网络。

用 Key Vault 托管 API Key，告别硬编码

API Key 写进代码或 .env 文件，一旦仓库泄露就跟着泄露。更稳妥的做法是存进 Azure Key Vault，函数只在运行时按需读取，代码库里不落地明文。

# Create Key Vault
az keyvault create \
  --name captchaai-vault \
  --resource-group myResourceGroup

# Store secret
az keyvault secret set \
  --vault-name captchaai-vault \
  --name CaptchaAIKey \
  --value "YOUR_API_KEY"

# Grant function access
az webapp identity assign \
  --name my-captcha-function \
  --resource-group myResourceGroup

az keyvault set-policy \
  --name captchaai-vault \
  --object-id <principal-id> \
  --secret-permissions get

给 Function App 开启托管身份（Managed Identity），再把 get 权限授予这个身份——运行时不用在任何地方明文保存密钥。

应用程序设置中引用：

[email protected](SecretUri=https://captchaai-vault.vault.azure.net/secrets/CaptchaAIKey/)

把这一行贴进 Application Settings 的 CAPTCHAAI_KEY，函数代码里仍然用 os.environ["CAPTCHAAI_KEY"] 读取，业务逻辑不用改。

队列触发批量识别

一次要识别几十上百个验证码时（比如批量巡检落地页），比起挨个发 HTTP 请求，更适合把任务丢进 Azure Queue Storage，交给队列触发的函数异步消费。

@app.queue_trigger(
    arg_name="msg",
    queue_name="captcha-tasks",
    connection="AzureWebJobsStorage",
)
def process_queue_task(msg: func.QueueMessage):
    """Process CAPTCHA task from queue."""
    task = json.loads(msg.get_body().decode())
    api_key = os.environ["CAPTCHAAI_KEY"]

    try:
        token = solve(api_key, task["method"], task["params"])
        logging.info(f"Task {task['id']} solved")

        # Store result in Table Storage or return queue
        _store_result(task["id"], "success", token)

    except Exception as e:
        logging.error(f"Task {task['id']} failed: {e}")
        _store_result(task["id"], "error", str(e))


def _store_result(task_id, status, value):
    """Store result (simplified — use Table Storage in production)."""
    logging.info(f"Result: {task_id} = {status}")

每条消息对应一个任务，结果生产环境应写回 Table Storage——示例用日志简化演示，接入时替换 _store_result()。

项目结构与本地开发

四个文件就能跑起来：

captcha-function/
├── function_app.py
├── requirements.txt
├── host.json
└── local.settings.json

requirements.txt：

azure-functions

host.json：

{
  "version": "2.0",
  "functionTimeout": "00:02:00",
  "logging": {
    "logLevel": {
      "default": "Information"
    }
  }
}

functionTimeout 从默认 5 分钟改成 2 分钟，避免轮询卡住整个实例。

local.settings.json：

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "CAPTCHAAI_KEY": "YOUR_API_KEY_FOR_LOCAL_DEV"
  }
}

这份文件只放本机调试用的 Key，务必加进 .gitignore。国内环境跑 pip install 建议加清华镜像，比如 pip install -i https://pypi.tuna.tsinghua.edu.cn/simple azure-functions，能省下载等待时间。

部署到 Azure 并验证

# Create function app
az functionapp create \
  --resource-group myResourceGroup \
  --consumption-plan-location westus2 \
  --runtime python \
  --runtime-version 3.11 \
  --functions-version 4 \
  --name my-captcha-solver \
  --storage-account mystorageaccount

# Deploy
func azure functionapp publish my-captcha-solver

# Test
curl -X POST https://my-captcha-solver.azurewebsites.net/api/solve \
  -H "Content-Type: application/json" \
  -d '{
    "method": "userrecaptcha",
    "params": {
      "googlekey": "SITE_KEY",
      "pageurl": "https://example.com"
    }
  }'

curl 直接测试线上端点，正常响应应带 token 字段；拿到 500 的话先查 Application Insights，多数是 CAPTCHAAI_KEY 没配置对，或 Key Vault 引用没 resolve 成功。

批量提交任务到队列

from azure.storage.queue import QueueClient
import json

queue = QueueClient.from_connection_string(
    conn_str="YOUR_STORAGE_CONNECTION_STRING",
    queue_name="captcha-tasks",
)

# Submit batch
for i in range(10):
    task = {
        "id": f"task-{i}",
        "method": "userrecaptcha",
        "params": {
            "googlekey": "SITE_KEY",
            "pageurl": f"https://example.com/page{i}",
        },
    }
    queue.send_message(json.dumps(task))
    print(f"Queued task-{i}")

send_message 只是把任务扔进队列，识别工作由 process_queue_task 异步完成，主流程不用等待，适合批量巡检和回归测试。

排障速查

问题 原因 处理方式
函数在 5 分钟后超时 用的是默认超时 在 host.json 里设置 functionTimeout
Key Vault 引用返回空 缺少托管身份或访问策略 分配托管身份，并授予 Key Vault 的 get 权限
队列消息不断重试 函数内部抛出了未捕获异常 捕获已知错误，记录日志后正常返回
冷启动超过 10 秒 Python 运行时初始化耗时 改用 Premium 计划，或设置 FUNCTIONS_WORKER_PROCESS_COUNT

常见问题

国内网络环境下调用会受限制吗？

不会。识别请求从 Azure 函数所在区域发出，直接访问 ocr.captchaai.com 和目标站点，不经过本机网络，不受本地能否访问 Google 服务影响。

消费计划和高级计划怎么选？

调用量不大（每天 100 次以内）用消费计划，按次计费划算。调用量稳定、要求低延迟的场景用 Premium——它能保持实例常驻、消除冷启动，还支持 VNET 集成。

Key Vault 里的密钥会不会被意外泄露？

按托管身份 + 访问策略配置后，密钥不会出现在代码、部署包或日志里。仍要定期检查谁有 Vault 访问权限，删除不再使用的授权。

Azure Functions 和 AWS Lambda 跑验证码识别，价格差多少？

两者对这类工作负载价格接近：256MB/60s 的调用大约都是每次 0.0001 美元左右，Azure 还包含每月 100 万次免费调用额度。

怎么查看验证码识别的调用量和费用？

打开 Function App 关联的 Application Insights，Overview 页能看到调用次数、失败率和平均响应时间；再结合 Azure Cost Management 按资源组核算实际花费，避免调用量突增时才发现账单异常。

相关指南

现在就把 CaptchaAI 部署到 Azure Functions —— 免费注册，拿到你的 API Key。

该文章已禁用评论。