使用验证码处理进行房地产数据抓取

挂牌价格、户型、历史成交记录本身是公开信息，脚本却常在验证码这一步卡住超时重试。MLS 聚合平台叠加 Cloudflare 验证流程，Zillow、Realtor.com 这类门户是 reCAPTCHA v3 隐形打分，拍卖网站则挂着 Turnstile 组件——防护方式各不相同，脚本却得统一处理。CaptchaAI 把这几种识别封装成同一套 API：拿到 sitekey、换回 token、原样提交表单继续抓取。

房产平台验证码防护类型速查

房产站点用的验证码跟平台类型强相关，摸清规律能省不少调试时间：

这六类 CaptchaAI 均为 GA 能力，不用为不同平台单独接服务。

用 CaptchaAI 自动处理房源页面的验证码

下面这段 Python 示例做了三件事：识别页面是 reCAPTCHA 还是 Turnstile、调用 CaptchaAI 换回 token、把 token 提交回原页面继续抓取。solve_captcha() 每 5 秒轮询一次 res.php；fetch() 用正则找 data-sitekey，再看页面是否加载了 recaptcha/api.js?render= 来判断走 v2 还是 v3：

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

search_urls 里直接把 page=1、page=2、page=3 拼进了 URL——大多数房产网站的分页就是 ?page=N 或 &offset=N 这类参数，翻页时把城市、价格区间等筛选条件原样带上即可。生产环境建议用线程池并发跑多个页面，具体能开多少并发，取决于你 CaptchaAI 套餐里的线程数（下文有对照）。

需要抓取哪些关键字段

从每日采集到月度市场报告

把上面这些字段按周期跑起来，能拼出一条完整的分析链路：

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

每日看绝对量，价格变动和新增/下架数量最先反映市场情绪；周报、月报再叠加中位价趋势、库存水位和单位面积价格。

按采集量选择 CaptchaAI 套餐

CaptchaAI 按并发线程数计费，同一线程内解决次数不限。单城市、日采集量 3,000–5,000 条挂牌，BASIC（$15/月，5 线程）够用；同时监控多个城市，ADVANCE（$90/月，50 线程）更合适；批量跑市场报告的团队可以用 PREMIUM（$170/月，100 线程）压缩采集耗时。

只采集公开的挂牌信息，避免收集卖家或经纪人的个人联系方式；国内团队还需留意《网络安全法》《数据安全法》和《个人信息保护法》（PIPL）对个人信息处理的要求，遵守目标网站的服务条款是最稳妥的边界。

常见问题

抓取房产网站数据会违反网络安全法或 PIPL 吗？

公开挂牌数据通常可以采集，但涉及能关联到具体自然人的信息就要按《个人信息保护法》处理——不采集非必要的个人信息，遵守目标网站的服务条款。

CaptchaAI 能处理房产网站常见的验证码类型吗？

能。reCAPTCHA v2/v3、Cloudflare Turnstile、Cloudflare 验证流程和图片验证码都是 GA（正式支持）能力，本文表格里的六类平台防护基本都能覆盖。

采集速度上不去，要不要加线程？

大概率是线程不够用。BASIC 只有 5 线程，同时监控多个城市容易排队；升级 ADVANCE 或更高档位能直接提升并发上限。

MLS 聚合平台的 Cloudflare 验证流程为什么比其他类型更难处理？

它通常叠加了 IP 信誉检测，光靠验证码识别不够，还要有稳定的出口 IP 配合，cloudflare_challenge 方法只负责解决验证码本身。

相关指南

• 电商价格监控实战
• 代理设置最佳实践
• 如何抓取有验证码防护的网站