谷歌云功能 + CaptchaAI 集成

验证码识别任务大多是脉冲式的：批量抓取跑起来时一分钟打几十次 API，空闲时可能几小时用不上一次。为此常年开一台虚拟机，多数时间都在空转。Google Cloud Functions 按调用次数和执行时长计费，配合 CaptchaAI 的 API，正好能把这种忽高忽低的负载交给 GCP 自动扩缩容——不用管服务器，账单也只在真正跑任务时才增长。下面先看两种方案的成本对比，再用 HTTP 触发和 Pub/Sub 触发两种方式接入 CaptchaAI（Cloud Functions 第二代），最后附排错清单。

成本对比：云函数 vs 常驻虚拟机

云函数最大的优势是空转成本为零。下表按每天求解次数粗略估算两种方案的计算费用（不含 CaptchaAI 本身的 API 费用）：

日调用量到万级左右两种方案基本打平；再往上虚拟机更划算，波动明显的场景云函数更省钱。这张表只是 GCP 侧成本——CaptchaAI 按线程数计费，和云函数账单是两笔独立的账。

HTTP 触发：提交任务并轮询结果

下面这个函数接收 POST 请求，从 Secret Manager 读取 API Key，调用 CaptchaAI 提交任务、轮询结果，最后把 token 返回给调用方。整个流程不需要维护任何常驻进程：

# main.py
import json
import time
import urllib.request
import urllib.parse
import functions_framework


@functions_framework.http
def solve_captcha(request):
    """HTTP Cloud Function for CAPTCHA solving."""
    # Parse request
    request_json = request.get_json(silent=True)
    if not request_json:
        return json.dumps({"error": "JSON body required"}), 400

    method = request_json.get("method", "userrecaptcha")
    params = request_json.get("params", {})

    # Get API key from Secret Manager
    api_key = _get_secret("captchaai-key")

    try:
        token = _solve(api_key, method, params)
        return json.dumps({"token": token})
    except Exception as e:
        return json.dumps({"error": str(e)}), 500


def _get_secret(secret_id):
    """Get secret from GCP Secret Manager."""
    from google.cloud import secretmanager
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/{_get_project_id()}/secrets/{secret_id}/versions/latest"
    response = client.access_secret_version(request={"name": name})
    return response.payload.data.decode("UTF-8")


def _get_project_id():
    """Get current GCP project ID."""
    import urllib.request
    req = urllib.request.Request(
        "http://metadata.google.internal/computeMetadata/v1/project/project-id",
        headers={"Metadata-Flavor": "Google"},
    )
    with urllib.request.urlopen(req) as resp:
        return resp.read().decode()


def _solve(api_key, method, params, timeout=90):
    """Solve CAPTCHA via CaptchaAI API."""
    # Submit
    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"]

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

_solve() 每 5 秒轮询一次 res.php，最长等待 90 秒；图片验证码之类耗时更长的类型，调大 timeout 参数即可。

安装依赖

运行环境是 Python 3.11，只需要两个包：

# requirements.txt
functions-framework==3.*
google-cloud-secret-manager==2.*

国内网络下本地装包慢，可以用清华 TUNA 镜像（pip install -i https://pypi.tuna.tsinghua.edu.cn/simple）；部署到 GCP 后不用额外配置，云端会正常走官方源。

部署与测试

把 API Key 写进 Secret Manager，再部署函数、开放 HTTP 触发器：

# Create secret
echo -n "YOUR_API_KEY" | gcloud secrets create captchaai-key --data-file=-

# Deploy function
gcloud functions deploy solve-captcha \
  --gen2 \
  --runtime=python311 \
  --region=us-central1 \
  --source=. \
  --entry-point=solve_captcha \
  --trigger-http \
  --allow-unauthenticated \
  --timeout=120s \
  --memory=256MB \
  --max-instances=100

