الدروس التطبيقية

سجلّات تدقيق حل CAPTCHA: تتبّع كل طلب للامتثال

سجلّ التدقيق هو ما يحوّل آلاف عمليات حل CAPTCHA اليومية من صندوق أسود إلى سجلّ يمكن مراجعته. لكل عملية، يجيب السجل عن أربعة أسئلة:

  • من أطلق الطلب؟
  • لأي موقع كان؟
  • كم استغرق وبأي تكلفة؟
  • متى حدث بالضبط؟

عند وصول أول سؤال من فريق الامتثال أو التدقيق الداخلي، إما أن تملك إجابة دقيقة لكل طلب أو تبدأ التخمين. في هذا الدليل نبني طبقة تسجيل تدقيق خفيفة فوق CaptchaAI بلغتي Python وNode.js، مستقلة عن سجلات التطبيق العادية، وتكفي مؤسسة في القاهرة أو الرياض تتعامل مع أنظمة حماية البيانات المحلية إلى جانب GDPR.

ما الحقول التي يجب تسجيلها

يجب أن يحفظ كل حل اختبار CAPTCHA الحقول التالية على الأقل:

الحقل الغرض مثال
timestamp لحظة إرسال الطلب 2026-04-04T14:30:00Z
request_id معرّف فريد لهذه العملية uuid4()
captcha_type طريقة CAPTCHA المستخدمة userrecaptcha
target_site عنوان URL للصفحة المطلوب حلّها https://example.com/login
task_id معرّف المهمة في CaptchaAI 73829451
status النتيجة solved، failed، timeout
solve_time_ms الزمن من الإرسال حتى النتيجة 18432
error_code رمز الخطأ عند الفشل ERROR_CAPTCHA_UNSOLVABLE
initiator الجهة التي أطلقت الحل scraper-job-42
cost التكلفة التقديرية 0.003

لا تُسجّل مطلقًا:

  • مفاتيح الـ API.
  • رموز CAPTCHA، فهي مؤقتة ولا قيمة تدقيقية لها.
  • أي بيانات تعريف شخصية تلتقطها من المواقع المستهدفة.

حصر السجل في حقول تشغيلية بلا بيانات شخصية يبقيه متوافقًا مع GDPR ومع أنظمة مثل PDPL السعودي دون معالجة إضافية.

تنفيذ التسجيل في Python

الدالة التالية تلفّ دورة الإرسال والاستطلاع كاملةً، وتكتب سجلّ تدقيق منظّمًا بصيغة JSONL في كل مسار — نجاح أو فشل أو انتهاء مهلة — مع تدوير تلقائي للملفات:

# audit_solver.py
import os
import uuid
import time
import json
import logging
from datetime import datetime, timezone
import requests

API_KEY = os.environ.get("CAPTCHAAI_KEY", "YOUR_API_KEY")

# Configure audit logger — separate from application logs
audit_logger = logging.getLogger("captcha_audit")
audit_logger.setLevel(logging.INFO)

# File handler with rotation
from logging.handlers import RotatingFileHandler
handler = RotatingFileHandler(
    "captcha_audit.jsonl",
    maxBytes=50_000_000,  # 50 MB per file
    backupCount=10,
)
handler.setFormatter(logging.Formatter("%(message)s"))
audit_logger.addHandler(handler)

def log_audit(record):
    """Write a structured audit record."""
    audit_logger.info(json.dumps(record, default=str))

def solve_with_audit(sitekey, pageurl, captcha_type="userrecaptcha",
                      initiator="unknown"):
    """Solve a CAPTCHA with full audit logging."""
    request_id = str(uuid.uuid4())
    start = time.time()

    audit_record = {
        "request_id": request_id,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "captcha_type": captcha_type,
        "target_site": pageurl,
        "initiator": initiator,
        "status": "submitted",
    }

    session = requests.Session()

    try:
        # Submit
        resp = session.get("https://ocr.captchaai.com/in.php", params={
            "key": API_KEY,
            "method": captcha_type,
            "googlekey": sitekey,
            "pageurl": pageurl,
            "json": "1",
        })
        result = resp.json()

        if result.get("status") != 1:
            audit_record.update({
                "status": "submit_failed",
                "error_code": result.get("request"),
                "solve_time_ms": int((time.time() - start) * 1000),
            })
            log_audit(audit_record)
            return None

        task_id = result["request"]
        audit_record["task_id"] = task_id

        # Poll
        time.sleep(15)
        for _ in range(25):
            poll = session.get("https://ocr.captchaai.com/res.php", params={
                "key": API_KEY, "action": "get",
                "id": task_id, "json": "1",
            })
            poll_result = poll.json()

            if poll_result.get("status") == 1:
                solve_time = int((time.time() - start) * 1000)
                audit_record.update({
                    "status": "solved",
                    "solve_time_ms": solve_time,
                    "cost_estimate": 0.003,  # Adjust per your rate
                })
                log_audit(audit_record)
                return poll_result["request"]

            if poll_result.get("request") != "CAPCHA_NOT_READY":
                audit_record.update({
                    "status": "failed",
                    "error_code": poll_result.get("request"),
                    "solve_time_ms": int((time.time() - start) * 1000),
                })
                log_audit(audit_record)
                return None

            time.sleep(5)

        audit_record.update({
            "status": "timeout",
            "solve_time_ms": int((time.time() - start) * 1000),
        })
        log_audit(audit_record)
        return None

    except Exception as e:
        audit_record.update({
            "status": "error",
            "error_code": str(e)[:200],
            "solve_time_ms": int((time.time() - start) * 1000),
        })
        log_audit(audit_record)
        raise

