Azure Functions + CaptchaAI：云集成

Azure Functions 提供无服务器验证码解决方案，并与 Azure 生态系统紧密集成 - 用于机密的 Key Vault、用于任务分发的队列存储以及用于监控的 Application Insights。

HTTP 触发函数

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

密钥库集成

将 API 密钥存储在 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

应用程序设置中参考：

CAPTCHAAI_KEY=@Microsoft.KeyVault(SecretUri=https://captchaai-vault.vault.azure.net/secrets/CaptchaAIKey/)

队列触发的批处理

从 Azure 队列存储处理 CAPTCHA 任务：

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

项目结构

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

需求.txt：

azure-functions

主机.json：

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

本地.settings.json：

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

部署

# 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"
    }
  }'

将任务提交到队列

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

故障排除

常问问题

消费计划与高级计划？

使用低容量消耗 (< 100/day)。使用 Premium 来实现一致的工作负载 — 它可以使实例保持温暖、消除冷启动并支持 VNET 集成。

Azure Functions 定价与 AWS Lambda 相比如何？

验证码工作负载非常相似。每次调用均收费约 0.0001 美元，费用为 256MB/60s. Azure 包含 100 万次免费调用/month.

我可以将 Durable Functions 用于复杂的工作流程吗？

是的。 Durable Functions 支持扇出 /fan-in 模式 - 并行提交 10 个验证码，然后收集所有结果。非常适合批处理。

相关指南

• AWS Lambda + CaptchaAI
• 谷歌云函数 + CaptchaAI

部署在 Azure 上 –获取您的 CaptchaAI 密钥今天。