أنشئ مجمعًا لقائمة الوظائف باستخدام CaptchaAI

تنتشر إعلانات الوظائف على عشرات اللوحات، وكل لوحة تعرض بياناتها بصيغة مختلفة، وبعضها يضع اختبار reCAPTCHA v2 أمام صفحة النتائج قبل أن تصل إلى القائمة. الحل هو مُجمِّع واحد يوحّد هذه المصادر في مكان واحد. في هذا الدليل تبنيه بلغة Python عبر أربع مراحل مترابطة: اكشط صفحات البحث، وحُل اختبار CAPTCHA عند ظهوره عبر CaptchaAI، ووحّد الحقول المتباينة في نموذج واحد، ثم خزّن النتائج في قاعدة بيانات SQLite جاهزة للبحث.

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

قبل أن تبدأ: المتطلبات

تحتاج قبل كتابة أول سطر إلى ثلاثة أشياء:

• بيئة Python 3.9 أو أحدث، مع تثبيت مكتبتَي requests وbeautifulsoup4 عبر pip.
• حساب على captchaai.com ومفتاح الـ API الخاص بك، مخزَّن في متغيّر البيئة CAPTCHAAI_API_KEY بدل كتابته داخل الشيفرة.
• إلمام أساسي بمحددات CSS لقراءة بطاقات الوظائف من صفحة كل لوحة.

بنية المُجمِّع

يمر كل طلب بأربع محطات: مصدر البيانات (لوحات الوظائف)، ثم طبقة الكشط التي تحل اختبار CAPTCHA عند الحاجة، ثم مُوحِّد الحقول، وأخيرًا قاعدة البيانات. يلخّص المخطط التالي هذا التدفّق:

[Job Board A] ──┐
[Job Board B] ──┼──> Scraper + CAPTCHA Solver ──> Normalizer ──> SQLite DB
[Job Board C] ──┘

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

نمذجة بيانات الوظيفة وتخزينها

نبدأ بتعريف شكل الوظيفة الواحدة وطريقة حفظها. الفئة JobListing تمثّل إعلانًا واحدًا بحقوله — العنوان والشركة والموقع والرابط والمصدر ونطاق الراتب — بينما تتكفّل JobDatabase بإنشاء الجدول والإدراج والبحث:

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


@dataclass
class JobListing:
    title: str
    company: str
    location: str
    url: str
    source: str
    salary_min: Optional[float] = None
    salary_max: Optional[float] = None
    posted_date: Optional[str] = None
    description: str = ""
    tags: list = field(default_factory=list)
    scraped_at: str = field(default_factory=lambda: datetime.now().isoformat())


class JobDatabase:
    def __init__(self, db_path="jobs.db"):
        self.conn = sqlite3.connect(db_path)
        self._create_table()

    def _create_table(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS jobs (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT NOT NULL,
                company TEXT NOT NULL,
                location TEXT,
                url TEXT UNIQUE,
                source TEXT,
                salary_min REAL,
                salary_max REAL,
                posted_date TEXT,
                description TEXT,
                tags TEXT,
                scraped_at TEXT
            )
        """)
        self.conn.commit()

    def insert(self, job: JobListing):
        try:
            self.conn.execute(
                """INSERT OR IGNORE INTO jobs
                   (title, company, location, url, source,
                    salary_min, salary_max, posted_date,
                    description, tags, scraped_at)
                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                (job.title, job.company, job.location, job.url,
                 job.source, job.salary_min, job.salary_max,
                 job.posted_date, job.description,
                 json.dumps(job.tags), job.scraped_at),
            )
            self.conn.commit()
        except sqlite3.IntegrityError:
            pass  # Duplicate URL

    def search(self, keyword, location=None):
        query = "SELECT * FROM jobs WHERE title LIKE ?"
        params = [f"%{keyword}%"]
        if location:
            query += " AND location LIKE ?"
            params.append(f"%{location}%")
        query += " ORDER BY scraped_at DESC"
        cursor = self.conn.execute(query, params)
        return cursor.fetchall()

لاحظ قيد UNIQUE على عمود url؛ فمع INSERT OR IGNORE يمنع هذا القيد تسجيل الإعلان نفسه مرتين حين يظهر على صفحات مختلفة. أما دالة search فتتيح تصفية النتائج حسب الكلمة المفتاحية والموقع مباشرةً من قاعدة البيانات، دون تحميل كل السجلات إلى الذاكرة.

قاعدة مكشطة تتعامل مع CAPTCHA

