التعامل مع اختبار CAPTCHA لجمع بيانات سوق الأوراق المالية

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

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

لماذا تستدعي المسارات المالية التحقق أسرع من غيرها

• دفعات طلبات كثيفة — حلقة تمر على 50 رمزاً خلال أقل من دقيقة تُنتج نمطاً لا يشبه تصفحاً بشرياً.
• جلسات قصيرة العمر — تشغيل يبدأ بجلسة جديدة يفقد كوكيز التحقق السابق ويعيد التحدي من الصفر.
• استعلامات ثقيلة — نتائج الفرز وتنزيلات CSV بالجملة تكلّف المصدر أكثر من صفحة سعر، فتُحاط بحماية أشد.

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

خريطة التحديات حسب نوع البيانات المالية

الصفّان الأولان يمثلان معظم حجم الطلبات، لكن نتائج الفرز تستهلك أكبر نصيب من عمليات الحل: كل تعديل في المرشّح يُعيد إطلاق التحدي.

ما يجب ضبطه قبل كتابة أول سطر برمجي

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

مثال من المنطقة: فريق في الرياض يتابع 40 رمزاً بين السوق السعودي والبورصة المصرية وسوق دبي المالي، مع أسهم أمريكية للمقارنة. الأولان يعملان من الأحد إلى الخميس، أما دبي ونيويورك فمن الاثنين إلى الجمعة؛ لذلك أي جدولة cron بأسبوع عمل واحد تُطلق دفعة على مصادر مغلقة: الأحد على دبي ونيويورك، والجمعة على الرياض والقاهرة — طلبات بلا بيانات جديدة، لكنها تُنتج تحديات حقيقية وتستهلك رصيداً. افصل الجداول حسب تقويم كل سوق.

جامع أسعار الأسهم في Python

الفئة التالية تجمع ثلاث مسؤوليات: جلسة requests واحدة تحتفظ بالكوكيز، كاشف يميّز صفحة التحقق من صفحة البيانات، ومسار حل يتفرّع بين Turnstile وreCAPTCHA بحسب شكل data-sitekey في الصفحة.

import requests
import time
import re
from datetime import datetime, timedelta

class StockDataCollector:
    def __init__(self, api_key):
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        })

    def get_quote(self, portal_url, symbol):
        """Get current stock quote, solving CAPTCHAs if needed."""
        url = f"{portal_url}/quote/{symbol}"
        response = self.session.get(url)

        if self._is_captcha_page(response):
            response = self._solve_and_retry(response, url)

        return self._parse_quote(response.text, symbol)

    def get_historical(self, portal_url, symbol, days=365):
        """Download historical price data."""
        url = f"{portal_url}/history/{symbol}"
        params = {
            "period": f"{days}d",
            "interval": "1d"
        }
        response = self.session.get(url, params=params)

        if self._is_captcha_page(response):
            response = self._solve_and_retry(response, url)

        return self._parse_historical(response.text)

    def scan_symbols(self, portal_url, symbols, delay=2):
        """Collect quotes for multiple symbols."""
        results = {}

        for symbol in symbols:
            try:
                results[symbol] = self.get_quote(portal_url, symbol)
                time.sleep(delay)
            except Exception as e:
                results[symbol] = {"error": str(e)}

        return results

    def _is_captcha_page(self, response):
        return (
            response.status_code == 403 or
            "cf-turnstile" in response.text or
            "challenges.cloudflare.com" in response.text
        )

    def _solve_and_retry(self, response, url):
        match = re.search(r'data-sitekey="(0x[^"]+)"', response.text)
        if not match:
            # Fall back to reCAPTCHA detection
            match = re.search(r'data-sitekey="([^"]+)"', response.text)
            if match:
                return self._solve_recaptcha_and_retry(match.group(1), url)
            raise ValueError("No CAPTCHA sitekey found")

        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key,
            "method": "turnstile",
            "sitekey": match.group(1),
            "pageurl": url,
            "json": 1
        })
        task_id = resp.json()["request"]

        for _ in range(60):
            time.sleep(3)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
                "json": 1
            })
            data = result.json()
            if data["status"] == 1:
                return self.session.post(url, data={
                    "cf-turnstile-response": data["request"]
                })

        raise TimeoutError("CAPTCHA solve timed out")

    def _solve_recaptcha_and_retry(self, site_key, url):
        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key,
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": url,
            "json": 1
        })
        task_id = resp.json()["request"]

        for _ in range(60):
            time.sleep(3)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
                "json": 1
            })
            data = result.json()
            if data["status"] == 1:
                return self.session.post(url, data={
                    "g-recaptcha-response": data["request"]
                })

        raise TimeoutError("reCAPTCHA solve timed out")

    def _parse_quote(self, html, symbol):
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, "html.parser")

        def text_or_none(node):
            return node.text.strip() if node and node.text else None

        return {
            "symbol": symbol,
            "price": text_or_none(soup.select_one("[data-field='regularMarketPrice'], .price")),
            "change": text_or_none(soup.select_one("[data-field='regularMarketChange'], .change")),
            "volume": text_or_none(soup.select_one("[data-field='regularMarketVolume'], .volume")),
            "market_cap": text_or_none(soup.select_one("[data-field='marketCap'], .market-cap")),
            "timestamp": datetime.now().isoformat()
        }

    def _parse_historical(self, html):
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, "html.parser")
        rows = []

        for row in soup.select("table tr")[1:]:  # Skip header
            cells = [td.text.strip() for td in row.select("td")]
            if len(cells) >= 6:
                rows.append({
                    "date": cells[0],
                    "open": cells[1],
                    "high": cells[2],
                    "low": cells[3],
                    "close": cells[4],
                    "volume": cells[5]
                })

        return rows


