أتمتة سير العمل متعدد الخطوات باستخدام CaptchaAI

عندما تدير عشرات الحسابات على منصة واحدة، تتعثّر الأتمتة عند النقطة نفسها: كابتشا تسجيل الدخول. الحل يقوم على أربعة مكوّنات مترابطة — سجل حسابات مركزي، مدير جلسات لملفات تعريف الارتباط والبروكسي، طبقة لحل CAPTCHA عبر CaptchaAI، ومنفّذ يشغّل المهمة بالتوازي. نبنيها هنا بلغة Python مع حل reCAPTCHA v2 وCloudflare Turnstile.

متى تحتاج إلى أتمتة متعددة الحسابات؟

تظهر الحاجة كلما تكرر الإجراء نفسه على حسابات متعددة تتطلب تسجيل دخول محمياً:

لماذا تُعدّ كابتشا تسجيل الدخول نقطة الاختناق؟

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

نفصل لهذا السبب حل CAPTCHA في طبقة مستقلة بدل دمجه داخل منطق تسجيل الدخول. فحين يتبدّل نوع الاختبار من reCAPTCHA v2 إلى Cloudflare Turnstile على إحدى المنصات، تُعدَّل الطبقة وحدها دون المساس ببقية المكوّنات.

بنية النظام: أربعة مكوّنات

تتدفق البيانات من سجل الحسابات إلى مدير الجلسات فطبقة حل CAPTCHA، ثم إلى منفّذ سير العمل.

┌────────────────┐     ┌──────────────┐     ┌───────────┐     ┌───────────────┐
│ Account        │────▶│ Session      │────▶│ CAPTCHA   │────▶│ Workflow      │
│ Registry       │     │ Manager      │     │ Solver    │     │ Executor      │
│ (credentials)  │     │ (cookies,    │     │           │     │ (per-account  │
│                │     │  proxies)    │     │           │     │  actions)     │
└────────────────┘     └──────────────┘     └───────────┘     └───────────────┘

بناء المكوّنات خطوة بخطوة

سجل الحسابات

import json
from dataclasses import dataclass, field, asdict
from typing import Optional


@dataclass
class Account:
    id: str
    platform: str
    username: str
    password: str
    proxy: Optional[str] = None
    cookies: dict = field(default_factory=dict)
    last_login: Optional[float] = None
    status: str = "active"


class AccountRegistry:
    def __init__(self, filepath="accounts.json"):
        self.filepath = filepath
        self.accounts = {}
        self._load()

    def _load(self):
        try:
            with open(self.filepath, "r") as f:
                data = json.load(f)
                for item in data:
                    acct = Account(**item)
                    self.accounts[acct.id] = acct
        except FileNotFoundError:
            pass

    def save(self):
        data = [asdict(a) for a in self.accounts.values()]
        with open(self.filepath, "w") as f:
            json.dump(data, f, indent=2)

    def add(self, account):
        self.accounts[account.id] = account
        self.save()

    def get(self, account_id):
        return self.accounts.get(account_id)

    def get_by_platform(self, platform):
        return [a for a in self.accounts.values() if a.platform == platform and a.status == "active"]

    def update_status(self, account_id, status):
        if account_id in self.accounts:
            self.accounts[account_id].status = status
            self.save()

يخزّن السجل بيانات الاعتماد والبروكسي وملفات تعريف الارتباط لكل حساب، ويجلبها حسب المنصة.

مدير الجلسات

import time
import requests
import pickle
import os


class SessionManager:
    def __init__(self, sessions_dir="sessions"):
        self.sessions_dir = sessions_dir
        os.makedirs(sessions_dir, exist_ok=True)
        self.sessions = {}

    def get_session(self, account):
        """Get or create a requests session for an account."""
        if account.id in self.sessions:
            return self.sessions[account.id]

        session = requests.Session()

        # Set proxy if configured
        if account.proxy:
            session.proxies = {
                "http": account.proxy,
                "https": account.proxy,
            }

        # Load saved cookies
        cookie_file = os.path.join(self.sessions_dir, f"{account.id}.cookies")
        if os.path.exists(cookie_file):
            with open(cookie_file, "rb") as f:
                session.cookies = pickle.load(f)

        self.sessions[account.id] = session
        return session

    def save_session(self, account):
        """Persist session cookies."""
        if account.id in self.sessions:
            cookie_file = os.path.join(self.sessions_dir, f"{account.id}.cookies")
            with open(cookie_file, "wb") as f:
                pickle.dump(self.sessions[account.id].cookies, f)

    def clear_session(self, account_id):
        if account_id in self.sessions:
            del self.sessions[account_id]
        cookie_file = os.path.join(self.sessions_dir, f"{account_id}.cookies")
        if os.path.exists(cookie_file):
            os.remove(cookie_file)