# Usage
token = solve_with_audit(
    sitekey="6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
    pageurl="https://www.google.com/recaptcha/api2/demo",
    initiator="price-scraper-v2",
)

مثال على مخرجات السجل (تنسيق JSONL)

كل سطر سجلّ مستقل وقابل للتحليل، ما يجعل الملف جاهزًا للإرسال إلى أي مجمّع سجلات لاحقًا:

{"request_id":"a1b2c3d4-...","timestamp":"2026-04-04T14:30:00+00:00","captcha_type":"userrecaptcha","target_site":"https://www.google.com/recaptcha/api2/demo","initiator":"price-scraper-v2","status":"solved","task_id":"73829451","solve_time_ms":18432,"cost_estimate":0.003}

تنفيذ التسجيل في Node.js

المنطق نفسه في بيئة JavaScript، مع كتابة إلحاقية إلى ملف JSONL واحد. الحقول متطابقة مع نسخة Python كي يبقى تنسيق السجل موحّدًا عبر خدماتك:

// audit_solver.js
const fs = require('fs');
const { v4: uuidv4 } = require('uuid');
const axios = require('axios');

const API_KEY = process.env.CAPTCHAAI_KEY || 'YOUR_API_KEY';
const AUDIT_FILE = 'captcha_audit.jsonl';

function logAudit(record) {
  fs.appendFileSync(AUDIT_FILE, JSON.stringify(record) + '\n');
}

async function solveWithAudit(sitekey, pageurl, initiator = 'unknown') {
  const requestId = uuidv4();
  const start = Date.now();
  const record = {
    request_id: requestId,
    timestamp: new Date().toISOString(),
    captcha_type: 'userrecaptcha',
    target_site: pageurl,
    initiator,
    status: 'submitted',
  };

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

    if (submit.data.status !== 1) {
      record.status = 'submit_failed';
      record.error_code = submit.data.request;
      record.solve_time_ms = Date.now() - start;
      logAudit(record);
      return null;
    }

    record.task_id = submit.data.request;
    await new Promise(r => setTimeout(r, 15000));

    for (let i = 0; i < 25; i++) {
      const poll = await axios.get('https://ocr.captchaai.com/res.php', {
        params: { key: API_KEY, action: 'get', id: submit.data.request, json: '1' },
      });

      if (poll.data.status === 1) {
        record.status = 'solved';
        record.solve_time_ms = Date.now() - start;
        record.cost_estimate = 0.003;
        logAudit(record);
        return poll.data.request;
      }
      if (poll.data.request !== 'CAPCHA_NOT_READY') {
        record.status = 'failed';
        record.error_code = poll.data.request;
        record.solve_time_ms = Date.now() - start;
        logAudit(record);
        return null;
      }
      await new Promise(r => setTimeout(r, 5000));
    }

    record.status = 'timeout';
    record.solve_time_ms = Date.now() - start;
    logAudit(record);
    return null;
  } catch (e) {
    record.status = 'error';
    record.error_code = e.message.slice(0, 200);
    record.solve_time_ms = Date.now() - start;
    logAudit(record);
    throw e;
  }
}

الاستعلام عن السجلّات وبناء الملخّصات

بعد تراكم السجلات، تحتاج إلى قراءتها بسرعة. الدالة التالية تمرّ على ملف JSONL وتُخرج لكل يوم:

  • إجمالي الطلبات وتوزيع الحالات (solved وfailed وtimeout).
  • التكلفة التقديرية الإجمالية.
  • وسيط زمن الحل بالمِلّي ثانية.

وفيما يلي التنفيذ:

import json
from collections import Counter
from datetime import date

