Google Cloud Functions + CaptchaAI Entegrasyonu

CAPTCHA çözümünü 7/24 açık bir sunucuda çalıştırmak, çoğu iş yükü için boşta geçen saatlerin parasını ödemek demektir. Google Cloud Functions bu denklemi tersine çevirir: yalnızca bir çözüm isteği geldiğinde işlev ayağa kalkar, token'ı döndürür ve saniyeler içinde tekrar sıfır maliyete iner. Bu rehber, CaptchaAI API'sini GCP'nin sunucusuz katmanına HTTP ve Pub/Sub tetikleyicileriyle bağlamanın; anahtarı Secret Manager'da tutmanın ve gerçek maliyet rakamlarını görmenin tam yolunu gösterir.

Sunucusuz CAPTCHA çözümü ne zaman mantıklı?

Sunucusuz model, iş yükünüz düzensiz veya ani yük artışlıysa öne çıkar. Gün içinde birkaç yüz çözüm yapıp gece hiç istek almayan bir kazıma işi ya da yalnızca kampanya dönemlerinde tetiklenen bir QA akışı için sürekli açık bir sunucu tutmak, boşta geçen saatlerin faturasını ödemek anlamına gelir. Cloud Functions'ta ise ölçek sıfıra iner: istek yoksa maliyet de yoktur.

Türkiye'deki geliştiriciler için bunun somut bir karşılığı var. CaptchaAI ödemeleri USD üzerinden yapıldığından ve kur oynaklığı aylık bütçeyi tahmin etmeyi zorlaştırdığından, kullanım başına ödenen sunucusuz altyapıyı thread bazlı sabit CaptchaAI planlarıyla birleştirmek maliyeti öngörülebilir kılar. Buna karşılık hacminiz sabit ve yüksekse — örneğin dakikada yüzlerce çözüm — sürekli açık bir VM daha ucuza gelebilir; aşağıdaki tablo kırılma noktasını gösterir.

HTTP tetikleyicili CAPTCHA çözüm işlevi

İlk işlev, gelen JSON gövdesini okur, method parametresini belirler (varsayılan userrecaptcha), anahtarı Secret Manager'dan çeker ve çözümü CaptchaAI'nin in.php / res.php uç noktalarına gönderip sorgular. Bağımlılıkları hafif tutmak için harici bir HTTP kütüphanesi yerine standart urllib kullanılır; bu, soğuk başlangıç süresini kısaltır.

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

Bağımlılıkları tanımlayın

İşlevin ihtiyaç duyduğu tek harici paket Secret Manager istemcisidir. Çözüm mantığı urllib ile yürüdüğü için requests gibi ağır bağımlılıklardan kaçınırsınız — bu, dağıtım boyutunu ve soğuk başlangıç gecikmesini düşük tutar.

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

İşlevi GCP'ye dağıtın

Önce API anahtarınızı Secret Manager'a yazın, ardından işlevi Gen2 çalışma zamanıyla dağıtın. --timeout değerini en yavaş CAPTCHA türünüzün çözüm süresinden uzun tutun; --max-instances ise paralel çözüm sayınızın üst sınırını belirler.

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

Pub/Sub ile toplu CAPTCHA işleme

Yüzlerce görevi tek tek HTTP isteğiyle çağırmak yerine, çözümleri bir Pub/Sub kuyruğu üzerinden ayrıştırabilirsiniz. İşlev her mesajı tüketir, CAPTCHA'yı çözer ve sonucu ikinci bir konuya yayımlar; bu desen, üretici ile tüketiciyi birbirinden bağımsız ölçeklendirmenizi sağlar.

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

Pub/Sub işlevini dağıtın

Bu işlev HTTP yerine bir konu tarafından tetiklenir; --trigger-topic bayrağı işlevi captcha-tasks konusuna bağlar.

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

Görevleri Pub/Sub kuyruğuna gönderin

