Chức năng Azure + CaptchaAI: Tích hợp đám mây

Azure Functions cung cấp khả năng giải quyết CAPTCHA không cần máy chủ với khả năng tích hợp chặt chẽ vào hệ sinh thái Azure - Key Vault để lưu trữ bí mật, Lưu trữ hàng đợi để phân phối tác vụ và Thông tin chi tiết về ứng dụng để giám sát.

Chức năng kích hoạt 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")

Tích hợp kho khóa

Lưu trữ khóa API trong 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

Tham khảo trong cài đặt ứng dụng:

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

Xử lý hàng loạt kích hoạt hàng đợi

Xử lý các tác vụ CAPTCHA từ Bộ lưu trữ hàng đợi Azure:

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

Cấu trúc dự án

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

local.settings.json:

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

Triển khai

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

Gửi nhiệm vụ vào hàng đợi

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

Khắc phục sự cố

Vấn đề	Nguyên nhân	Cách xử lý
Chức năng hết thời gian sau 5 phút	Thời gian chờ mặc định	Đặt `functionTimeout` trong Host.json
Tham chiếu Key Vault trả về trống	Thiếu danh tính/policy	Chỉ định danh tính được quản lý và chính sách Key Vault
Tin nhắn hàng đợi thử lại không ngừng	Hàm ném ngoại lệ	Xử lý các lỗi đã biết, đăng nhập và trả lại
Khởi động nguội > 10 giây	Khởi tạo thời gian chạy Python	Sử dụng gói Premium hoặc đặt `FUNCTIONS_WORKER_PROCESS_COUNT`

Câu hỏi thường gặp

Gói tiêu dùng và gói Premium?

Sử dụng Mức tiêu thụ cho âm lượng thấp (< 100/day). Sử dụng Premium để có khối lượng công việc ổn định — nó giữ cho phiên bản luôn ấm, loại bỏ khởi động nguội và hỗ trợ tích hợp VNET.

Giá của Azure Functions so với AWS Lambda như thế nào?

Rất giống với khối lượng công việc CAPTCHA. Cả hai đều tính phí ~ 0,0001 USD cho mỗi lệnh gọi ở mức 256MB/60s. Azure bao gồm 1 triệu lệnh gọi miễn phí/month.

Tôi có thể sử dụng Hàm bền vững cho quy trình công việc phức tạp không?

Vâng. Các chức năng bền bỉ hỗ trợ các mẫu phân xuất/fan-in - gửi song song 10 CAPTCHA, sau đó thu thập tất cả kết quả. Rất phù hợp để xử lý hàng loạt.

Hướng dẫn liên quan

Triển khai trên Azure —lấy khóa CaptchaAI của bạnHôm nay.