مراقبة مخزون التجزئة والتعامل مع اختبارات CAPTCHA

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

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

لماذا تظهر اختبارات CAPTCHA أثناء مراقبة المخزون؟

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

• معدل طلبات أعلى من سلوك زائر بشري على المسار نفسه.
• عنوان IP من مركز بيانات، أو عنوان سبق أن وُضعت عليه علامة.
• رؤوس طلب ناقصة أو غير متسقة مع بصمة المتصفح المعلنة.
• تكرار فتح صفحات عالية القيمة: إصدارات محدودة وصفحات الإضافة إلى السلة.

ضبط هذه العوامل يقلّل ظهور الاختبار ولا يلغيه؛ فليتعامل السكربت معه كحالة متوقعة لا كخطأ يوقف الدورة.

وتيرة مراقبة المخزون المناسبة لكل نوع منتج

ابدأ من الجدول التالي بدل فحص كل شيء كل دقيقة؛ الوتيرة الزائدة أسرع طريق لاستدعاء الاختبارات.

بالأرقام: الفحص كل 30 دقيقة يولّد 48 طلباً يومياً للمنتج الواحد، والفحص كل دقيقة يولّد 1,440 طلباً — أي ثلاثون ضعف فرص ظهور الاختبار.

أنواع الاختبارات الشائعة في منصات التجزئة

ما يلي ملخّص لما يُرصد عادةً على المنصات الكبرى؛ والإعدادات تتغيّر بين سوق وآخر، فتحقّق من الصفحة قبل بناء المراقب.

الأنواع الثلاثة هنا مدعومة في CaptchaAI عبر نمط الإرسال والاستطلاع نفسه، أي مسار واحد مهما اختلف المتجر.

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

يجمع الكود التالي المسار كاملاً: جلسة برؤوس واقعية، اكتشاف الاختبار داخل HTML، إرسال مفتاح الموقع إلى in.php، استطلاع النتيجة من res.php، ثم إعادة الطلب بالرمز المحلول في الحقل المناسب للنوع — cf-turnstile-response أو g-recaptcha-response.

import requests
import time
import re
import json
from datetime import datetime
from bs4 import BeautifulSoup

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


def solve_captcha(method, sitekey, pageurl, **kwargs):
    data = {
        "key": CAPTCHAAI_KEY, "method": method,
        "googlekey": sitekey, "pageurl": pageurl, "json": 1,
    }
    data.update(kwargs)
    resp = requests.post(f"{CAPTCHAAI_URL}/in.php", data=data)
    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,
        })
        r = result.json()
        if r["request"] != "CAPCHA_NOT_READY":
            return r["request"]
    raise TimeoutError("Timeout")


class RetailMonitor:
    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",
            "Accept": "text/html,application/xhtml+xml,*/*;q=0.8",
            "Accept-Language": "en-US,en;q=0.9",
        })

    def check_product(self, url):
        """Check single product's price and availability."""
        resp = self.session.get(url, timeout=30)

        # Handle CAPTCHA
        if self._has_captcha(resp.text):
            resp = self._solve_and_retry(resp.text, url)

        soup = BeautifulSoup(resp.text, "html.parser")
        return {
            "url": url,
            "title": self._text(soup, "h1, .product-title, #productTitle"),
            "price": self._text(soup, ".price, .a-price .a-offscreen, .prod-price"),
            "availability": self._text(soup, "#availability, .stock-status, .fulfillment"),
            "in_stock": self._check_stock(soup),
            "timestamp": datetime.now().isoformat(),
        }

    def monitor_products(self, product_urls, interval_sec=1800):
        """Continuously monitor products for changes."""
        history = {}

        while True:
            for url in product_urls:
                try:
                    current = self.check_product(url)

                    # Check for changes
                    prev = history.get(url)
                    if prev:
                        changes = self._detect_changes(prev, current)
                        if changes:
                            self._alert(current["title"], changes)

                    history[url] = current
                    time.sleep(3)

                except Exception as e:
                    print(f"Error checking {url}: {e}")

            print(f"Cycle complete: {len(product_urls)} products checked")
            time.sleep(interval_sec)

    def track_prices(self, product_urls, output_file="prices.json"):
        """Single price check across all products."""
        results = []
        for url in product_urls:
            try:
                data = self.check_product(url)
                results.append(data)
                time.sleep(3)
            except Exception as e:
                results.append({"url": url, "error": str(e)})

        with open(output_file, "w") as f:
            json.dump(results, f, indent=2)
        print(f"Tracked {len(results)} products → {output_file}")
        return results

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

    def _solve_and_retry(self, html, url):
        match = re.search(r'data-sitekey="([^"]+)"', html)
        if not match:
            return self.session.get(url)

        sitekey = match.group(1)
        if 'cf-turnstile' in html:
            token = solve_captcha("turnstile", sitekey, url)
            return self.session.post(url, data={"cf-turnstile-response": token})
        else:
            token = solve_captcha("userrecaptcha", sitekey, url)
            return self.session.post(url, data={"g-recaptcha-response": token})

    def _text(self, soup, selector):
        el = soup.select_one(selector)
        return el.get_text(strip=True) if el else ""

    def _check_stock(self, soup):
        stock_el = soup.select_one("#availability, .stock-status")
        if stock_el:
            text = stock_el.get_text(strip=True).lower()
            return "in stock" in text or "available" in text
        return None

    def _detect_changes(self, prev, current):
        changes = []
        if prev["price"] != current["price"]:
            changes.append(f"Price: {prev['price']} → {current['price']}")
        if prev["in_stock"] != current["in_stock"]:
            status = "In Stock" if current["in_stock"] else "Out of Stock"
            changes.append(f"Stock: → {status}")
        return changes

    def _alert(self, title, changes):
        print(f"ALERT [{title}]: {', '.join(changes)}")


