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

لبناء قاعدة بيانات عقارية محدّثة يوميًا تحتاج إلى ثلاثة عناصر: مصدر قوائم موثوق، ومُحلِّل يستخرج السعر والمساحة والموقع، وطريقة ثابتة للتعامل مع اختبارات CAPTCHA التي تعترض الطلبات الآلية. يتكفّل CaptchaAI بالعنصر الأخير عبر حلّ reCAPTCHA وCloudflare Turnstile وتحدّيات Cloudflare وصور التحقق تلقائيًا، فيبقى خط الجمع مستقرًا من دون تدخّل يدوي عند كل صفحة.

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

البيانات التي تستحق الجمع من القوائم العقارية

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

أنواع حماية CAPTCHA على المنصات العقارية

تختلف طبقة الحماية بحسب نوع المنصة، ومعرفة النوع مسبقًا تحدّد آلية الحل المناسبة قبل أن يتعطّل الجمع. الجدول التالي يلخّص ما ستقابله غالبًا:

جميع هذه الأنواع مدعومة في CaptchaAI، لذا يتعامل الجامع الواحد معها من دون كتابة منطق منفصل لكل موقع.

بناء جامع بيانات العقارات

يجمع الصنف التالي ثلاث وظائف في مكان واحد: جلب الصفحة، والكشف التلقائي عن نوع CAPTCHA فيها، ثم إرسال الرمز المحلول وإعادة الطلب. عند العثور على reCAPTCHA يقرأ مفتاح الموقع ويحدّد ما إذا كانت النسخة v3 من وجود معامل render، ثم يرسل الطلب المناسب؛ وعند وجود Turnstile يتبع المسار نفسه بمعامل مختلف. أما بقية الشيفرة فتفصل السعر والعنوان والمساحة عبر تعبيرات نمطية وتصدّرها إلى ملف CSV جاهز للتحليل.

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

API_KEY = os.environ["CAPTCHAAI_API_KEY"]


