مراقبة سلسلة التوريد والتعامل مع اختبارات CAPTCHA

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

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

أين تعترض اختبارات CAPTCHA بيانات سلسلة التوريد

حدّد أولاً أين يقف الاختبار في مسار البيانات. الأنماط التالية تتكرر في معظم مشاريع الرؤية التشغيلية:

زمن الحل يختلف كثيراً بين هذه الصفوف: الصور أقل من 0.5 ثانية، وCloudflare Turnstile أقل من 10 ثوانٍ، وسقف reCAPTCHA v2 أقل من 60 ثانية. لذلك افصل المصادر البطيئة في دفعة مستقلة.

رتّب مصادر سلسلة التوريد حسب حساسيتها للوقت

ليست كل بوابة تستحق فحصاً كل ساعة، والمشروع الذي يعامل كل المصادر بالوتيرة نفسها ينتهي إلى بوابات أكثر تشدداً:

القاعدة مختصرة: كلما ارتفعت كلفة قرار مبني على بيانات قديمة، ارتفعت الوتيرة المبرَّرة؛ وكلما زادت بلا مبرّر، زاد احتمال أن تفرض البوابة تحققاً على كل طلب. واقتصر على البوابات التي تملك عليها حساباً مصرّحاً به.

سيناريو من السوق: مستورد في جدة يتابع 40 مورّداً

خذ فريق مشتريات في جدة يدير 40 بوابة مورّد وستّ شركات شحن. ثمانية موردين فقط يوردون مكوّنات حرجة تُفحص كل ساعة، والبقية تكفيها دورة يومية. مع سقف حل reCAPTCHA v2 البالغ أقل من 60 ثانية، تنتهي دفعة الثمانية خلال دقيقتين تقريباً على 5 threads متزامنة — وهو ما تغطيه خطة BASIC — ‏$15 شهرياً مع 5 threads. والدورة اليومية لأربعين بوابة أكثر راحة على STANDARD — ‏$30 شهرياً مع 15 thread، بينما تنتقل الفرق متعددة الأسواق إلى ADVANCE — ‏$90 شهرياً مع 50 thread.

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

مراقب موحّد لبوابات الموردين بلغة Python

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

import requests
import time
import re
import json
import base64
from datetime import datetime

CAPTCHAAI_KEY = "YOUR_API_KEY"
CAPTCHAAI_URL = "https://ocr.captchaai.com"


def solve_recaptcha(sitekey, pageurl):
    resp = requests.post(f"{CAPTCHAAI_URL}/in.php", data={
        "key": CAPTCHAAI_KEY, "method": "userrecaptcha",
        "googlekey": sitekey, "pageurl": pageurl, "json": 1,
    })
    task_id = resp.json()["request"]
    for _ in range(60):
        time.sleep(5)
        result = requests.get(f"{CAPTCHAAI_URL}/res.php", params={
            "key": CAPTCHAAI_KEY, "action": "get",
            "id": task_id, "json": 1,
        })
        data = result.json()
        if data["request"] != "CAPCHA_NOT_READY":
            return data["request"]
    raise TimeoutError("Timeout")


def solve_turnstile(sitekey, pageurl):
    resp = requests.post(f"{CAPTCHAAI_URL}/in.php", data={
        "key": CAPTCHAAI_KEY, "method": "turnstile",
        "sitekey": sitekey, "pageurl": pageurl, "json": 1,
    })
    task_id = resp.json()["request"]
    for _ in range(60):
        time.sleep(5)
        result = requests.get(f"{CAPTCHAAI_URL}/res.php", params={
            "key": CAPTCHAAI_KEY, "action": "get",
            "id": task_id, "json": 1,
        })
        data = result.json()
        if data["request"] != "CAPCHA_NOT_READY":
            return data["request"]
    raise TimeoutError("Timeout")


def solve_image(image_bytes):
    img_b64 = base64.b64encode(image_bytes).decode()
    resp = requests.post(f"{CAPTCHAAI_URL}/in.php", data={
        "key": CAPTCHAAI_KEY, "method": "base64",
        "body": img_b64, "json": 1,
    })
    task_id = resp.json()["request"]
    for _ in range(20):
        time.sleep(3)
        result = requests.get(f"{CAPTCHAAI_URL}/res.php", params={
            "key": CAPTCHAAI_KEY, "action": "get",
            "id": task_id, "json": 1,
        })
        data = result.json()
        if data["request"] != "CAPCHA_NOT_READY":
            return data["request"]
    raise TimeoutError("Timeout")


