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

Zillow, các cổng MLS và trang hồ sơ thuế tài sản hiếm khi chỉ dùng một lớp CAPTCHA — có thể gặp cả reCAPTCHA v3 chạy nền, Cloudflare Challenge và CAPTCHA hình ảnh trong cùng một phiên thu thập. CaptchaAI xử lý các lớp này qua API, giúp việc lấy danh sách, giá và dữ liệu thuế chạy đúng lịch thay vì bị chặn giữa chừng.

Đội ngũ outsourcing tại TP.HCM và Hà Nội thường xây pipeline này cho khách hàng bất động sản Mỹ, nơi CAPTCHA nhiều lớp là chuyện thường ngày.

Vì sao một trang bất động sản hiếm khi chỉ có một lớp CAPTCHA

Các nền tảng MLS, cổng niêm yết và hồ sơ thuế thường chồng nhiều lớp phòng thủ vì mỗi lớp chặn một kiểu bot khác nhau: Cloudflare Challenge lọc traffic ở tầng mạng trước khi trang tải, reCAPTCHA v3 chấm điểm hành vi trong lúc duyệt, còn CAPTCHA hình ảnh chặn ở bước cuối khi submit form tra cứu. Một script chỉ xử lý được một lớp sẽ dừng ngay ở lớp thứ hai. Vì vậy pipeline thu thập dữ liệu bất động sản cần:

• Phát hiện đúng loại CAPTCHA đang gặp, không giả định trước loại nào
• Gửi task tới CaptchaAI với method tương ứng — userrecaptcha, turnstile…
• Polling res.php tới khi có token, rồi submit lại đúng field mà server mong đợi
• Ghi log từng lần giải để đối chiếu chi phí theo thread khi mở rộng sang thị trường mới

Các loại CAPTCHA thường gặp trên trang bất động sản

Script Python thu thập dữ liệu bất động sản kèm xử lý CAPTCHA

Class PropertyCollector dưới đây minh họa luồng đầy đủ: phát hiện sitekey, gửi task tới in.php, polling res.php tới khi có token, rồi submit lại request. Với reCAPTCHA, script tự phân biệt v2/v3 qua tham số render; với Turnstile, nó tìm data-sitekey trong DOM. Đổi API_KEY (biến môi trường CAPTCHAAI_API_KEY) và search_urls cho thị trường bạn cần theo dõi.

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

Kết quả xuất ra austin_listings.csv; đổi search_urls cho đúng thị trường. Script giới hạn 50 tin/trang, nghỉ 3 giây giữa các request để tránh quá tải.

Những trường dữ liệu bất động sản đáng thu thập

• Giá niêm yết (trang chi tiết bất động sản) — dùng để định giá thị trường theo khu vực.
• Địa chỉ (trang chi tiết bất động sản) — nền cho phân tích theo khu vực, phường xã.
• Số phòng ngủ, phòng tắm, diện tích (chi tiết tài sản) — dùng so sánh các sản phẩm tương đương.
• Số ngày trên thị trường (metadata danh sách) — đo tốc độ thanh khoản của thị trường.
• Lịch sử giá (nhật ký thay đổi giá) — phân tích xu hướng theo thời gian.
• Thuế tài sản (hồ sơ thuế) — đầu vào cho phân tích đầu tư.
• Phí HOA (chi tiết danh sách) — đầu vào cho phân tích chi phí sở hữu.

Biến dữ liệu thô thành báo cáo thị trường

Tổng hợp dữ liệu thô theo ba tầng thời gian — ngày, tuần, tháng — để nó phục vụ được quyết định, không chỉ nằm trong 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

Với khách hàng ở múi giờ Mỹ, đội outsourcing thường chạy job thu thập vào ban đêm giờ Việt Nam — trùng ban ngày ở Mỹ khi giá niêm yết được cập nhật — rồi gửi báo cáo tuần/tháng dạng file tổng hợp thay vì log thô, giúp giảm thời gian review phía khách hàng.

Lưu ý tuân thủ: nếu dữ liệu thu thập chứa thông tin liên hệ của chủ nhà hoặc môi giới, Nghị định 13/2023/NĐ-CP về bảo vệ dữ liệu cá nhân yêu cầu hạn chế lưu trữ những gì không cần thiết. An toàn nhất là chỉ giữ lại dữ liệu thị trường đã tổng hợp và xóa các trường liên hệ cá nhân sau khi xử lý.

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

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

Dữ liệu niêm yết công khai (giá, địa chỉ, diện tích) thường thu thập được. Tránh lấy thông tin cá nhân người bán/môi giới; kiểm tra điều khoản dịch vụ trước khi chạy ở quy mô lớn.

Giải CAPTCHA cho pipeline thu thập bất động sản tốn bao nhiêu?

Gói BASIC ($15/tháng, 5 thread) đủ cho vài nghìn lượt giải mỗi tháng. Mở rộng nhiều thành phố thì cân nhắc ADVANCE ($90/tháng, 50 thread) — CaptchaAI tính giá theo thread, không theo từng lần giải.

Nên polling hay dùng callback khi thu thập dữ liệu bất động sản theo lịch?

Với job chạy định kỳ (cron hàng đêm), polling như trong script ở trên là đủ đơn giản và dễ debug. Nếu pipeline thu thập liên tục suốt ngày với nhiều nghìn task, webhook/callback giảm số request tới res.php và tải trên server thu thập.

reCAPTCHA v3 điểm thấp trên các cổng bất động sản thì xử lý thế nào?

Gửi tham số action khớp với hành động thực tế trên trang (ví dụ search) và giữ pageurl đúng URL đang truy cập — hai tham số này ảnh hưởng trực tiếp tới điểm số CaptchaAI trả về cho reCAPTCHA v3 chạy nền.

Hướng dẫn liên quan

• Giám sát biến động giá thương mại điện tử
• Thiết lập proxy đúng cách khi scrape
• Vượt CAPTCHA khi scrape các trang được bảo vệ