# Test
curl -X POST https://us-central1-PROJECT.cloudfunctions.net/solve-captcha \
  -H "Content-Type: application/json" \
  -d '{
    "method": "userrecaptcha",
    "params": {
      "googlekey": "SITE_KEY",
      "pageurl": "https://example.com"
    }
  }'

示例用的是 us-central1；面向亚太用户时，asia-east1（台湾）或 asia-southeast1（新加坡）通常延迟更低。只在内部调用就把 --allow-unauthenticated 换成 --no-allow-unauthenticated，用 IAM 控制访问；要给外部客户端用，前面挂一层 API Gateway 校验 API Key。

常见故障排查

用 Pub/Sub 做批量处理

一次要处理成百上千个 URL 时，比逐个 HTTP 请求更合适的方式是把任务丢进 Pub/Sub 队列，让云函数按消息触发消费：

import base64
import json
import functions_framework
from google.cloud import pubsub_v1


@functions_framework.cloud_event
def process_captcha_task(cloud_event):
    """Process CAPTCHA task from Pub/Sub message."""
    data = base64.b64decode(cloud_event.data["message"]["data"])
    task = json.loads(data)

    api_key = _get_secret("captchaai-key")

    try:
        token = _solve(api_key, task["method"], task["params"])
        # Publish result
        publisher = pubsub_v1.PublisherClient()
        topic = f"projects/{_get_project_id()}/topics/captcha-results"
        publisher.publish(topic, json.dumps({
            "task_id": task["id"],
            "status": "success",
            "token": token,
        }).encode())

    except Exception as e:
        print(f"Task {task.get('id')} failed: {e}")

部署方式和 HTTP 函数类似，只是触发器换成了 topic：

gcloud functions deploy process-captcha-task \
  --gen2 \
  --runtime=python311 \
  --trigger-topic=captcha-tasks \
  --timeout=120s \
  --memory=256MB

向队列提交任务

客户端（或另一个函数）把待处理的 URL 列表发布到 topic，云函数会自动并发消费：

from google.cloud import pubsub_v1
import json

publisher = pubsub_v1.PublisherClient()
topic = "projects/YOUR_PROJECT/topics/captcha-tasks"

# Submit batch
urls = ["https://site1.com", "https://site2.com", "https://site3.com"]
for i, url in enumerate(urls):
    task = {
        "id": f"task-{i}",
        "method": "userrecaptcha",
        "params": {"googlekey": "SITE_KEY", "pageurl": url},
    }
    publisher.publish(topic, json.dumps(task).encode())
    print(f"Published task-{i}")

常见问题

该用 Gen1 还是 Gen2？

选 Gen2。最长超时可到 60 分钟，内存和并发上限也更高，跑图片验证码这类偶尔耗时较长的任务更稳。Gen1 基本只用于维护老项目。

冷启动会不会拖慢首次请求？

会，尤其是依赖包重的时候。示例代码只用标准库 urllib，冷启动一般在几百毫秒量级；换成 requests 之类第三方库会明显变长。在意延迟就把 --min-instances 设成 1，让至少一个实例常驻，成本约 $7/月。

中国大陆网络能稳定调用部署在 GCP 的服务吗？

调用你自己的端点没问题，但任务里含 reCAPTCHA 这类需要加载 Google 脚本的验证码时，大陆网络发起的测试可能不稳定。更稳妥的做法是把测试环境和 CI 放在境外区域，识别交给 CaptchaAI 服务器完成。

一个云函数能同时处理多种验证码类型吗？

能。method 字段直接对应 CaptchaAI 的 API 方法名，比如 userrecaptcha、turnstile、geetest（GeeTest v3）。按需传不同的 method 和 params，同一个云函数就能覆盖 reCAPTCHA、Turnstile、GeeTest v3 等类型，不用为每种验证码单独部署函数。

相关阅读

• AWS Lambda 部署 CaptchaAI 的方法
• Azure Functions 集成 CaptchaAI 指南

在 GCP 上跑无服务器验证码识别——获取你的 CaptchaAI API Key，几分钟内接入。