class SupplyChainMonitor:
    def __init__(self, suppliers, proxy=None):
        self.suppliers = suppliers
        self.session = requests.Session()
        if proxy:
            self.session.proxies = {"http": proxy, "https": proxy}
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 Chrome/126.0.0.0 Safari/537.36",
        })

    def check_all(self):
        """Check inventory and pricing across all suppliers."""
        report = {
            "timestamp": datetime.now().isoformat(),
            "suppliers": {},
        }

        for supplier in self.suppliers:
            try:
                data = self._check_supplier(supplier)
                report["suppliers"][supplier["name"]] = {
                    "status": "success",
                    "data": data,
                }
            except Exception as e:
                report["suppliers"][supplier["name"]] = {
                    "status": "error",
                    "error": str(e),
                }
            time.sleep(3)

        return report

    def _check_supplier(self, supplier):
        url = supplier["url"]
        resp = self.session.get(url, timeout=30)

        # Handle CAPTCHA based on type
        captcha_type = supplier.get("captcha_type")
        if captcha_type and self._has_captcha(resp.text):
            resp = self._solve_captcha(resp, url, supplier)

        from bs4 import BeautifulSoup
        soup = BeautifulSoup(resp.text, "html.parser")

        return {
            "products": self._extract_inventory(soup),
            "last_updated": self._extract_date(soup),
        }

    def _has_captcha(self, html):
        return any(tag in html.lower() for tag in [
            'data-sitekey', 'g-recaptcha', 'cf-turnstile', 'captcha',
        ])

    def _solve_captcha(self, resp, url, supplier):
        captcha_type = supplier.get("captcha_type", "recaptcha")
        sitekey = supplier.get("sitekey", "")

        if not sitekey:
            match = re.search(r'data-sitekey="([^"]+)"', resp.text)
            sitekey = match.group(1) if match else ""

        if captcha_type == "turnstile":
            token = solve_turnstile(sitekey, url)
            return self.session.post(url, data={"cf-turnstile-response": token})
        elif captcha_type == "image":
            match = re.search(r'src="(/captcha[^"]+)"', resp.text)
            if match:
                img_resp = self.session.get(url.rstrip("/") + match.group(1))
                answer = solve_image(img_resp.content)
                return self.session.post(url, data={"captcha": answer})
        else:
            token = solve_recaptcha(sitekey, url)
            return self.session.post(url, data={"g-recaptcha-response": token})

        return resp

    def _extract_inventory(self, soup):
        items = []
        for row in soup.select("table.inventory tr, .product-row"):
            cols = row.select("td, .col")
            if len(cols) >= 3:
                items.append({
                    "sku": cols[0].get_text(strip=True),
                    "stock": cols[1].get_text(strip=True),
                    "price": cols[2].get_text(strip=True),
                })
        return items

    def _extract_date(self, soup):
        date_el = soup.select_one(".last-updated, .update-time")
        return date_el.get_text(strip=True) if date_el else ""


# Configure suppliers
suppliers = [
    {
        "name": "Supplier A",
        "url": "https://supplier-a.example.com/inventory",
        "captcha_type": "recaptcha",
        "sitekey": "6Lc_xxxxxxx",
    },
    {
        "name": "Carrier B",
        "url": "https://carrier-b.example.com/rates",
        "captcha_type": "turnstile",
        "sitekey": "0x4AAAAAAA_xxx",
    },
    {
        "name": "Manufacturer C",
        "url": "https://manufacturer-c.example.com/catalog",
        "captcha_type": "image",
    },
]

monitor = SupplyChainMonitor(
    suppliers=suppliers,
    proxy="http://user:[email protected]:5000",
)
report = monitor.check_all()
print(json.dumps(report, indent=2))

ثلاث نقاط تستحق الانتباه:

• كل مورّد مغلَّف داخل try/except مستقل، فالبوابة المتعطلة تُسجَّل كخطأ بدل أن توقف الدفعة.
• مفتاح الموقع يُقرأ من الصفحة عند غيابه من الإعدادات، فلا ينكسر السكربت حين يغيّره المورّد.
• الفاصل الزمني بين مورّد وآخر مقصود؛ إزالته أسرع طريقة لفرض اختبار على كل طلب.

قراءة أسعار الشحن خلف Cloudflare Turnstile

أسعار النقل تتغيّر خلال اليوم نفسه، ومعظم بوابات الناقلين تضع Turnstile أمام نموذج التسعير لا أمام الصفحة الرئيسية. يُقرأ مفتاح الموقع من النموذج أولاً، ثم يُرسل الرمز مع بيانات الشحنة:

class ShippingRateTracker:
    def __init__(self, proxy=None):
        self.session = requests.Session()
        if proxy:
            self.session.proxies = {"http": proxy, "https": proxy}
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 Chrome/126.0.0.0 Safari/537.36",
        })

    def get_rates(self, carrier_url, origin, destination, weight):
        """Fetch shipping rates, handling Turnstile CAPTCHA."""
        resp = self.session.get(carrier_url, timeout=30)

        sitekey_match = re.search(r'data-sitekey="([^"]+)"', resp.text)
        if sitekey_match:
            token = solve_turnstile(sitekey_match.group(1), carrier_url)
            resp = self.session.post(carrier_url, data={
                "origin": origin,
                "destination": destination,
                "weight": weight,
                "cf-turnstile-response": token,
            })

        if resp.status_code == 200:
            return resp.json().get("rates", [])
        return []

الرمز يُرسل في الحقل cf-turnstile-response داخل الطلب الذي يحمل المنشأ والوجهة والوزن. الفصل بين الخطوتين يُبطله عملياً لأنه صالح لاستخدام واحد ولفترة قصيرة، وهذا سبب شائع لعودة الأسعار فارغة.

تنبيهات المخزون عند حدّ إعادة الطلب

الفائدة الحقيقية تظهر عند التنبيه لا عند الجمع. الدالة التالية تقارن كل دورة بسابقتها وتُطلق تنبيهاً واحداً عند عبور الحدّ نزولاً:

def monitor_with_alerts(monitor, alert_thresholds, check_interval=3600):
    """Continuously monitor and alert on inventory changes."""
    previous_data = {}

    while True:
        report = monitor.check_all()

        for supplier, info in report["suppliers"].items():
            if info["status"] != "success":
                continue

            for product in info["data"].get("products", []):
                sku = product["sku"]
                stock = product.get("stock", "")

                # Parse stock level
                try:
                    stock_qty = int(re.sub(r'\D', '', stock))
                except ValueError:
                    continue

                key = f"{supplier}:{sku}"
                prev_qty = previous_data.get(key, stock_qty)

                threshold = alert_thresholds.get(sku, 10)
                if stock_qty < threshold and prev_qty >= threshold:
                    print(f"ALERT: {supplier} - {sku} dropped to {stock_qty}")

                previous_data[key] = stock_qty

        time.sleep(check_interval)

في التشغيل الفعلي استبدل القاموس previous_data بتخزين دائم مثل Redis أو ملف JSON، وإلا ضاعت الحالة عند كل إعادة تشغيل. وحدّد عتبة لكل SKU: مكوّن مهلته ستة أسابيع يحتاج إنذاراً أبكر من صنف يصل خلال يومين.

قلّل عدد اختبارات CAPTCHA قبل حلّها

كل اختبار لا يظهر أصلاً هو ثوانٍ موفّرة من الدورة. أربعة إجراءات تصنع معظم الفارق:

• وزّع الطلبات على عناوين IP متعددة بدل تكرارها من عنوان واحد؛ التفاصيل في أفضل ممارسات الوكيل السكني الدوّار.
• ثبّت الجلسة على البوابات التي تتطلب تسجيل دخول، وبدّلها في القراءة العامة — الموازنة مشروحة في الجلسات الثابتة مقابل الدوّارة.
• اجعل وتيرة الفحص متغيّرة قليلاً حول القيمة المستهدفة بدل رأس الساعة تماماً.
• خزّن نتيجة كل فحص مع رمز الحالة وسببه؛ معظم ما يُسمّى «عطل CAPTCHA» يتبيّن أنه تغيير في الصفحة أو انتهاء جلسة.

أعطال متكررة أثناء المراقبة وحلولها

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

كم thread أحتاج لمراقبة 40 بوابة مورّد؟

احسب العدد بناءً على البوابات التي يجب أن تنتهي داخل نافذة زمنية واحدة، لا على إجمالي الموردين. دورة يومية موزّعة على أربعين بوابة تكفيها خطة STANDARD — ‏$30 شهرياً مع 15 thread.

هل يُبطئ حلّ CAPTCHA دورة المراقبة كثيراً؟

يعتمد على النوع: الصور أقل من 0.5 ثانية، وCloudflare Turnstile أقل من 10 ثوانٍ، وسقف reCAPTCHA v2 أقل من 60 ثانية. ضع بوابات reCAPTCHA v2 في دفعة مستقلة حتى لا تفرض إيقاعها على البقية.

هل يدعم CaptchaAI كل الاختبارات في بوابات الموردين؟

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

ماذا أفعل إذا كانت بيانات المورّد خلف تسجيل دخول؟

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

كيف أعرف أن الرقم المعروض على اللوحة قديم؟

سجّل مع كل قيمة زمن آخر فحص ناجح لا زمن آخر محاولة؛ وحين يتجاوز الفارق ضعف الوتيرة المتوقعة، تعامل مع الرقم كبيانات منتهية الصلاحية.

ابدأ بمفتاح CaptchaAI على بوابة واحدة تعرف بياناتها جيداً، وقِس نسبة الدورات المكتملة أسبوعاً كاملاً قبل إضافة بقية الموردين.