نمط قاطع الدائرة لاستدعاءات CAPTCHA API

عندما تكون واجهة برمجة التطبيقات لحل اختبار CAPTCHA بطيئة أو تعرض أخطاء، فإن الاستمرار في إرسال الطلبات يضيع الوقت والمال. يتوقف نمط قاطع الدائرة عن استدعاء واجهة برمجة التطبيقات الفاشلة، وينتظر الاسترداد، ويستأنف العمل تلقائيًا - مما يمنع حالات الفشل المتتالية في خط الأنابيب الخاص بك.

كيف تعمل قواطع الدائرة

ثلاث ولايات:

1. مغلق — تشغيل عادي. تمر الطلبات. يتم احتساب الفشل.
2. مفتوح — هناك عدد كبير جدًا من حالات الفشل. يتم رفض جميع الطلبات على الفور دون استدعاء واجهة برمجة التطبيقات.
3. نصف مفتوح — بعد فترة تهدئة، يُسمح بطلب اختبار واحد. إذا نجحت، يتم إغلاق الدائرة. إذا فشلت، يتم فتح الدائرة مرة أخرى.

تنفيذ بايثون

import time
import threading
import requests

SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"
API_KEY = "YOUR_API_KEY"


class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = 0
        self.state = "closed"  # closed, open, half-open
        self._lock = threading.Lock()

    def call(self, func, *args, **kwargs):
        with self._lock:
            if self.state == "open":
                if time.time() - self.last_failure_time > self.recovery_timeout:
                    self.state = "half-open"
                    print("[circuit] State: half-open — testing one request")
                else:
                    remaining = self.recovery_timeout - (
                        time.time() - self.last_failure_time
                    )
                    raise CircuitOpenError(
                        f"Circuit open — retry in {remaining:.0f}s"
                    )

        try:
            result = func(*args, **kwargs)
            with self._lock:
                self.failure_count = 0
                if self.state == "half-open":
                    print("[circuit] State: closed — API recovered")
                self.state = "closed"
            return result
        except Exception as e:
            with self._lock:
                self.failure_count += 1
                self.last_failure_time = time.time()
                if self.failure_count >= self.failure_threshold:
                    self.state = "open"
                    print(
                        f"[circuit] State: open — "
                        f"{self.failure_count} failures"
                    )
            raise


class CircuitOpenError(Exception):
    pass


def solve_captcha(sitekey, page_url):
    resp = requests.post(SUBMIT_URL, data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": page_url,
        "json": "1",
    }, timeout=15)
    data = resp.json()
    if data["status"] != 1:
        raise Exception(f"Submit error: {data['request']}")

    task_id = data["request"]
    for _ in range(24):
        time.sleep(5)
        poll = requests.get(RESULT_URL, params={
            "key": API_KEY,
            "action": "get",
            "id": task_id,
            "json": "1",
        }, timeout=15).json()
        if poll["status"] == 1:
            return poll["request"]
        if poll["request"] != "CAPCHA_NOT_READY":
            raise Exception(f"Poll error: {poll['request']}")
    raise TimeoutError(f"Task {task_id} timed out")


# Usage
breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30)

for i in range(10):
    try:
        token = breaker.call(
            solve_captcha, "6Le-SITEKEY", "https://example.com"
        )
        print(f"[task-{i}] Solved: {token[:40]}...")
    except CircuitOpenError as e:
        print(f"[task-{i}] Skipped: {e}")
    except Exception as e:
        print(f"[task-{i}] Failed: {e}")

الناتج المتوقع:

[task-0] Solved: 03AGdBq26ZfPxL...
[task-1] Solved: 03AGdBq27AbCdE...
[task-2] Failed: Submit error: ERROR_NO_SLOT_AVAILABLE
[task-3] Failed: Submit error: ERROR_NO_SLOT_AVAILABLE
[task-4] Failed: Submit error: ERROR_NO_SLOT_AVAILABLE
[circuit] State: open — 3 failures
[task-5] Skipped: Circuit open — retry in 28s
[task-6] Skipped: Circuit open — retry in 25s
...
[circuit] State: half-open — testing one request
[task-8] Solved: 03AGdBq28FgHiJ...
[circuit] State: closed — API recovered

تنفيذ JavaScript