# Usage
collector = StockDataCollector("YOUR_API_KEY")

# Single quote
quote = collector.get_quote("https://finance.example.com", "AAPL")
print(f"AAPL: ${quote['price']} ({quote['change']})")

# Scan multiple symbols
portfolio = collector.scan_symbols(
    "https://finance.example.com",
    ["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA"]
)

الكاشف _is_captcha_page يعتمد على ثلاث إشارات مجتمعة — رمز الحالة 403، ووجود cf-turnstile في جسم الصفحة، والإشارة إلى نطاق تحدي Cloudflare — لأن إشارة واحدة تُنتج إنذارات كاذبة عند صفحات الخطأ أو الصيانة. وإعادة الإرسال بعد الحل تمر عبر self.session لا عبر طلب مستقل، وهو الفرق بين رمز يُقبل ورمز يُرفض بصمت.

وscan_symbols تفصل بين الرموز بفاصل زمني قابل للضبط؛ ارفعه عند تكرار التحديات قبل أي إجراء آخر.

أداة فرز الأسهم في JavaScript

مسار الفرز يختلف عن صفحة السعر: المرشّحات تُرسل كمعاملات استعلام، والنتيجة جدول يُعاد بناؤه مع كل تعديل، ولذلك يظهر التحدي هنا أكثر.

class MarketScreener {
  constructor(apiKey) {
    this.apiKey = apiKey;
  }

  async screenStocks(portalUrl, filters) {
    const params = new URLSearchParams(filters);
    const response = await fetch(`${portalUrl}/screener?${params}`);
    const html = await response.text();

    if (html.includes('cf-turnstile') || response.status === 403) {
      return this.solveAndScreen(portalUrl, filters, html);
    }

    return this.parseScreenerResults(html);
  }

  async solveAndScreen(portalUrl, filters, html) {
    const match = html.match(/data-sitekey="(0x[^"]+)"/);
    if (!match) throw new Error('Turnstile sitekey not found');

    const submitResp = await fetch('https://ocr.captchaai.com/in.php', {
      method: 'POST',
      body: new URLSearchParams({
        key: this.apiKey,
        method: 'turnstile',
        sitekey: match[1],
        pageurl: portalUrl,
        json: '1'
      })
    });
    const { request: taskId } = await submitResp.json();

    for (let i = 0; i < 60; i++) {
      await new Promise(r => setTimeout(r, 3000));
      const result = await fetch(
        `https://ocr.captchaai.com/res.php?key=${this.apiKey}&action=get&id=${taskId}&json=1`
      );
      const data = await result.json();
      if (data.status === 1) {
        const response = await fetch(`${portalUrl}/screener`, {
          method: 'POST',
          body: new URLSearchParams({
            ...filters,
            'cf-turnstile-response': data.request
          })
        });
        return this.parseScreenerResults(await response.text());
      }
    }
    throw new Error('Turnstile solve timed out');
  }

  parseScreenerResults(html) {
    const rows = [];
    const tableMatch = html.match(/<table[^>]*>[\s\S]*?<\/table>/i);
    if (!tableMatch) return rows;

    const rowMatches = tableMatch[0].matchAll(/<tr[^>]*>([\s\S]*?)<\/tr>/gi);
    for (const row of rowMatches) {
      const cells = [...row[1].matchAll(/<td[^>]*>([\s\S]*?)<\/td>/gi)]
        .map(m => m[1].replace(/<[^>]+>/g, '').trim());
      if (cells.length >= 4) {
        rows.push({
          symbol: cells[0],
          price: cells[1],
          change: cells[2],
          volume: cells[3]
        });
      }
    }
    return rows;
  }
}