هنا يدخل CaptchaAI. الفئة BaseScraper تجلب الصفحة، وتفحص محتواها بحثًا عن اختبار reCAPTCHA v2، وعند وجوده ترسل sitekey إلى نقطة النهاية in.php بالطريقة userrecaptcha، ثم تستطلع res.php دوريًا حتى يعود الرمز المحلول:

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


class BaseScraper:
    API_KEY = os.environ["CAPTCHAAI_API_KEY"]

    def __init__(self, source_name):
        self.source = source_name
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                          "AppleWebKit/537.36 Chrome/125.0.0.0 Safari/537.36",
        })

    def fetch(self, url):
        resp = self.session.get(url, timeout=20)

        if self._has_captcha(resp.text):
            token = self._solve_captcha(url, resp.text)
            resp = self.session.post(url, data={
                "g-recaptcha-response": token,
            }, timeout=30)

        return resp.text

    def _has_captcha(self, html):
        return "data-sitekey" in html or "g-recaptcha" in html

    def _solve_captcha(self, url, html):
        match = re.search(r'data-sitekey="([^"]+)"', html)
        if not match:
            raise ValueError("No sitekey found")

        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:
                return data["request"]
            if data["request"] != "CAPCHA_NOT_READY":
                raise RuntimeError(data["request"])
            time.sleep(5)

        raise TimeoutError("CAPTCHA solve timeout")

تدفّق الحل واضح: بعد إرسال المهمة تنتظر الدالة 15 ثانية، ثم تستفسر عن النتيجة كل 5 ثوانٍ إلى أن تصل الحالة status == 1؛ عندها يُعاد الرمز في الحقل g-recaptcha-response مع إعادة إرسال الطلب. يعتمد تسعير CaptchaAI على عدد الخيوط المتزامنة لا على عدد عمليات الحل، مع حل غير محدود لكل خيط؛ لذلك تكفي خطة BASIC — 15 دولارًا شهريًا مع 5 خيوط — لمُجمِّع صغير يعالج بضع لوحات بالتوازي. ويحل CaptchaAI إلى جانب reCAPTCHA v2 أنواعًا أخرى مثل reCAPTCHA v3 وCloudflare Turnstile وGeeTest v3، وهو ما يفيدك متى أضفت لوحات تعتمد حماية مختلفة.

استخراج الوظائف من كل لوحة

تبني الفئة GenericJobScraper على القاعدة السابقة، وتتعامل مع اختلاف تخطيط كل لوحة عبر قاموس محددات selectors. تمر على صفحات البحث من الأولى حتى max_pages، وتتوقف مبكرًا متى خلت الصفحة من البطاقات:

# scrapers.py
from bs4 import BeautifulSoup
from scraper_base import BaseScraper
from models import JobListing
import re


class GenericJobScraper(BaseScraper):
    """Scrape a job board search results page."""

    def __init__(self, source_name, base_url, selectors):
        super().__init__(source_name)
        self.base_url = base_url
        self.selectors = selectors

    def scrape_search(self, keyword, location="", max_pages=3):
        jobs = []

        for page in range(1, max_pages + 1):
            url = self.base_url.format(
                keyword=keyword.replace(" ", "+"),
                location=location.replace(" ", "+"),
                page=page,
            )
            html = self.fetch(url)
            page_jobs = self._parse_listings(html)

            if not page_jobs:
                break
            jobs.extend(page_jobs)

        return jobs

    def _parse_listings(self, html):
        soup = BeautifulSoup(html, "html.parser")
        cards = soup.select(self.selectors["card"])
        jobs = []

        for card in cards:
            title_el = card.select_one(self.selectors["title"])
            company_el = card.select_one(self.selectors["company"])
            location_el = card.select_one(self.selectors.get("location", ".location"))
            link_el = card.select_one(self.selectors.get("link", "a"))

            if not title_el or not company_el:
                continue

            salary = self._extract_salary(card.get_text())

            jobs.append(JobListing(
                title=title_el.get_text(strip=True),
                company=company_el.get_text(strip=True),
                location=location_el.get_text(strip=True) if location_el else "",
                url=link_el["href"] if link_el else "",
                source=self.source,
                salary_min=salary[0],
                salary_max=salary[1],
            ))

        return jobs

    def _extract_salary(self, text):
        match = re.search(
            r'\$?([\d,]+)\s*[-–to]+\s*\$?([\d,]+)', text
        )
        if match:
            return (
                float(match.group(1).replace(",", "")),
                float(match.group(2).replace(",", "")),
            )
        return (None, None)

