Quét dữ liệu bất động sản bằng cách xử lý CAPTCHA

Nền tảng bất động sản được bảo vệ nghiêm ngặt trước việc thu thập dữ liệu tự động. CaptchaAI giúp bạn duy trì quyền truy cập đáng tin cậy vào danh sách bất động sản, dữ liệu giá cả và phân tích thị trường.

Bảo vệ CAPTCHA trên các trang web bất động sản

Loại nền tảng	Bảo vệ	Loại CAPTCHA
Công cụ tổng hợp MLS	Cloudflare Challenge	Thử thách đầy đủ + proxy
Cổng loại Zillow	reCAPTCHA v3	Vô hình, hành vi
Danh mục nhà môi giới bất động sản	reCAPTCHA v2	Hộp kiểm hoặc vô hình
Hồ sơ thuế tài sản	CAPTCHA hình ảnh	Nhận dạng văn bản
Trang web đấu giá	Cloudflare Turnstile	Thử thách tiện ích
Danh sách thương mại	reCAPTCHA v2 Enterprise	Xác minh nâng cao

Người thu thập dữ liệu thuộc tính

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")

Điểm dữ liệu cần thu thập

trường	Nguồn	Trường hợp sử dụng
Giá niêm yết	Trang thuộc tính	Định giá thị trường
Địa chỉ	Trang thuộc tính	Phân tích địa lý
Giường/Baths/Sqft	Chi tiết tài sản	Phân tích so sánh
Ngày trên thị trường	Danh sách siêu dữ liệu	Vận tốc thị trường
Lịch sử giá	Nhật ký thay đổi giá	Phân tích xu hướng
Thuế tài sản	Hồ sơ thuế	Phân tích đầu tư
Phí HOA	Chi tiết danh sách	Phân tích chi phí

Quy trình phân tích thị trường

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

Câu hỏi thường gặp

Việc thu thập dữ liệu bất động sản có hợp pháp không?

Dữ liệu niêm yết công khai thường có thể thu thập được. Tránh thu thập thông tin cá nhân về người bán hoặc đại lý. Luôn tuân thủ các điều khoản dịch vụ của trang web.

Làm cách nào để xử lý phân trang?

Tăng tham số trang trong URL của bạn. Hầu hết các trang web bất động sản đều sử dụng mẫu ?page=N hoặc &offset=N.

Loại CAPTCHA nào khó nhất trên các trang web bất động sản?

Cloudflare Challenge trên các trang tổng hợp MLS là phức tạp nhất — nó yêu cầu các tham số proxy. reCAPTCHA v3 trên các cổng chính là phổ biến nhưng được CaptchaAI giải quyết một cách đáng tin cậy.