// Usage
const screener = new MarketScreener('YOUR_API_KEY');
const results = await screener.screenStocks('https://finance.example.com', {
  sector: 'technology',
  marketCap: 'large',
  peRatio: '<25'
});

حلقة الاستطلاع تنتظر ثلاث ثوانٍ بين محاولة وأخرى وتتوقف عند سقف محدد بدل الانتظار المفتوح؛ الفشل الصريح أفضل من عملية معلّقة تُبقي thread مشغولاً.

إيقاع الجمع الموصى به لكل نوع بيانات

القاعدة خلف الجدول: كل زيادة في التكرار تُترجم إلى تحديات أكثر، لا إلى بيانات أدق.

كم thread تحتاج فعلياً؟

تسعير CaptchaAI قائم على عدد الـ threads المتزامنة لا على عدد عمليات الحل، وكل خطة تشمل عمليات حل غير محدودة ضمن threads الخطة. الـ thread الواحد تحدٍّ واحد قيد المعالجة، وفور انتهائه يلتقط التالي.

لذلك السؤال الصحيح ليس «كم رمزاً أجمع يومياً؟» بل «كم تحدياً يُعالَج في اللحظة نفسها؟». خط تسلسلي على 40 رمزاً يبقى ضمن عدد صغير من الـ threads، أما أربع بوابات بالتوازي مع إعادة محاولات فترفع الرقم بسرعة.

الخطط تبدأ من BASIC بسعر $15 شهرياً مع 5 threads، ثم STANDARD بسعر $30 مع 15 thread، وADVANCE بسعر $90 مع 50 thread، حتى VIP-3 بسعر $7,500 مع 5,000 thread. الأسعار بالدولار الأمريكي. ابدأ من الأصغر، وراقب ذروة التزامن أسبوعاً، ثم ارفع الخطة عند الحاجة.

أعطال متكررة في خطوط البيانات المالية

راقب يومياً ثلاثة أرقام فقط: معدل ظهور التحديات لكل مصدر، ووقت الحل الوسيط، وعدد الحقول الفارغة في كل دورة — وهو أبكر مؤشر على تغيّر بنية الصفحة.

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

هل يحل CaptchaAI كل أنواع التحقق التي تظهر على البوابات المالية؟

يغطي CaptchaAI عائلة reCAPTCHA بنسخها v2 وv3 وInvisible وEnterprise، وCloudflare Turnstile وCloudflare Challenge، وGeeTest v3، وتحديات الصور وOCR والشبكات الصورية وBLS، مع ثلاثة أنواع في مرحلة beta هي CaptchaFox وFriendly Captcha وLemin. في المقابل لا يدعم CaptchaAI حالياً hCaptcha ولا FunCaptcha، أما GeeTest v4 فمعلن كـ«قريباً» وغير متاح بعد.

كيف أقدّر تكلفة تشغيل يومي بهذا الحجم؟

احسب على أساس التزامن لا الحجم: عدّ التحديات التي تُعالَج في اللحظة نفسها عند الذروة لا عدد الرموز. ابدأ من BASIC، وراقب أسبوعاً، ثم انتقل إلى STANDARD أو ADVANCE إذا صار الطابور ينتظر.

هل أحتاج إلى وكلاء سكنيين لجمع بيانات الأسعار؟

ليس كخطوة أولى. خفّض إيقاع الطلبات وثبّت الجلسة أولاً، فأغلب حالات تكرار التحدي سببها الإيقاع لا العنوان؛ الخادم الوسيط يعالج سمعة عنوان IP وحدها.

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

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

أدلة ذات صلة

• البدء السريع: أول عملية حل خلال دقائق
• حل reCAPTCHA v2 عبر الـ API خطوة بخطوة
• التعامل مع Cloudflare Turnstile عبر الـ API
• حل GeeTest v3 عبر الـ API

افتح حسابك على captchaai.com، وابدأ بمصدر واحد وإيقاع محافظ، ثم وسّع التغطية بعد استقرار المؤشرات.