def solve_captcha(params):
    params["key"] = API_KEY
    resp = requests.get("https://ocr.captchaai.com/in.php", params=params)
    if not resp.text.startswith("OK|"):
        raise Exception(f"Submit: {resp.text}")

    task_id = resp.text.split("|")[1]
    for _ in range(60):
        time.sleep(5)
        result = requests.get("https://ocr.captchaai.com/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: {result.text}")
    raise TimeoutError()


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

    def fetch(self, url):
        """Fetch page with automatic CAPTCHA handling."""
        resp = self.session.get(url)

        # reCAPTCHA
        match = re.search(
            r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', resp.text
        )
        if match:
            # Detect v3
            is_v3 = "recaptcha/api.js?render=" in resp.text
            params = {
                "method": "userrecaptcha",
                "googlekey": match.group(1),
                "pageurl": url,
            }
            if is_v3:
                params["version"] = "v3"
                params["action"] = "search"

            token = solve_captcha(params)
            resp = self.session.post(url, data={
                "g-recaptcha-response": token,
            })

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

        return resp.text

    def collect_listings(self, urls):
        """Collect property listings from multiple pages."""
        listings = []
        for url in urls:
            try:
                html = self.fetch(url)
                page_listings = self._parse_listings(html)
                listings.extend(page_listings)
                print(f"  {len(page_listings)} listings from {url}")
                time.sleep(3)
            except Exception as e:
                print(f"  Error: {url} - {e}")
        return listings

    def _parse_listings(self, html):
        """Extract property data from HTML."""
        listings = []

        # Price extraction
        prices = re.findall(r'\$\s*([\d,]+)', html)
        # Address extraction
        addresses = re.findall(
            r'class="address"[^>]*>(.*?)</(?:div|span|p)', html
        )
        # Bed/Bath extraction
        beds = re.findall(r'(\d+)\s*(?:bed|br|bedroom)', html, re.I)
        baths = re.findall(r'(\d+)\s*(?:bath|ba|bathroom)', html, re.I)
        # Sqft extraction
        sqft = re.findall(r'([\d,]+)\s*(?:sq\s*ft|sqft)', html, re.I)

        # Combine available data
        count = max(len(prices), len(addresses), 1)
        for i in range(min(count, 50)):  # Cap at 50 per page
            listing = {
                "price": prices[i] if i < len(prices) else None,
                "address": (
                    addresses[i].strip() if i < len(addresses) else None
                ),
                "beds": beds[i] if i < len(beds) else None,
                "baths": baths[i] if i < len(baths) else None,
                "sqft": sqft[i] if i < len(sqft) else None,
                "collected_at": datetime.utcnow().isoformat(),
            }
            if listing["price"] or listing["address"]:
                listings.append(listing)

        return listings

    def export_csv(self, listings, filename):
        if not listings:
            print("No listings to export")
            return

        keys = ["price", "address", "beds", "baths", "sqft", "collected_at"]
        with open(filename, "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=keys)
            writer.writeheader()
            writer.writerows(listings)
        print(f"Exported {len(listings)} listings to {filename}")


# Usage
collector = PropertyCollector()

search_urls = [
    "https://example-realty.com/search?city=austin&type=sale&page=1",
    "https://example-realty.com/search?city=austin&type=sale&page=2",
    "https://example-realty.com/search?city=austin&type=sale&page=3",
]

listings = collector.collect_listings(search_urls)
collector.export_csv(listings, "austin_listings.csv")

من الجمع اليومي إلى تقرير السوق العقاري

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

Daily Collection
    → Property listings (500-1000 per market)
    → Price changes (delta from previous day)
    → New listings vs delisted

Weekly Analysis
    → Median price trends
    → Inventory levels
    → Days-on-market averages
    → Price-per-sqft by neighborhood

Monthly Report
    → Market heat map
    → Competitive pricing analysis
    → Investment opportunity scoring

سيناريو من السوق الإقليمي

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

التكلفة والتوسّع

يعتمد CaptchaAI على تسعير قائم على الـ Threads بدلًا من الدفع لكل عملية حل؛ تشترك في عدد من المسارات المتزامنة، ويحلّ كل مسار عددًا غير محدود من الاختبارات خلال الشهر. لخط جمع صغير يفحص بضع مئات من القوائم يوميًا تكفي خطة BASIC — 15 دولارًا شهريًا و5 مسارات متزامنة. وعند توسيع الجمع إلى عدة أسواق بالتوازي، أي آلاف القوائم يوميًا، توفّر خطة ADVANCE — 90 دولارًا شهريًا و50 مسارًا — أو PREMIUM — 170 دولارًا شهريًا و100 مسار — التزامن الكافي من دون رسوم إضافية لكل اختبار. الميزة العملية أنّ تكلفة الجمع تبقى ثابتة ومتوقّعة بصرف النظر عن عدد الاختبارات التي تظهر خلال الشهر، فتخطّط ميزانيتك بحسب حجم التزامن لا بحسب عدد القوائم.

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

ما أنواع الحماية التي يتعامل معها CaptchaAI على المواقع العقارية؟

يحلّ CaptchaAI اختبارات reCAPTCHA v2 وv3 وv2 Enterprise وCloudflare Turnstile وتحدّيات Cloudflare وصور التحقق النصية، وهي الأنواع الأكثر شيوعًا على بوابات العقارات. لا حاجة إلى منطق منفصل لكل نوع؛ يرسل الجامع بيانات الطلب وتعود الاستجابة جاهزة للاستخدام.

كيف أُبقي خط الجمع مستقرًا عند تغيّر بنية الصفحات؟

اقرأ الـ sitekey ديناميكيًا من HTML بدل تثبيته في الشيفرة، وأضِف منطق إعادة المحاولة مع مهلة انتهاء واضحة، وسجّل خطأ كل صفحة على حدة حتى لا يوقف فشلٌ واحد بقيّة الدُفعة. هذه العادات الثلاث تحمي الخط من التغييرات الصغيرة في تصميم المواقع.

كيف أتجنّب الحظر وتحديد معدل الطلبات أثناء الجمع؟

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

هل جمع بيانات القوائم العقارية مسموح؟

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

أدلة ذات صلة

• مراقبة أسعار المتاجر الإلكترونية عند ظهور اختبارات CAPTCHA
• ابدأ مع CaptchaAI: أول اختبار محلول في خمس دقائق