Hướng Dẫn Thực Hành

DynamoDB dành cho tính năng theo dõi giải quyết CAPTCHA không có máy chủ

Tự động hóa Python Tất cả loại CAPTCHA
Por Đội ngũ CaptchaAI Việt Nam
Publicado em May 29, 2026
DynamoDB dành cho tính năng theo dõi giải quyết CAPTCHA không có máy chủ
Imagem em destaque indisponível

DynamoDB phù hợp một cách tự nhiên với quy trình làm việc của CAPTCHA không có máy chủ - không cần phải đau đầu khi gộp nhóm kết nối, TTL tích hợp để tự động dọn dẹp và hiệu suất nhất quán ở mọi quy mô. Hướng dẫn này đề cập đến thiết kế bảng, cấu trúc mục và mẫu truy vấn để theo dõi cách giải CAPTCHA trong kiến trúc dựa trên Lambda.

Thiết kế bảng

Mẫu bàn đơn

Một bảng DynamoDB xử lý lịch sử giải quyết, tác vụ đang hoạt động và số liệu thống kê tổng hợp:

Khóa phân vùng (PK) Khóa sắp xếp (SK) Mục đích
SOLVE#{captcha_id} META Giải quyết bản ghi
SITE#{sitekey} SOLVE#{timestamp} Lịch sử giải quyết trên mỗi trang web
STATS#{date} TYPE#{captcha_type} Số liệu thống kê tổng hợp hàng ngày
ACTIVE#{captcha_id} TASK Theo dõi nhiệm vụ trên chuyến bay

Định nghĩa bảng

{
  "TableName": "CaptchaSolves",
  "KeySchema": [
    { "AttributeName": "PK", "KeyType": "HASH" },
    { "AttributeName": "SK", "KeyType": "RANGE" }
  ],
  "AttributeDefinitions": [
    { "AttributeName": "PK", "KeyType": "S" },
    { "AttributeName": "SK", "KeyType": "S" },
    { "AttributeName": "GSI1PK", "KeyType": "S" },
    { "AttributeName": "GSI1SK", "KeyType": "S" }
  ],
  "GlobalSecondaryIndexes": [
    {
      "IndexName": "GSI1",
      "KeySchema": [
        { "AttributeName": "GSI1PK", "KeyType": "HASH" },
        { "AttributeName": "GSI1SK", "KeyType": "RANGE" }
      ],
      "Projection": { "ProjectionType": "ALL" }
    }
  ],
  "BillingMode": "PAY_PER_REQUEST",
  "TimeToLiveSpecification": {
    "AttributeName": "ttl",
    "Enabled": true
  }
}

Triển khai Python

thiết lập

import os
import time
from datetime import datetime, timezone
import boto3
import requests

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ.get("DYNAMODB_TABLE", "CaptchaSolves"))
API_KEY = os.environ["CAPTCHAAI_API_KEY"]

Giải quyết và theo dõi

