مراقبة أسعار التجارة الإلكترونية عند ظهور اختبارات CAPTCHA

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

أين يظهر CAPTCHA في هذا النوع من المراقبة

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

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

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

الهندسة المعمارية

Scheduler (every 30 min)
    → URL Queue
        → Scraper Workers (5-10 concurrent)
            → Fetch page
            → CAPTCHA detected?
                → Yes → CaptchaAI → Solve → Retry page
                → No → Parse prices
            → Store in database
        → Alert on price changes

التنفيذ

مراقب الأسعار (بايثون)

import requests
import time
import re
import json
import os
from datetime import datetime

API_KEY = os.environ["CAPTCHAAI_API_KEY"]
BASE_URL = "https://ocr.captchaai.com"


def solve_captcha(method, params):
    params["key"] = API_KEY
    params["method"] = method

    resp = requests.get(f"{BASE_URL}/in.php", params=params)
    if not resp.text.startswith("OK|"):
        raise Exception(f"Submit failed: {resp.text}")

    task_id = resp.text.split("|")[1]

    for _ in range(60):
        time.sleep(5)
        result = requests.get(f"{BASE_URL}/res.php", params={
            "key": API_KEY, "action": "get", "id": task_id,
        })
        if result.text == "CAPCHA_NOT_READY":
            continue
        if result.text.startswith("OK|"):
            return result.text.split("|", 1)[1]
        raise Exception(f"Solve failed: {result.text}")

    raise TimeoutError("CAPTCHA solve timed out")


def fetch_with_captcha(url, session):
    """Fetch a page, solving CAPTCHAs if encountered."""
    resp = session.get(url)

    # Check for reCAPTCHA
    match = re.search(r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', resp.text)
    if match:
        site_key = match.group(1)
        token = solve_captcha("userrecaptcha", {
            "googlekey": site_key,
            "pageurl": url,
        })
        resp = session.post(url, data={"g-recaptcha-response": token})

    # Check for Turnstile
    match = re.search(
        r'class="cf-turnstile"[^>]*data-sitekey=["\']([^"\']+)', resp.text
    )
    if match:
        site_key = match.group(1)
        token = solve_captcha("turnstile", {
            "sitekey": site_key,
            "pageurl": url,
        })
        resp = session.post(url, data={"cf-turnstile-response": token})

    return resp


def extract_price(html, selectors):
    """Extract price from HTML using regex patterns."""
    for pattern in selectors:
        match = re.search(pattern, html)
        if match:
            price_str = match.group(1).replace(",", "")
            return float(price_str)
    return None


def monitor_prices(products):
    """Monitor prices for a list of products."""
    session = requests.Session()
    session.headers["User-Agent"] = (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 Chrome/120.0.0.0"
    )

    results = []
    for product in products:
        try:
            resp = fetch_with_captcha(product["url"], session)
            price = extract_price(resp.text, product["selectors"])

            results.append({
                "name": product["name"],
                "url": product["url"],
                "price": price,
                "timestamp": datetime.utcnow().isoformat(),
                "status": "ok",
            })
            print(f"  {product['name']}: ${price}")

        except Exception as e:
            results.append({
                "name": product["name"],
                "url": product["url"],
                "price": None,
                "timestamp": datetime.utcnow().isoformat(),
                "status": f"error: {e}",
            })
            print(f"  {product['name']}: ERROR - {e}")

    return results


# Define products to monitor
products = [
    {
        "name": "Wireless Headphones",
        "url": "https://example.com/product/headphones",
        "selectors": [
            r'class="price"[^>]*>\$?([\d,]+\.?\d*)',
            r'itemprop="price" content="([\d.]+)"',
        ],
    },
    {
        "name": "Bluetooth Speaker",
        "url": "https://example.com/product/speaker",
        "selectors": [
            r'class="price"[^>]*>\$?([\d,]+\.?\d*)',
        ],
    },
]

print("Starting price check...")
results = monitor_prices(products)

# Save results
with open("prices.json", "w") as f:
    json.dump(results, f, indent=2)

تنفيذ Node.js

const axios = require("axios");
const cheerio = require("cheerio");

const API_KEY = process.env.CAPTCHAAI_API_KEY;

async function solveCaptcha(method, params) {
  params.key = API_KEY;
  params.method = method;

  const submit = await axios.get("https://ocr.captchaai.com/in.php", {
    params,
  });
  const taskId = String(submit.data).split("|")[1];

  for (let i = 0; i < 60; i++) {
    await new Promise((r) => setTimeout(r, 5000));
    const poll = await axios.get("https://ocr.captchaai.com/res.php", {
      params: { key: API_KEY, action: "get", id: taskId },
    });
    const text = String(poll.data);
    if (text === "CAPCHA_NOT_READY") continue;
    if (text.startsWith("OK|")) return text.split("|").slice(1).join("|");
    throw new Error(text);
  }
  throw new Error("Timeout");
}

async function monitorPrice(url) {
  const resp = await axios.get(url);
  const $ = cheerio.load(resp.data);

  // Check for reCAPTCHA
  const siteKey = $(".g-recaptcha").attr("data-sitekey");
  if (siteKey) {
    const token = await solveCaptcha("userrecaptcha", {
      googlekey: siteKey,
      pageurl: url,
    });
    // Re-fetch with token
    const formResp = await axios.post(url, { "g-recaptcha-response": token });
    return cheerio.load(formResp.data);
  }

  const price = $('[itemprop="price"]').attr("content") || $(".price").text();
  return parseFloat(price.replace(/[^0-9.]/g, ""));
}

الجدولة

قم بإجراء عمليات التحقق كل 30 دقيقة باستخدام cron:

# crontab -e
*/30 * * * * cd /opt/monitor && python price_monitor.py >> /var/log/prices.log 2>&1

أو استخدم مكتبة Python schedule:

import schedule

schedule.every(30).minutes.do(lambda: monitor_prices(products))

while True:
    schedule.run_pending()
    time.sleep(60)

تقدير التكلفة

لا تؤدي جميع عمليات تحميل الصفحات إلى تشغيل اختبارات CAPTCHA. وقد تكون التكاليف الفعلية أقل بنسبة 50-70%.

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

كيف يمكنني اكتشاف تغيرات الأسعار؟

مقارنة الأسعار الحالية بالقيم المخزنة. يساعد التنبيه بالتغييرات التي تزيد عن 5% على تصفية الضوضاء الناتجة عن التقلبات الطفيفة.

هل سيتم حظري حتى مع حل اختبار CAPTCHA؟

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

هل يمكنني مراقبة الأسعار عبر عملات متعددة؟

نعم. قم بتحليل رمز العملة بجانب السعر. يعمل CaptchaAI عالميًا بغض النظر عن لغة الموقع المستهدف.

أدلة ذات صلة

• البدء السريع مع CaptchaAI: حلّ أول كابتشا في 5 دقائق
• DynamoDB لـ CAPTCHA بدون خادم، حل التتبع