تقرأ _parse_listings كل بطاقة فتستخرج العنوان والشركة والموقع والرابط، وتتجاهل البطاقات الناقصة. أما _extract_salary فتلتقط نطاق الراتب عبر تعبير نمطي يتعامل مع صيغ مثل $60,000–$90,000، ويعيده حقلين رقميين. بهذا الفصل، يصبح دعم لوحة جديدة مسألة إضافة محدداتها فحسب في معظم الحالات.

تشغيل المُجمِّع

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

# main.py
import time
from models import JobDatabase
from scrapers import GenericJobScraper

BOARDS = [
    {
        "name": "Board A",
        "base_url": "https://board-a.example.com/search?q={keyword}&l={location}&p={page}",
        "selectors": {
            "card": ".job-card",
            "title": ".job-title",
            "company": ".company-name",
            "location": ".job-location",
            "link": "a.job-link",
        },
    },
]


def main():
    db = JobDatabase()
    keywords = ["python developer", "data engineer"]

    for board in BOARDS:
        scraper = GenericJobScraper(board["name"], board["base_url"], board["selectors"])

        for keyword in keywords:
            print(f"Scraping {board['name']} for '{keyword}'...")
            jobs = scraper.scrape_search(keyword, location="Remote")

            for job in jobs:
                db.insert(job)
                print(f"  {job.title} at {job.company}")

            time.sleep(5)

    # Search example
    results = db.search("python", "Remote")
    print(f"\nFound {len(results)} matching jobs")


if __name__ == "__main__":
    main()

تفصل time.sleep(5) بين اللوحات لتخفيف الضغط على الخوادم. عدّل قائمة keywords، وأضف لوحاتك إلى BOARDS، ثم شغّل python main.py لتبدأ عملية التجميع. في النهاية يطبع المثال عدد الوظائف المطابقة داخل قاعدة البيانات.

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

نصائح لتوسيع المُجمِّع بأمان

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

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

ما أنواع اختبارات CAPTCHA التي يتعامل معها هذا المُجمِّع؟

يركّز المثال على reCAPTCHA v2 لأنه الأكثر شيوعًا في صفحات نتائج الوظائف. لكن CaptchaAI يحل أيضًا reCAPTCHA v3 وCloudflare Turnstile وGeeTest v3 وصور OCR، فتستطيع توسيع دالة الحل لتغطية لوحات ذات حماية مختلفة دون تغيير بنية المُجمِّع.

كم تكلفة تشغيل مُجمِّع وظائف على CaptchaAI؟

يعتمد التسعير على عدد الخيوط المتزامنة لا على عدد عمليات الحل. تبدأ خطة BASIC من 15 دولارًا شهريًا مع 5 خيوط وحل غير محدود، وهي كافية لمُجمِّع صغير؛ ومع نمو عدد اللوحات ترقّي إلى خطة ذات خيوط أكثر لرفع مستوى التوازي.

كيف أوحّد الرواتب والمواقع المختلفة بين اللوحات؟

عالج التوحيد في طبقة منفصلة بعد الكشط: حوّل نطاقات الرواتب إلى حقلين رقميين — salary_min وsalary_max — كما في _extract_salary، ووحّد أسماء المواقع (مثل توحيد «عن بُعد» و«Remote») قبل الإدراج، لتبقى قاعدة البيانات نظيفة وقابلة للفلترة.

هل يمكن جدولة المُجمِّع ليعمل تلقائيًا؟

نعم. غلّف main() بمهمة مجدولة عبر cron على Linux أو Task Scheduler على Windows، واجعلها تعمل في ساعات الذروة المنخفضة. وبتخزين scraped_at لكل إعلان تستطيع تتبّع الجديد منذ آخر تشغيل.

ماذا أفعل إذا كانت صفحة النتائج تعتمد على JavaScript بالكامل؟

إذا لم يظهر المحتوى في HTML الأولي، استبدل طبقة الجلب بمتصفّح مؤتمت مثل Selenium أو Playwright، وأبقِ منطق حل CAPTCHA عبر CaptchaAI كما هو. عندها تُنفَّذ الشيفرة داخل صفحة محمّلة بالكامل قبل قراءة البطاقات.

أدلة ذات صلة

• لوحة تحليل المنافسين باستخدام CaptchaAI
• خط أنابيب الاختبار الآلي مع CaptchaAI

جهّز بيانات وظائفك في مكان واحد — ابدأ الآن مع CaptchaAI.