def daily_summary(log_file, target_date=None):
    """Generate a daily summary from audit logs."""
    target = target_date or date.today().isoformat()
    statuses = Counter()
    total_cost = 0
    solve_times = []

    with open(log_file) as f:
        for line in f:
            record = json.loads(line)
            if record["timestamp"].startswith(target):
                statuses[record["status"]] += 1
                total_cost += record.get("cost_estimate", 0)
                if record.get("solve_time_ms"):
                    solve_times.append(record["solve_time_ms"])

    print(f"Date: {target}")
    print(f"Total requests: {sum(statuses.values())}")
    print(f"Statuses: {dict(statuses)}")
    print(f"Estimated cost: ${total_cost:.2f}")
    if solve_times:
        print(f"Median solve time: {sorted(solve_times)[len(solve_times)//2]}ms")

daily_summary("captcha_audit.jsonl")

انتبه إلى أن حقل cost_estimate رقم إرشادي لكل عملية تضبطه بنفسك، لأن فوترة CaptchaAI قائمة على عدد الـ Threads المتزامنة لا على عدد العمليات: كل خطة تمنحك حلولًا غير محدودة ضمن حصّتك من الـ Threads. استخدم هذا الحقل لتوزيع التكلفة داخليًا بين الفرق والمشاريع، ثم طابِق الاتجاه الشهري مع اشتراكك الفعلي.

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

قبل التفكير في التخزين، تأكّد أن طبقة التسجيل نفسها سليمة. أكثر المشكلات شيوعًا وحلولها:

المشكلة السبب الإجراء
ملف السجل يكبر أكثر من اللازم لا يوجد تدوير مُعدّ استخدم RotatingFileHandler أو أداة logrotate
سجلات تدقيق مفقودة استثناء يقع قبل الكتابة سجّل داخل كتلة finally لضمان كتابة السطر دائمًا
كتابة بطيئة عند الحجم العالي إدخال/إخراج متزامن للملف اكتب بشكل غير متزامن أو استخدم مخزنًا مؤقتًا (buffer)
طوابع زمنية غير متسقة انحراف ساعة النظام فعّل NTP وسجّل دائمًا بتوقيت UTC

الاحتفاظ بالسجلّات وتخزينها

يتغيّر خيار التخزين المناسب مع الحجم اليومي. استخدم الجدول التالي كنقطة انطلاق:

الحجم اليومي حجم السجل اليومي التخزين الشهري التوصية
100 حل/يوم ~30 كيلوبايت ~1 ميجابايت ملف محلي
1000 حل/يوم ~300 كيلوبايت ~10 ميجابايت ملف محلي مع تدوير
10000 حل/يوم ~3 ميجابايت ~100 ميجابايت إرسال إلى مجمّع سجلات
100000 حل/يوم ~30 ميجابايت ~1 جيجابايت تسجيل مركزي (ELK أو Datadog)

أما مدة الاحتفاظ فتُحدَّد وفق الدافع:

  • تشغيلي: 90 يومًا خيار شائع لتتبّع الأعطال والاتجاهات.
  • امتثال تنظيمي: التزم بمتطلبات قطاعك (SOC 2 أو GDPR أو HIPAA) وأنظمة حماية البيانات في السوق الذي تعمل فيه.

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

ما الذي يميّز سجل التدقيق عن سجلات التصحيح (debug)؟

سجل التصحيح مؤقت وموجّه للمطوّر أثناء تتبّع خلل. سجل التدقيق سجلّ دائم ومنظّم يجيب عن "من ومتى وبأي تكلفة" لكل عملية حل، ويُحفظ لمدة معلومة للمراجعة والمطابقة، لذلك نفصله في مُسجّل ومسار ملف مستقلّين.

كيف أضمن توافق السجلات مع أنظمة حماية البيانات في المنطقة؟

اقصر السجل على حقول تشغيلية بلا بيانات شخصية: لا عناوين بريد، ولا أسماء مستخدمين، ولا محتوى صفحات مستهدفة. عندها يبقى متوافقًا مع GDPR وأنظمة مثل PDPL السعودي وقانون حماية البيانات المصري دون خطوات إضافية، لأنه لا يحتوي أصلًا على معطيات خاضعة لها.

هل يمكن الاعتماد على سجلات التدقيق لمطابقة الفاتورة الشهرية؟

جزئيًا. إجماليات سجلاتك تعطيك صورة داخلية عن الاستهلاك، لكن CaptchaAI يفوتر على أساس عدد الـ Threads المتزامنة لا على كل عملية. استخدم الملخّص اليومي لتوزيع التكلفة بين الفرق، ثم قارن الاتجاه العام باشتراكك.

أين أخزّن السجلات عند تجاوز عشرات آلاف الطلبات يوميًا؟

بدءًا من حوالي 10000 حل يوميًا، تجاوز الملف المحلي وأرسل السجلات إلى مجمّع مركزي مثل ELK أو Datadog؛ فهذا يتيح البحث السريع والتنبيهات ويحمي السجل من الضياع عند تدوير الخوادم.


الخطوات التالية

أدلة ذات صلة

التعليقات غير مفعّلة لهذا المقال.