用于无服务器验证码解决跟踪的 DynamoDB (实用指南)

DynamoDB 自然适合无服务器验证码工作流程 - 没有令人头痛的连接池、用于自动清理的内置 TTL 以及任何规模的一致性能。本指南涵盖了表设计、项目结构以及用于在基于 Lambda 的架构中跟踪验证码解决方案的查询模式。

桌子设计

单表模式

一张 DynamoDB 表处理解决历史记录、活动任务和聚合统计数据：

分区键 (PK)	排序键 (SK)	目的
`SOLVE#{captcha_id}`	`META`	解决记录
`SITE#{sitekey}`	`SOLVE#{timestamp}`	每个站点的解决历史记录
`STATS#{date}`	`TYPE#{captcha_type}`	每日汇总统计数据
`ACTIVE#{captcha_id}`	`TASK`	飞行中任务跟踪

表定义

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

Python实现

设置

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

求解和跟踪

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
    )

查询模式

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

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

成本优化

战略	影响
对可变工作负载使用按需计费	没有过度配置
启用 TTL 以自动记录清理	降低存储成本
仅项目查询中需要的属性	更低的读取单位消耗
使用 `BatchWriteItem` 批量写入	更少的 API 调用
使用 DynamoDB Streams 进行分析	将聚合卸载到 Lambda

故障排除

问题	原因	处理方式
`ProvisionedThroughputExceededException`	每秒写入次数太多	切换到按需计费或增加 WCU
TTL 项目未立即删除	最终将删除 DynamoDB TTL（约 48 小时）	不要依赖 TTL 进行实时清理；过滤查询中过期的项目
`STATS#{date}` 上的热分区	所有工作人员写入同一分区	使用随机后缀：`STATS#{date}#shard{0-9}`
查询返回太多项目	宽分区键	添加 SK 条件以缩小结果范围

常问问题

为什么使用 DynamoDB 而不是 RDS 来进行无服务器验证码跟踪？

DynamoDB 没有连接限制 - 非常适合 Lambda，每次调用都会打开一个新连接。RDS 需要连接池（RDS 代理），这会增加成本和复杂性。

DynamoDB 验证码跟踪的费用是多少？

采用按需计费：每百万次写入约 1.25 美元，每百万次读取约 0.25 美元。 10,000 次解决 /day 时，预计存储和访问费用将低于 1/month。

我可以查询所有验证码类型吗？

使用GSI1索引跨类型按状态查询。对于跨类型分析，请使用 DynamoDB Streams 和写入 STATS# 分区的 Lambda 函数进行聚合。

下一步

构建可自动扩展的无服务器验证码跟踪 –”获取您的 CaptchaAI API 密钥。

相关指南：

AWS Lambda + CaptchaAI
MongoDB 验证码历史
Redis 令牌 TTL 管理

用于无服务器验证码解决跟踪的 DynamoDB

桌子设计

单表模式

表定义

Python实现

设置

求解和跟踪

查询模式

JavaScript 实现

成本优化

故障排除

常问问题

为什么使用 DynamoDB 而不是 RDS 来进行无服务器验证码跟踪？

DynamoDB 验证码跟踪的费用是多少？

我可以查询所有验证码类型吗？

下一步

使用 AWS SNS 和 CaptchaAI 构建事件驱动的验证码解决方案

AWS Lambda + CaptchaAI：无服务器验证码解决

用于 CaptchaAI Worker 部署的 Ansible Playbook

验证码解决基础设施的蓝绿部署

Azure Functions + CaptchaAI：云集成

构建 CaptchaAI 使用仪表板和监控

桌子设计

单表模式

表定义

Python实现

设置

求解和跟踪

查询模式

JavaScript 实现

成本优化

故障排除

常问问题

为什么使用 DynamoDB 而不是 RDS 来进行无服务器验证码跟踪？

DynamoDB 验证码跟踪的费用是多少？

我可以查询所有验证码类型吗？

下一步

相关文章

使用 AWS SNS 和 CaptchaAI 构建事件驱动的验证码解决方案

AWS Lambda + CaptchaAI：无服务器验证码解决

用于 CaptchaAI Worker 部署的 Ansible Playbook

验证码解决基础设施的蓝绿部署

Azure Functions + CaptchaAI：云集成

构建 CaptchaAI 使用仪表板和监控