Kuyruğu beslemek için, çözülmesi gereken her sayfayı ayrı bir mesaj olarak yayımlayın. İşlev bu mesajları otomatik olarak alıp paralel çalıştırır.

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

Maliyet karşılaştırması: sunucusuz mu, sürekli VM mi?

Aşağıdaki değerler gözlemlenen kullanım verilerine dayanan tahminlerdir; gerçek maliyet bölgeye, bellek ayarına ve hacme göre değişir. Genel örüntü nettir: düşük ve orta hacimde sunucusuz kazanır, hacim büyüdükçe makas kapanır.

Sorun giderme

Üretim için güvenlik ve gözlemlenebilirlik

API anahtarınızı asla kaynak koda veya düz ortam değişkenine gömmeyin; örnekteki gibi Secret Manager'da tutup çalışma anında okuyun. İşleve atanan servis hesabına yalnızca secretmanager.secretAccessor rolünü tanıyın — en az ayrıcalık ilkesi, sızan bir anahtarın etki alanını daraltır. İşlev günlüklerini Cloud Logging'e yazın ve başarısız çözümleri task_id ile ilişkilendirin; böylece Pub/Sub akışında hangi görevin neden düştüğünü saniyeler içinde görürsünüz.

Kazıdığınız veriler kişisel veri içeriyorsa, Türkiye'de KVKK kapsamına girdiğini unutmayın. CaptchaAI'yi yalnızca yetkiniz olan QA, test ve veri toplama akışlarında kullanın; token doğrulaması yapan hedef sitelere karşı izinsiz otomasyon çalıştırmayın.

Sık sorulan sorular

Bu işlevle hangi CAPTCHA türlerini gönderebilirim?

method parametresi doğrudan API'ye iletildiği için CaptchaAI'nin desteklediği tüm türleri kullanabilirsiniz: reCAPTCHA v2/v3, Cloudflare Turnstile ve Challenge, GeeTest v3, görüntü/OCR, grid ve BLS. CaptchaFox, Friendly Captcha ve Lemin beta olarak sunulur. hCaptcha ile FunCaptcha desteklenmez; GeeTest v4 için destek çok yakında planlanıyor.

Sunucusuz mimaride hangi CaptchaAI planı yeterli olur?

CaptchaAI thread bazlı faturalandırır: her plan eşzamanlı çözüm sayısını (thread) belirler ve thread başına sınırsız çözüm sunar. Düşük hacimli veya değişken iş yükleri için BASIC ($15/ay, 5 thread) çoğu zaman yeterlidir; Cloud Functions örnekleri paralel çalışıp daha fazla eşzamanlılık gerektirdiğinde STANDARD ($30/ay, 15 thread) veya ADVANCE ($90/ay, 50 thread) planlarına çıkabilirsiniz.

Çözüm işlevi için zaman aşımını nasıl ayarlamalıyım?

İşlevin --timeout değeri, en yavaş CAPTCHA türünüzün çözüm süresinden uzun olmalıdır. reCAPTCHA ve Turnstile için 120 saniye rahat bir tampondur; koddaki _solve(..., timeout=90) sorgulama üst sınırını da buna uyumlu bırakın. Çok kısa bir değer, çözüm tamamlanmadan TimeoutError üretir.

İşlevi yetkisiz erişime karşı nasıl korurum?

Dahili kullanım için --allow-unauthenticated yerine --no-allow-unauthenticated ile kimlik doğrulamayı zorunlu kılın. İşlevi dışarıya açıyorsanız önüne API anahtarı doğrulaması yapan bir API Gateway koyun ve CaptchaAI anahtarınızı yalnızca Secret Manager üzerinden okuyun.

İlgili kılavuzlar

• AWS Lambda ile CaptchaAI entegrasyonu
• Azure Functions ile CaptchaAI entegrasyonu

GCP'de sunucusuz çözüme geçin — CaptchaAI anahtarınızı alın ve ilk işlevinizi bugün dağıtın.