def solve_and_track(sitekey, pageurl, captcha_type="recaptcha_v2", project=None):
    now = datetime.now(timezone.utc)
    timestamp = now.isoformat()
    ttl_90_days = int(now.timestamp()) + (90 * 24 * 3600)

    # Submit to CaptchaAI
    resp = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": 1
    })
    data = resp.json()

    if data.get("status") != 1:
        # Store error record
        table.put_item(Item={
            "PK": f"SITE#{sitekey}",
            "SK": f"SOLVE#{timestamp}",
            "captcha_type": captcha_type,
            "pageurl": pageurl,
            "status": "error",
            "error": data.get("request"),
            "submitted_at": timestamp,
            "project": project or "default",
            "ttl": ttl_90_days,
            "GSI1PK": f"STATUS#error",
            "GSI1SK": timestamp
        })
        return {"error": data.get("request")}

    captcha_id = data["request"]

    # Track active task
    table.put_item(Item={
        "PK": f"ACTIVE#{captcha_id}",
        "SK": "TASK",
        "sitekey": sitekey,
        "pageurl": pageurl,
        "captcha_type": captcha_type,
        "submitted_at": timestamp,
        "ttl": int(now.timestamp()) + 600  # Auto-clean in 10 min
    })

    # Poll for result
    polls = 0
    for _ in range(60):
        time.sleep(5)
        polls += 1
        result = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY, "action": "get",
            "id": captcha_id, "json": 1
        }).json()

        if result.get("status") == 1:
            solved_at = datetime.now(timezone.utc).isoformat()
            elapsed_ms = int(
                (datetime.now(timezone.utc) - now).total_seconds() * 1000
            )

            # Store success record
            table.put_item(Item={
                "PK": f"SOLVE#{captcha_id}",
                "SK": "META",
                "captcha_type": captcha_type,
                "sitekey": sitekey,
                "pageurl": pageurl,
                "status": "solved",
                "submitted_at": timestamp,
                "solved_at": solved_at,
                "elapsed_ms": elapsed_ms,
                "polls": polls,
                "project": project or "default",
                "ttl": ttl_90_days,
                "GSI1PK": f"STATUS#solved",
                "GSI1SK": timestamp
            })

            # Also store in site history
            table.put_item(Item={
                "PK": f"SITE#{sitekey}",
                "SK": f"SOLVE#{timestamp}",
                "captcha_id": captcha_id,
                "status": "solved",
                "elapsed_ms": elapsed_ms,
                "ttl": ttl_90_days
            })

            # Remove active task
            table.delete_item(Key={
                "PK": f"ACTIVE#{captcha_id}", "SK": "TASK"
            })

            # Update daily stats
            update_daily_stats(captcha_type, True, elapsed_ms)

            return {"solution": result["request"]}

        if result.get("request") != "CAPCHA_NOT_READY":
            table.put_item(Item={
                "PK": f"SITE#{sitekey}",
                "SK": f"SOLVE#{timestamp}",
                "captcha_id": captcha_id,
                "status": "error",
                "error": result.get("request"),
                "ttl": ttl_90_days
            })
            table.delete_item(Key={
                "PK": f"ACTIVE#{captcha_id}", "SK": "TASK"
            })
            update_daily_stats(captcha_type, False, 0)
            return {"error": result.get("request")}

    table.delete_item(Key={"PK": f"ACTIVE#{captcha_id}", "SK": "TASK"})
    update_daily_stats(captcha_type, False, 0)
    return {"error": "TIMEOUT"}

def update_daily_stats(captcha_type, success, elapsed_ms):
    date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    update_expr = "SET total_solves = if_not_exists(total_solves, :zero) + :one"
    expr_values = {":zero": 0, ":one": 1}

    if success:
        update_expr += ", successful = if_not_exists(successful, :zero) + :one"
        update_expr += ", total_elapsed = if_not_exists(total_elapsed, :zero) + :elapsed"
        expr_values[":elapsed"] = elapsed_ms
    else:
        update_expr += ", failed = if_not_exists(failed, :zero) + :one"

    table.update_item(
        Key={"PK": f"STATS#{date_str}", "SK": f"TYPE#{captcha_type}"},
        UpdateExpression=update_expr,
        ExpressionAttributeValues=expr_values
    )

Mẫu truy vấn

def get_site_history(sitekey, limit=50):
    """Get recent solves for a specific site key."""
    response = table.query(
        KeyConditionExpression="PK = :pk",
        ExpressionAttributeValues={":pk": f"SITE#{sitekey}"},
        ScanIndexForward=False,
        Limit=limit
    )
    return response["Items"]

def get_daily_stats(date_str=None):
    """Get stats for a specific date (default: today)."""
    if not date_str:
        date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")

    response = table.query(
        KeyConditionExpression="PK = :pk",
        ExpressionAttributeValues={":pk": f"STATS#{date_str}"}
    )
    return response["Items"]

def get_active_tasks():
    """List all currently active CAPTCHA tasks."""
    response = table.query(
        IndexName="GSI1",
        KeyConditionExpression="GSI1PK = :pk",
        ExpressionAttributeValues={":pk": "STATUS#polling"}
    )
    return response["Items"]

Triển khai JavaScript

const { DynamoDBClient } = require("@aws-sdk/client-dynamodb");
const { DynamoDBDocumentClient, PutCommand, QueryCommand, UpdateCommand } = require("@aws-sdk/lib-dynamodb");
const axios = require("axios");

const client = DynamoDBDocumentClient.from(new DynamoDBClient({}));
const TABLE = process.env.DYNAMODB_TABLE || "CaptchaSolves";
const API_KEY = process.env.CAPTCHAAI_API_KEY;

