أنشئ لوحة معلومات لتحليل المنافسين باستخدام CaptchaAI

مراقبة المنافسين يدوياً لا تصمد أمام سوق يتغيّر يوماً بيوم. في هذا الدليل تبني لوحة تحليل بلغة Python تجمع أسعار المنافسين وقوائم منتجاتهم وميزاتهم آلياً، تخزّن كل قراءة في قاعدة بيانات SQLite لبناء سجل تاريخي، ثم تولّد تقارير مقارنة واتجاهات جاهزة للعرض. وحين تعترض صفحة تسعير منافس اختبار reCAPTCHA v2، يتكفّل CaptchaAI بحلّه ضمن المسار نفسه دون توقّف يدوي.

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

متى تحتاج إلى أتمتة مراقبة المنافسين؟

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

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

المتطلبات الأساسية

قبل البدء، جهّز البيئة التالية:

• Python 3.9 أو أحدث، مع مكتبتَي requests وbeautifulsoup4.
• حساب على CaptchaAI ومفتاح الـ API الخاص به من لوحة تحكم الحساب.
• إلمام أساسي بمحدّدات CSS لقراءة عناصر الصفحة.
• قاعدة بيانات SQLite — وهي مضمّنة داخل Python فلا تحتاج خادماً منفصلاً.

خزّن مفتاح الـ API في متغيّر بيئة باسم CAPTCHAAI_API_KEY بدل كتابته داخل الكود، حفاظاً على أمان بيانات الاعتماد.

نظرة على بنية النظام

يتكوّن الإعداد من أربع طبقات متتابعة: مستخرِج يقرأ صفحات المنافسين، ووحدة حل تتعامل مع reCAPTCHA v2 عند ظهوره، ومخزن SQLite يحتفظ بالسجل التاريخي، ثم مولّد تقارير يقرأ من المخزن:

Competitor Sites ──> CAPTCHA Solver ──> Data Extractors
                                             │
                                        SQLite Store
                                             │
                                      Dashboard Report

كل طبقة مستقلة عن الأخرى، ما يعني أنك تستطيع تبديل مصدر البيانات أو صيغة التقرير دون المساس ببقية النظام.

الخطوة 1: تصميم نموذج البيانات وتخزينه في SQLite

نبدأ بطبقة التخزين لأنها تحدّد شكل كل ما يليها. الفئة CompetitorData تمثّل قراءة واحدة — منافس، ومقياس، وقيمة نصية، وقيمة رقمية اختيارية للمقارنة الحسابية، ورابط المصدر، وطابع زمني يُضبط تلقائياً عند الإنشاء. أما CompetitorDB فتدير جدول metrics وتتيح ثلاث عمليات: الحفظ، وقراءة السجل التاريخي لمقياس معيّن، والمقارنة اللحظية بين المنافسين:

# models.py
import sqlite3
from datetime import datetime
from dataclasses import dataclass
from typing import Optional


@dataclass
class CompetitorData:
    competitor: str
    metric: str
    value: str
    numeric_value: Optional[float] = None
    url: str = ""
    scraped_at: str = ""

    def __post_init__(self):
        if not self.scraped_at:
            self.scraped_at = datetime.now().isoformat()