يحتفظ مدير الجلسات بجلسة requests مستقلة لكل حساب، ويستعيد ملفات تعريف الارتباط فلا يُعاد تسجيل الدخول في كل تشغيل.

تسجيل الدخول مع حل CAPTCHA

هنا يدخل CaptchaAI. عند reCAPTCHA v2 أو Cloudflare Turnstile يرسل CaptchaSolver المهمة إلى in.php ويستطلع res.php حتى يعود الرمز، فيُضاف إلى الحمولة تحت الحقل المناسب — g-recaptcha-response لـ reCAPTCHA أو cf-turnstile-response لـ Turnstile.

import time
import requests


class CaptchaSolver:
    BASE = "https://ocr.captchaai.com"

    def __init__(self, api_key):
        self.api_key = api_key

    def solve(self, params, initial_wait=10):
        params["key"] = self.api_key
        params["json"] = 1
        resp = requests.post(f"{self.BASE}/in.php", data=params).json()
        if resp["status"] != 1:
            raise Exception(resp["request"])
        task_id = resp["request"]
        time.sleep(initial_wait)
        for _ in range(60):
            result = requests.get(
                f"{self.BASE}/res.php",
                params={"key": self.api_key, "action": "get", "id": task_id, "json": 1},
            ).json()
            if result["request"] == "CAPCHA_NOT_READY":
                time.sleep(5)
                continue
            if result["status"] == 1:
                return result["request"]
            raise Exception(result["request"])
        raise TimeoutError("Timed out")


class LoginHandler:
    def __init__(self, captcha_solver):
        self.solver = captcha_solver

    def login(self, session, account, login_config):
        """
        login_config: {
            "url": login page URL,
            "submit_url": login form action URL,
            "captcha_type": "recaptcha_v2" | "turnstile" | None,
            "sitekey": "...",
            "username_field": "username",
            "password_field": "password",
            "captcha_field": "g-recaptcha-response",
        }
        """
        # Get login page (for CSRF token / cookies)
        session.get(login_config["url"])

        payload = {
            login_config.get("username_field", "username"): account.username,
            login_config.get("password_field", "password"): account.password,
        }

        # Solve CAPTCHA if present
        captcha_type = login_config.get("captcha_type")
        if captcha_type:
            if captcha_type == "recaptcha_v2":
                token = self.solver.solve({
                    "method": "userrecaptcha",
                    "googlekey": login_config["sitekey"],
                    "pageurl": login_config["url"],
                })
            elif captcha_type == "turnstile":
                token = self.solver.solve({
                    "method": "turnstile",
                    "sitekey": login_config["sitekey"],
                    "pageurl": login_config["url"],
                })
            else:
                raise ValueError(f"Unknown captcha type: {captcha_type}")

            captcha_field = login_config.get("captcha_field", "g-recaptcha-response")
            payload[captcha_field] = token

        submit_url = login_config.get("submit_url", login_config["url"])
        resp = session.post(submit_url, data=payload, allow_redirects=True)

        # Check login success
        success = resp.status_code == 200 and "login" not in resp.url.lower()
        if success:
            account.last_login = time.time()
        return success

يسجّل المنفّذ الدخول عند الحاجة ثم يشغّل الدالة المطلوبة على كل حسابات المنصة بالتوازي عبر ThreadPoolExecutor بحدّ max_workers.

منفّذ سير العمل

import time
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed

logger = logging.getLogger("workflow")