class CircuitBreaker {
  constructor(options = {}) {
    this.failureThreshold = options.failureThreshold || 5;
    this.recoveryTimeout = options.recoveryTimeout || 60000;
    this.failureCount = 0;
    this.lastFailureTime = 0;
    this.state = 'closed';
  }

  async call(fn, ...args) {
    if (this.state === 'open') {
      if (Date.now() - this.lastFailureTime > this.recoveryTimeout) {
        this.state = 'half-open';
        console.log('[circuit] State: half-open');
      } else {
        const remaining = this.recoveryTimeout - (Date.now() - this.lastFailureTime);
        throw new Error(`Circuit open — retry in ${Math.ceil(remaining / 1000)}s`);
      }
    }

    try {
      const result = await fn(...args);
      this.failureCount = 0;
      if (this.state === 'half-open') {
        console.log('[circuit] State: closed — recovered');
      }
      this.state = 'closed';
      return result;
    } catch (error) {
      this.failureCount++;
      this.lastFailureTime = Date.now();
      if (this.failureCount >= this.failureThreshold) {
        this.state = 'open';
        console.log(`[circuit] State: open — ${this.failureCount} failures`);
      }
      throw error;
    }
  }
}

// Usage
const axios = require('axios');

const API_KEY = 'YOUR_API_KEY';
const breaker = new CircuitBreaker({ failureThreshold: 3, recoveryTimeout: 30000 });

async function solveCaptcha(sitekey, pageurl) {
  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) throw new Error(submit.data.request);
  const taskId = submit.data.request;

  for (let i = 0; i < 24; i++) {
    await new Promise(r => setTimeout(r, 5000));
    const poll = await axios.get('https://ocr.captchaai.com/res.php', {
      params: { key: API_KEY, action: 'get', id: taskId, json: 1 }
    });
    if (poll.data.status === 1) return poll.data.request;
    if (poll.data.request !== 'CAPCHA_NOT_READY') throw new Error(poll.data.request);
  }
  throw new Error('Timeout');
}

(async () => {
  for (let i = 0; i < 10; i++) {
    try {
      const token = await breaker.call(solveCaptcha, '6Le-SITEKEY', 'https://example.com');
      console.log(`[task-${i}] Solved: ${token.substring(0, 40)}...`);
    } catch (err) {
      console.log(`[task-${i}] ${err.message}`);
    }
  }
})();

اختيار العتبات

قم بتعيين حد الفشل مرتفعًا بما يكفي لتحمل الأخطاء المتقطعة (لا ينبغي أن تؤدي مهلة واحدة إلى تعطيل الدائرة) ولكن منخفضة بما يكفي لإيقاف ضرب واجهة برمجة التطبيقات الفاشلة.

الجمع مع منطق إعادة المحاولة

استخدم منطق إعادة المحاولة داخل قاطع الدائرة. يحسب قاطع الدائرة حالات الفشل النهائية (بعد استنفاد عمليات إعادة المحاولة):

def solve_with_retry(sitekey, page_url, max_retries=2):
    for attempt in range(max_retries + 1):
        try:
            return solve_captcha(sitekey, page_url)
        except Exception:
            if attempt == max_retries:
                raise
            time.sleep(2 ** attempt)

# Circuit breaker wraps the retry function
token = breaker.call(solve_with_retry, "6Le-SITEKEY", "https://example.com")

استكشاف الأخطاء وإصلاحها

الأسئلة الشائعة

هل يجب علي استخدام قواطع دوائر منفصلة للإرسال والاستطلاع؟

نعم، للأنظمة واسعة النطاق. قد تفشل نقطة نهاية الإرسال بينما لا يزال الاستقصاء يعمل (أو العكس). تمنحك القواطع المنفصلة تحكمًا أكثر دقة.

ماذا يجب أن أفعل عندما تكون الدائرة مفتوحة؟

قم بوضع مهمة CAPTCHA في قائمة الانتظار لوقت لاحق، أو قم بإظهار واجهة مستخدم احتياطية، أو تخطي العملية. يرىالتدهور اللطيف عند فشل الحل.

قم ببناء سير عمل CAPTCHA مرن باستخدام CaptchaAI

احصل على مفتاح API الخاص بك علىcaptchaai.com.

أدلة ذات صلة

• مرجع رموز أخطاء CaptchaAI: الدليل الكامل والإصلاحات
• البدء السريع مع CaptchaAI: حلّ أول كابتشا في 5 دقائق