class CompetitorDB:
    def __init__(self, path="competitor_data.db"):
        self.conn = sqlite3.connect(path)
        self._init()

    def _init(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS metrics (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                competitor TEXT,
                metric TEXT,
                value TEXT,
                numeric_value REAL,
                url TEXT,
                scraped_at TEXT
            )
        """)
        self.conn.commit()

    def save(self, data: CompetitorData):
        self.conn.execute(
            """INSERT INTO metrics
               (competitor, metric, value, numeric_value, url, scraped_at)
               VALUES (?, ?, ?, ?, ?, ?)""",
            (data.competitor, data.metric, data.value,
             data.numeric_value, data.url, data.scraped_at),
        )
        self.conn.commit()

    def get_history(self, competitor, metric, limit=30):
        cursor = self.conn.execute(
            """SELECT value, numeric_value, scraped_at
               FROM metrics
               WHERE competitor = ? AND metric = ?
               ORDER BY scraped_at DESC LIMIT ?""",
            (competitor, metric, limit),
        )
        return cursor.fetchall()

    def latest_comparison(self, metric):
        cursor = self.conn.execute(
            """SELECT competitor, value, numeric_value, MAX(scraped_at) as latest
               FROM metrics WHERE metric = ?
               GROUP BY competitor ORDER BY numeric_value""",
            (metric,),
        )
        return cursor.fetchall()

لاحظ دالة latest_comparison: تجمع الصفوف حسب المنافس وتختار أحدث طابع زمني عبر MAX(scraped_at)، فلا تشوّش على التقرير إدخالاتٌ مكرّرة من تشغيلات سابقة. هذا التصميم البسيط يكفي لآلاف القراءات قبل الحاجة إلى قاعدة بيانات أكبر.

الخطوة 2: حل reCAPTCHA v2 عبر CaptchaAI

حين تكتشف صفحة التسعير حركة آلية، قد تعرض اختبار reCAPTCHA v2. الوحدة التالية تفحص كود الصفحة، فإن وجدت data-sitekey أرسلت المهمة إلى CaptchaAI عبر نقطة النهاية in.php بالأسلوب userrecaptcha، ثم استطلعت النتيجة من res.php حتى تجهز:

# solver.py
import requests
import time
import re
import os


class CaptchaSolver:
    def __init__(self):
        self.api_key = os.environ["CAPTCHAAI_API_KEY"]

    def solve_if_needed(self, session, url, html):
        if "data-sitekey" not in html:
            return html

        match = re.search(r'data-sitekey="([^"]+)"', html)
        if not match:
            return html

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

        time.sleep(15)
        for _ in range(24):
            resp = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get",
                "id": task_id, "json": 1,
            }, timeout=15)
            data = resp.json()
            if data.get("status") == 1:
                post_resp = session.post(url, data={
                    "g-recaptcha-response": data["request"],
                }, timeout=30)
                return post_resp.text
            if data["request"] != "CAPCHA_NOT_READY":
                raise RuntimeError(data["request"])
            time.sleep(5)

        raise TimeoutError("CAPTCHA solve timeout")

المسار هنا أربع خطوات ثابتة: أرسِل مفتاح الموقع ورابط الصفحة، احفظ معرّف المهمة، انتظر 15 ثانية ثم استطلع النتيجة دورياً، وأخيراً أعِد إرسال النموذج ومعه الرمز g-recaptcha-response. يعالج CaptchaAI اختبار reCAPTCHA v2 ضمن أنواعه المدعومة رسمياً، ويتبع تسعيره نموذج الـ threads المتزامنة برصيد حلول غير محدود لكل thread خلال الشهر — أي لا رسوم لكل عملية حل ولا حدود يومية.

الخطوة 3: بناء وحدة جمع بيانات المنافس

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

# scraper.py
import requests
import re
from bs4 import BeautifulSoup
from solver import CaptchaSolver
from models import CompetitorData


class CompetitorScraper:
    def __init__(self):
        self.solver = CaptchaSolver()
        self.session = requests.Session()
        self.session.headers["User-Agent"] = (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 Chrome/125.0.0.0 Safari/537.36"
        )

    def scrape_pricing(self, competitor_name, url, plan_selector, price_selector):
        html = self._fetch(url)
        soup = BeautifulSoup(html, "html.parser")
        plans = soup.select(plan_selector)
        data = []

        for plan in plans:
            name_el = plan.select_one("h3, h2, .plan-name")
            price_el = plan.select_one(price_selector)

            if not name_el or not price_el:
                continue

            price_text = price_el.get_text(strip=True)
            match = re.search(r'[\d,.]+', price_text)
            numeric = float(match.group().replace(",", "")) if match else None

            data.append(CompetitorData(
                competitor=competitor_name,
                metric=f"price_{name_el.get_text(strip=True).lower().replace(' ', '_')}",
                value=price_text,
                numeric_value=numeric,
                url=url,
            ))

        return data

    def scrape_features(self, competitor_name, url, feature_list_selector):
        html = self._fetch(url)
        soup = BeautifulSoup(html, "html.parser")
        features = soup.select(f"{feature_list_selector} li")

        return [
            CompetitorData(
                competitor=competitor_name,
                metric="feature",
                value=f.get_text(strip=True),
                url=url,
            )
            for f in features if f.get_text(strip=True)
        ]

    def scrape_product_count(self, competitor_name, url, count_selector):
        html = self._fetch(url)
        soup = BeautifulSoup(html, "html.parser")
        el = soup.select_one(count_selector)

        if el:
            text = el.get_text(strip=True)
            match = re.search(r'[\d,]+', text)
            if match:
                count = int(match.group().replace(",", ""))
                return CompetitorData(
                    competitor=competitor_name,
                    metric="product_count",
                    value=text,
                    numeric_value=count,
                    url=url,
                )
        return None

    def _fetch(self, url):
        resp = self.session.get(url, timeout=20)
        return self.solver.solve_if_needed(self.session, url, resp.text)

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

الخطوة 4: توليد تقارير المقارنة والاتجاهات

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

# report.py
from models import CompetitorDB


def generate_report(db: CompetitorDB, metrics):
    lines = ["=" * 60, "Competitor Analysis Report", "=" * 60, ""]

    for metric in metrics:
        results = db.latest_comparison(metric)
        if not results:
            continue

        lines.append(f"--- {metric.replace('_', ' ').title()} ---")
        for comp, value, numeric, ts in results:
            marker = ""
            if numeric is not None:
                marker = f" (${numeric:,.2f})" if "price" in metric else f" ({numeric:,.0f})"
            lines.append(f"  {comp}: {value}{marker}")
        lines.append("")

    return "\n".join(lines)


def generate_trend(db: CompetitorDB, competitor, metric, periods=10):
    history = db.get_history(competitor, metric, limit=periods)
    if not history:
        return f"No data for {competitor} — {metric}"

    lines = [f"Trend: {competitor} — {metric}", "-" * 40]
    for value, numeric, ts in reversed(history):
        date = ts[:10]
        lines.append(f"  {date}: {value}")

    return "\n".join(lines)

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

الخطوة 5: تشغيل المشغّل الرئيسي وجدولته

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

# main.py
import time
from models import CompetitorDB
from scraper import CompetitorScraper
from report import generate_report

COMPETITORS = [
    {
        "name": "Competitor A",
        "pricing_url": "https://competitor-a.example.com/pricing",
        "plan_selector": ".pricing-plan",
        "price_selector": ".price",
    },
    {
        "name": "Competitor B",
        "pricing_url": "https://competitor-b.example.com/pricing",
        "plan_selector": ".plan-card",
        "price_selector": ".plan-price",
    },
]


def main():
    db = CompetitorDB()
    scraper = CompetitorScraper()

    for comp in COMPETITORS:
        print(f"Scraping {comp['name']}...")

        try:
            pricing = scraper.scrape_pricing(
                comp["name"], comp["pricing_url"],
                comp["plan_selector"], comp["price_selector"],
            )
            for p in pricing:
                db.save(p)
                print(f"  {p.metric}: {p.value}")
        except Exception as e:
            print(f"  Error: {e}")

        time.sleep(5)

    # Generate report
    metrics = ["price_basic", "price_pro", "price_enterprise", "product_count"]
    report = generate_report(db, metrics)
    print(report)

    with open("competitor_report.txt", "w") as f:
        f.write(report)


if __name__ == "__main__":
    main()

انتبه إلى time.sleep(5) بين المنافسين: فاصل مهذّب يخفّف الضغط على مواقعهم ويقلّل فرص إطلاق أنظمة كشف الحركة الآلية. لتشغيل اللوحة يومياً دون تدخّل، اجدُل main.py عبر cron على Linux أو Task Scheduler على Windows؛ ومع كل تشغيل يتراكم السجل التاريخي وتتّضح الاتجاهات أكثر.

اختيار خطة CaptchaAI المناسبة لحجم المراقبة

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

• BASIC بسعر 15 دولاراً شهرياً و5 threads: كافٍ للوحة تراقب عدداً محدوداً من المنافسين بجدولة يومية متسلسلة.
• STANDARD بسعر 30 دولاراً شهرياً و15 thread: يناسب توسّع القائمة وزيادة وتيرة الجمع.
• ADVANCE بسعر 90 دولاراً شهرياً و50 thread: للجمع المتوازي عبر عشرات المواقع في آن واحد.

بما أن السجل التاريخي يتراكم من تشغيلات متسلسلة، تكفي الخططُ الأدنى معظم لوحات المراقبة؛ وارفع عدد الـ threads فقط حين تحتاج جمعاً متزامناً فعلياً. الأسعار بالدولار الأمريكي وقد تتغيّر، فراجع صفحة التسعير الرسمية قبل الاعتماد.

استكشاف الأخطاء الشائعة وحلولها

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

ما أنواع اختبارات CAPTCHA التي يغطّيها هذا الإعداد؟

يتعامل الكود هنا مع reCAPTCHA v2 عبر الأسلوب userrecaptcha. ويدعم CaptchaAI رسمياً أنواعاً أخرى قد تظهر على صفحات المنافسين مثل reCAPTCHA v3 وCloudflare Turnstile وCloudflare Challenge وGeeTest v3 واختبارات الصور، في حين لا يدعم hCaptcha أو FunCaptcha حالياً.

كيف أخفّف احتمال حظر عنوان IP أثناء الجمع؟

وزّع الطلبات بفواصل زمنية كما في time.sleep(5)، وثبّت User-Agent واقعياً، وأبقِ جلسة واحدة لكل تشغيل. ومع الحجم الأكبر أضِف خادماً وسيطاً (proxy) دوّاراً لتوزيع الطلبات على عناوين مختلفة.

كيف أجدول التشغيل اليومي تلقائياً؟

اربط main.py بمهمة مجدولة: cron على Linux أو Task Scheduler على Windows، بوتيرة يومية أو كل بضع ساعات بحسب سرعة تغيّر أسعار سوقك.

كيف أتلقّى تنبيهاً فور تغيّر سعر منافس؟

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

ما خطة CaptchaAI الأنسب مع تزايد عدد المنافسين؟

ابدأ بخطة BASIC للوحة صغيرة متسلسلة، وانتقل إلى STANDARD أو ADVANCE حين تحتاج جمعاً متوازياً عبر عدد أكبر من المواقع في وقت واحد.

أدلة ذات صلة

• بناء خط أنابيب اختبار آلي مع CaptchaAI
• الحفاظ على جلسة المتصفح في سير عمل CAPTCHA

راقب أسعار منافسيك على نطاق واسع — ابدأ الآن مع CaptchaAI.