Azure Functions est un bon point d'entrée si votre infrastructure tourne déjà sur Azure. Vous pouvez exposer un endpoint HTTP pour lancer la résolution, utiliser Key Vault pour la clé API, distribuer les tâches via Queue Storage et surveiller le tout avec les outils Azure habituels.
L'intérêt principal est de conserver une intégration serverless simple sans déployer de service dédié juste pour résoudre des CAPTCHA.
Fonction HTTP déclenchée à la demande
# 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")
Intégration avec Azure Key Vault
La clé API doit rester hors du code et des fichiers de configuration partagés. Key Vault est l'endroit naturel pour cela dans Azure.
# 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
Référence à utiliser dans les settings de l'application :
CAPTCHAAI_KEY=@Microsoft.KeyVault(SecretUri=https://captchaai-vault.vault.azure.net/secrets/CaptchaAIKey/)
Traitement batch via Queue Storage
Pour les charges plus importantes, vous pouvez découpler la soumission de la résolution avec une file 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}")
Structure du projet
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"
}
}
Deploiement
# 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"
}
}'
Soumettre des tâches à la file
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}")
Depannage
| Probleme | Cause probable | Correctif |
|---|---|---|
| La fonction expire trop vite | Timeout par défaut mal calibré | Définissez functionTimeout dans host.json |
| Key Vault ne renvoie rien | Identité ou policy absente | Activez l'identité managée et la policy d'accès |
| Les messages sont réessayés sans fin | Exception non gérée dans la function | Capturez les erreurs connues et journalisez-les |
| Le cold start est trop visible | Plan Consumption et runtime Python | Passez en Premium ou ajustez le worker count |
FAQ
Consumption ou Premium ?
Consumption convient aux faibles volumes ou aux usages irréguliers. Premium devient plus pertinent si vous voulez limiter les cold starts et garder des instances chaudes.
Azure Functions coûte-t-il plus cher qu'AWS Lambda ?
Pour ce type de charge, les deux plateformes restent proches. Le choix se fait surtout sur votre écosystème existant, vos contraintes réseau et vos besoins d'observabilité.
Peut-on utiliser Durable Functions ?
Oui. C'est une bonne option dès que vous voulez orchestrer plusieurs résolutions en parallèle puis agréger les résultats proprement.
Guides connexes
- AWS Lambda + CaptchaAI
- Fonctions Google Cloud + CaptchaAI
Déployez sur Azure, puis récupérez votre clé CaptchaAI pour brancher votre fonction serverless à vos workflows.