class WorkflowExecutor:
    def __init__(self, api_key, max_workers=5):
        self.solver = CaptchaSolver(api_key)
        self.registry = AccountRegistry()
        self.sessions = SessionManager()
        self.login_handler = LoginHandler(self.solver)
        self.max_workers = max_workers

    def run_for_account(self, account, workflow_fn, login_config):
        """Execute a workflow for a single account."""
        session = self.sessions.get_session(account)

        # Login if needed
        if not account.last_login or time.time() - account.last_login > 3600:
            logger.info(f"Logging in: {account.id}")
            if not self.login_handler.login(session, account, login_config):
                logger.error(f"Login failed: {account.id}")
                self.registry.update_status(account.id, "login_failed")
                return {"account": account.id, "status": "login_failed"}

            self.sessions.save_session(account)
            self.registry.save()

        # Execute workflow
        try:
            result = workflow_fn(session, account)
            return {"account": account.id, "status": "success", "data": result}
        except Exception as e:
            logger.error(f"Workflow failed for {account.id}: {e}")
            return {"account": account.id, "status": "error", "error": str(e)}

    def run_for_all(self, platform, workflow_fn, login_config):
        """Execute a workflow across all accounts on a platform."""
        accounts = self.registry.get_by_platform(platform)
        results = []

        with ThreadPoolExecutor(max_workers=self.max_workers) as pool:
            futures = {
                pool.submit(self.run_for_account, acct, workflow_fn, login_config): acct
                for acct in accounts
            }
            for future in as_completed(futures):
                result = future.result()
                results.append(result)
                logger.info(f"  {result['account']}: {result['status']}")

        return results

مثال تشغيلي كامل

يجمع المثال التالي المكوّنات الأربعة: يهيّئ المنفّذ، يعرّف إعدادات تسجيل الدخول ونوع CAPTCHA، ثم يشغّل فحص الإشعارات عبر الحسابات.

executor = WorkflowExecutor("YOUR_API_KEY", max_workers=3)

# Define platform login config
login_config = {
    "url": "https://platform.example.com/login",
    "submit_url": "https://platform.example.com/api/login",
    "captcha_type": "recaptcha_v2",
    "sitekey": "6Le-wvkSAAAA...",
    "username_field": "email",
    "password_field": "password",
    "captcha_field": "g-recaptcha-response",
}


# Define workflow
def check_notifications(session, account):
    resp = session.get("https://platform.example.com/api/notifications")
    data = resp.json()
    return {
        "unread": data.get("unread_count", 0),
        "latest": data.get("notifications", [])[:5],
    }


# Run across all accounts
results = executor.run_for_all("example_platform", check_notifications, login_config)

for r in results:
    if r["status"] == "success":
        print(f"{r['account']}: {r['data']['unread']} unread notifications")
    else:
        print(f"{r['account']}: {r['status']}")

اضبط عدد العمال حسب خطة الـ Threads

فوترة CaptchaAI قائمة على الـ Threads (الطلبات المتزامنة) لا على عدد عمليات الحل، فلا تجعل max_workers أكبر من ثريدات خطتك. تبدأ الخطط من BASIC ($15 شهرياً، 5 threads) ثم STANDARD ($30 شهرياً، 15 thread) فأعلى، مع حل غير محدود ضمن كل خطة.

تخيّل وكالة تسويق في الرياض تدير 40 حساب بائع على منصة تجارة إلكترونية إقليمية. مع STANDARD (15 thread) يُضبط max_workers=15 وتُوزَّع الحسابات على دفعات؛ فرفع الرقم فوق الثريدات يزيد قائمة الانتظار فقط دون تسريع.

أمثلة على مهام سير العمل

تصدير البيانات

def export_data(session, account):
    resp = session.get("https://platform.example.com/api/export")
    filename = f"export_{account.id}.json"
    with open(filename, "w") as f:
        f.write(resp.text)
    return {"file": filename, "size": len(resp.text)}

فحص حالة الحساب

def check_status(session, account):
    resp = session.get("https://platform.example.com/api/account/status")
    return resp.json()

تحديث الإعدادات

def update_settings(session, account):
    resp = session.post(
        "https://platform.example.com/api/settings",
        json={"timezone": "UTC", "notifications": True},
    )
    return {"updated": resp.status_code == 200}

تأمين بيانات الاعتماد والجلسات

لا تُضمّن كلمات المرور أو مفاتيح الـ API داخل الكود مباشرة؛ احفظها في متغيرات البيئة أو في مدير أسرار، واقصر صلاحية الوصول على من يشغّل سير العمل فعلياً.

• خزّن مفتاح الـ API في متغير بيئة بدل كتابته عند إنشاء WorkflowExecutor.
• انقل ملف accounts.json إلى مخزن مشفّر عند التشغيل في بيئة مشتركة.
• ألغِ صلاحية أي حساب يظهر في سجل الحالة بوصف login_failed حتى مراجعته.