async function solveAndTrack(sitekey, pageurl, type = "recaptcha_v2") {
  const now = new Date();
  const timestamp = now.toISOString();
  const ttl = Math.floor(now.getTime() / 1000) + 90 * 24 * 3600;

  const submit = await axios.post("https://ocr.captchaai.com/in.php", null, {
    params: { key: API_KEY, method: "userrecaptcha", googlekey: sitekey, pageurl, json: 1 },
  });

  if (submit.data.status !== 1) {
    await client.send(new PutCommand({
      TableName: TABLE,
      Item: { PK: `SITE#${sitekey}`, SK: `SOLVE#${timestamp}`, status: "error", error: submit.data.request, ttl },
    }));
    return { error: submit.data.request };
  }

  const captchaId = submit.data.request;
  let polls = 0;

  for (let i = 0; i < 60; i++) {
    await new Promise((r) => setTimeout(r, 5000));
    polls++;
    const poll = await axios.get("https://ocr.captchaai.com/res.php", {
      params: { key: API_KEY, action: "get", id: captchaId, json: 1 },
    });

    if (poll.data.status === 1) {
      const elapsed = Date.now() - now.getTime();
      await client.send(new PutCommand({
        TableName: TABLE,
        Item: {
          PK: `SOLVE#${captchaId}`, SK: "META", captcha_type: type,
          sitekey, pageurl, status: "solved", submitted_at: timestamp,
          solved_at: new Date().toISOString(), elapsed_ms: elapsed, polls, ttl,
        },
      }));
      return { solution: poll.data.request };
    }

    if (poll.data.request !== "CAPCHA_NOT_READY") {
      return { error: poll.data.request };
    }
  }
  return { error: "TIMEOUT" };
}

async function getSiteHistory(sitekey, limit = 50) {
  const result = await client.send(new QueryCommand({
    TableName: TABLE,
    KeyConditionExpression: "PK = :pk",
    ExpressionAttributeValues: { ":pk": `SITE#${sitekey}` },
    ScanIndexForward: false,
    Limit: limit,
  }));
  return result.Items;
}

Tối ưu hóa chi phí

Chiến lược tác động
Sử dụng thanh toán theo yêu cầu cho khối lượng công việc thay đổi Không cung cấp quá mức
Bật TTL để tự động dọn dẹp bản ghi Giảm chi phí lưu trữ
Dự án chỉ cần các thuộc tính cần thiết trong truy vấn Tiêu thụ đơn vị đọc thấp hơn
Viết hàng loạt bằng BatchWriteItem Ít lệnh gọi API hơn
Sử dụng Luồng DynamoDB để phân tích Giảm tải tổng hợp cho Lambda

Khắc phục sự cố

Vấn đề Nguyên nhân Cách xử lý
ProvisionedThroughputExceededException Quá nhiều lần viết mỗi giây Chuyển sang thanh toán theo yêu cầu hoặc tăng WCU
Các mục TTL không bị xóa ngay lập tức Việc xóa DynamoDB TTL cuối cùng sẽ xảy ra (~48 giờ) Đừng dựa vào TTL để dọn dẹp theo thời gian thực; lọc các mục đã hết hạn trong truy vấn
Phân vùng nóng trên STATS#{date} Tất cả công nhân ghi vào cùng một phân vùng Sử dụng hậu tố ngẫu nhiên: STATS#{date}#shard{0-9}
Truy vấn trả về quá nhiều mục Khóa phân vùng rộng Thêm điều kiện SK để thu hẹp kết quả

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

Tại sao DynamoDB thay vì RDS để theo dõi CAPTCHA không có máy chủ?

DynamoDB không có giới hạn kết nối — hoàn hảo cho Lambda nơi mỗi lệnh gọi sẽ mở ra một kết nối mới. RDS yêu cầu tổng hợp kết nối (RDS Proxy), điều này làm tăng thêm chi phí và độ phức tạp.

DynamoDB tốn bao nhiêu tiền cho việc theo dõi CAPTCHA?

Với thanh toán theo yêu cầu: ~ 1,25 USD trên một triệu lượt ghi và ~ 0,25 USD trên một triệu lượt đọc. Với 10.000 giải/day, chi phí lưu trữ và truy cập sẽ dưới $1/month.

Tôi có thể truy vấn tất cả các loại CAPTCHA không?

Sử dụng chỉ mục GSI1 để truy vấn theo trạng thái trên các loại. Đối với phân tích nhiều loại, hãy tổng hợp bằng cách sử dụng Luồng DynamoDB và hàm Lambda ghi vào phân vùng STATS#.

Các bước tiếp theo

Xây dựng tính năng theo dõi CAPTCHA không cần máy chủ có khả năng tự động chia tỷ lệ —lấy khóa API CaptchaAI của bạn.

Hướng dẫn liên quan:

Os comentários estão desativados para este artigo.