# Usage
monitor = RetailMonitor(
    proxy="http://user:[email protected]:5000"
)

products = [
    "https://store.example.com/product/abc123",
    "https://store.example.com/product/def456",
    "https://store.example.com/product/ghi789",
]

# One-time price check
results = monitor.track_prices(products)

# Or continuous monitoring (every 30 min)
# monitor.monitor_products(products, interval_sec=1800)

الاستطلاع كل 5 ثوانٍ بحدّ أقصى 60 محاولة يمنع الانتظار اللانهائي، والفاصل بين المنتجات — 3 ثوانٍ — يبقي وتيرة الطلبات قريبة من السلوك الطبيعي.

مسح فئة كاملة بحثاً عن حالة التوفر

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

def scan_category(base_url, category, max_pages=20):
    """Scan an entire product category for stock status."""
    monitor = RetailMonitor(
        proxy="http://user:[email protected]:5000"
    )

    all_products = []
    for page in range(1, max_pages + 1):
        url = f"{base_url}/{category}?page={page}"
        resp = monitor.session.get(url, timeout=30)

        if monitor._has_captcha(resp.text):
            resp = monitor._solve_and_retry(resp.text, url)

        soup = BeautifulSoup(resp.text, "html.parser")
        items = soup.select(".product-card, .s-result-item")

        if not items:
            break

        for item in items:
            all_products.append({
                "name": monitor._text(item, ".product-name, .a-text-normal"),
                "price": monitor._text(item, ".price, .a-price"),
                "stock": monitor._text(item, ".stock, .a-color-success"),
                "url": item.select_one("a")["href"] if item.select_one("a") else "",
            })

        time.sleep(3)

    return all_products

مقارنة سعر المنتج بين عدة متاجر

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

def compare_product_across_stores(product_name, stores):
    """Compare prices across retailers for the same product."""
    results = []

    for store in stores:
        monitor = RetailMonitor(proxy=store.get("proxy"))
        search_url = f"{store['base_url']}/search?q={product_name}"

        try:
            resp = monitor.session.get(search_url, timeout=30)
            if monitor._has_captcha(resp.text):
                resp = monitor._solve_and_retry(resp.text, search_url)

            soup = BeautifulSoup(resp.text, "html.parser")
            first_result = soup.select_one(".product-card, .s-result-item")

            if first_result:
                results.append({
                    "store": store["name"],
                    "price": monitor._text(first_result, ".price"),
                    "in_stock": "in stock" in first_result.get_text().lower(),
                })
        except Exception as e:
            results.append({"store": store["name"], "error": str(e)})

        time.sleep(5)

    results.sort(key=lambda x: x.get("price", "zzzz"))
    return results

النتيجة قائمة مرتّبة تصلح لتقرير يومي أو لتنبيه عند نزول سعر منافس تحت عتبة تحدّدها.

سيناريو عملي: متابعة توفر منتج عبر ثلاثة أسواق عربية

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

1. اربط كل سوق بوكلاء داخل الدولة نفسها، ولا تخلط بينها في الدورة الواحدة.
2. ثبّت لغة الواجهة والعملة في رؤوس الطلب حتى لا تتغيّر بنية الصفحة وتكسر محددات CSS.
3. شغّل الدورات خارج ذروة التسوق المحلية، بعد منتصف الليل بتوقيت السوق مثلاً.
4. سجّل نوع الاختبار مع كل طلب فاشل؛ تكراره في سوق واحد مؤشر على وكلاء مستهلَكين لا على خلل في السكربت.

وحالة "غير متوفر" ليست دائماً حقيقية: أحياناً تكون قيداً جغرافياً، ومقارنة بين وكيلين من دولتين تكشف ذلك خلال دقائق.

كم thread تحتاج؟ حساب سريع للتكلفة

الحسبة تبدأ من عدد الاختبارات المتوقعة في الدقيقة، لا من عدد المنتجات. إذا انتهى 5% من الطلبات إلى اختبار، فإن دورة من 400 منتج بفاصل 3 ثوانٍ تُنتج نحو 20 اختباراً على مدى 20 دقيقة — حمل يستوعبه عدد صغير من الـ threads.

يعتمد CaptchaAI على خطط قائمة على عدد الـ threads المتزامنة مع حلول غير محدودة داخل كل thread، لا على الدفع مقابل كل عملية حل. تبدأ خطة BASIC من $15 شهرياً بخمسة threads، وهي كافية لقوائم المراقبة الصغيرة، بينما توفّر ADVANCE بسعر $90 شهرياً خمسين thread لفرق تشغّل عدة أسواق معاً. المقياس الحاسم هو ذروة التزامن لا المجموع اليومي.

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

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

كم thread أحتاج لمراقبة ألف منتج؟

العدد تحدّده ذروة التزامن لا عدد المنتجات. بفاصل 3 ثوانٍ، تستغرق دورة من 1,000 منتج نحو 50 دقيقة، ولا ينتهي إلى اختبار سوى جزء صغير منها. ابدأ صغيراً، راقب ساعة الذروة، ثم ارفع عدد الـ threads عند الحاجة.

لماذا يظهر الاختبار على بعض المنتجات دون غيرها؟

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

هل تختلف النتائج بين نسخة المتجر في كل دولة؟

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

ماذا لو انتقل المتجر إلى نوع اختبار غير مدعوم؟

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

أدلة ذات صلة

• ابدأ مع CaptchaAI: أول عملية حل خلال دقائق
• حلّ reCAPTCHA v2 عبر الـ API خطوة بخطوة

راقب مخزون متجرك دون انقطاع — أنشئ مفتاح CaptchaAI الخاص بك وشغّل أول دورة فحص اليوم.