يبقى سجل الحسابات بذلك مصدراً واحداً للحقيقة دون تسريب أسرار في مستودع الكود أو في سجلات التشغيل.

اربط بروكسي مستقلاً بكل حساب

يطبّق SessionManager قيمة الحقل proxy من كل كائن Account على جلسته تلقائياً، فتخرج طلبات كل حساب من عنوان IP خاص به. هذا يحافظ على سمعة عنوان IP مستقلة ويقلّل تداخل الجلسات المتوازية.

يفضّل كثير من الفرق في السوق الإقليمي الوكيل السكني القريب جغرافياً من المنصة المستهدفة لتقليل زمن الاستجابة. اضبط الحقل لكل حساب داخل سجل الحسابات.

لا تشارك البروكسي نفسه بين عدد كبير من الحسابات حتى لا يتحول إلى نقطة فشل واحدة تُسقط تسجيل الدخول لكل الحسابات دفعةً واحدة.

اضبط المهلات وأعد المحاولة بذكاء

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

• الأخطاء العابرة (مهلة أو انقطاع شبكة): أعد المحاولة مع تأخير تصاعدي.
• رمز الخطأ 429: خفّض max_workers وباعد بين الطلبات.
• الأخطاء الدائمة (بيانات اعتماد خاطئة): سجّلها وأوقف الحساب بدل تكرار المحاولة.

بهذا التمييز بين الأخطاء تتجنّب إغراق الخدمة بمحاولات فاشلة متكررة تستهلك ثريدات خطتك دون فائدة.

راقب التشغيل وسجّل النتائج

يكتب WorkflowExecutor حالة كل حساب عبر logger، ما يمنحك أثراً واضحاً لكل تشغيل: من نجح تسجيل دخوله، ومن فشل، وأي مهمة أطلقت استثناءً. وجّه هذا السجل إلى ملف أو نظام مراقبة مركزي عند التشغيل المجدول.

راجع دورياً الحسابات ذات الحالة error أو login_failed؛ فارتفاع نسبتها غالباً مؤشر على تغيّر مفتاح الموقع أو انتهاء صلاحية الجلسات لا على خلل في الكود نفسه.

يساعد الأثر المسجّل أيضاً في ضبط عدد العمال: إذا رأيت مهامّ تنتظر طويلاً قبل بدء الحل، فذلك مؤشر على أنك تجاوزت عدد ثريدات خطتك.

معالجة المشكلات الشائعة

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

ما أنواع CAPTCHA التي يغطيها هذا المسار؟

reCAPTCHA v2 وCloudflare Turnstile، وكلاهما مدعوم بالكامل في CaptchaAI. أما hCaptcha فغير مدعوم حالياً.

كيف أربط عدد العمال بخطة CaptchaAI؟

الفوترة قائمة على الـ Threads لا على عدد عمليات الحل. اضبط max_workers بحيث لا يفوق ثريدات خطتك — 5 مع BASIC ($15 شهرياً) أو 15 مع STANDARD ($30 شهرياً).

ماذا يحدث عند انتهاء صلاحية الجلسة؟

تصبح ملفات تعريف الارتباط غير صالحة فيفشل الطلب. استدعِ clear_session، وسيعيد المنفّذ تسجيل الدخول تلقائياً لاحقاً.

هل أشغّل منصات مختلفة في آنٍ واحد؟

نعم. السجل مُفهرس حسب المنصة، وget_by_platform يجلب حسابات كل منصة، فتشغّل تدفقاً مستقلاً بإعدادات خاصة بها.

كيف أؤمّن مفاتيح الـ API وكلمات المرور؟

احفظها في متغيرات البيئة أو في مدير أسرار، ولا تكتبها داخل الكود. امنح صلاحية القراءة لمن يشغّل سير العمل فقط، ودوّر المفاتيح دورياً وألغِ صلاحية أي مفتاح مكشوف فوراً.

أدلة ذات صلة

• أنماط معمارية لحل CAPTCHA بأحجام كبيرة
• ابدأ مع CaptchaAI: حلّ أول كابتشا في خمس دقائق

وسّع أتمتة حساباتك المتعددة مع حل CAPTCHA عبر CaptchaAI.