Cloudflareチャレンジの解決方法：不動産価格モニタリングのための

monitoring_job = {
    "canonical_property_id": "prop_83f12",
    "market": "US-CA-San-Francisco",
    "source_priority": [
        "official_transaction_dataset",
        "licensed_listing_feed",
        "partner_api",
        "authorized_public_page",
    ],
    "price_types": ["ASKING_SALE", "RECORDED_SALE"],
    "alert_rules": {
        "asking_change_percent": 3.0,
        "status_changes": ["ACTIVE_TO_PENDING", "PENDING_TO_SOLD"],
    },
}

from dataclasses import dataclass

@dataclass(frozen=True)
class PropertyIdentity:
    canonical_property_id: str
    country: str
    administrative_area: str
    locality: str
    postal_code: str
    street_number: str
    street_name: str
    unit: str | None
    parcel_id: str | None
    latitude: float | None
    longitude: float | None

from dataclasses import dataclass, field
from datetime import datetime, timezone

@dataclass
class PriceObservation:
    canonical_property_id: str
    source: str
    source_record_id: str | None
    source_url: str | None
    price_type: str
    amount: float
    currency: str
    area_value: float | None
    area_unit: str | None
    status: str | None
    market: str
    evidence_hash: str
    parser_version: str
    observed_at: str = field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat()
    )

PRICE_TYPES = {
    "ASKING_SALE",
    "RECORDED_SALE",
    "ASKING_RENT_MONTHLY",
    "VALUATION_ESTIMATE",
    "PRICE_PER_AREA",
    "MARKET_INDEX",
}

async def classify_property_page(page, expected_selector: str) -> str:
    if await page.locator(expected_selector).count():
        return "PROPERTY_PAGE"

    title = (await page.title()).strip().lower()
    html = (await page.content()).lower()

    if "just a moment" in title:
        return "CLOUDFLARE_CHALLENGE"
    if "challenge-platform" in html or "cf-chl-" in html:
        return "CLOUDFLARE_CHALLENGE"
    if "listing no longer available" in html:
        return "LISTING_UNAVAILABLE"
    return "UNKNOWN_PAGE"

import os
from urllib.parse import urlparse

SOURCE_POLICY = {
    "approved-listings.example": {
        "purpose": "asking-price-validation",
        "proxy_profile": "property_us_west",
        "max_checks_per_hour": 2,
    },
}

PROXY_VAULT = {
    "property_us_west": os.environ["PROPERTY_PROXY_US_WEST"],
}


def approved_source(url: str) -> dict:
    host = urlparse(url).hostname
    policy = SOURCE_POLICY.get(host)
    if not policy:
        raise PermissionError("Property source is not approved")
    return policy

import capsolver

capsolver.api_key = os.environ["CAPSOLVER_API_KEY"]


def solve_property_page_challenge(
    url: str,
    chrome_user_agent: str,
    fresh_html: str = "",
) -> dict:
    policy = approved_source(url)
    proxy = PROXY_VAULT[policy["proxy_profile"]]

    task = {
        "type": "AntiCloudflareTask",
        "websiteURL": url,
        "proxy": proxy,
        "userAgent": chrome_user_agent,
    }
    if fresh_html:
        task["html"] = fresh_html

    solution = capsolver.solve(task)
    cookies = solution.get("cookies") or {}
    clearance = cookies.get("cf_clearance") or solution.get("token")
    if not clearance:
        raise RuntimeError("Cloudflare Challenge solution is incomplete")

    return {
        "cookies": cookies,
        "user_agent": solution.get("userAgent") or chrome_user_agent,
        "proxy_profile": policy["proxy_profile"],
    }

async def recover_property_document(
    source_adapter,
    url: str,
    challenge_html: str,
    user_agent: str,
) -> str:
    solution = solve_property_page_challenge(
        url=url,
        chrome_user_agent=user_agent,
        fresh_html=challenge_html,
    )

    response = await source_adapter.fetch_with_session(
        url=url,
        proxy_profile=solution["proxy_profile"],
        user_agent=solution["user_agent"],
        cookies=solution["cookies"],
    )

    if response.status_code != 200:
        raise RuntimeError("Recovered request did not return HTTP 200")
    if "challenge-platform" in response.text.lower():
        raise RuntimeError("Challenge remained after one recovery attempt")
    return response.text

from decimal import Decimal, InvalidOperation

class SourceAdapter:
    source_name = "approved-listings"
    parser_version = "2026-08-28.1"

    def parse_asking_price(self, document) -> dict:
        raw = document.select_one('[data-testid="asking-price"]')
        if raw is None:
            raise ValueError("Asking-price field not found")

        currency = raw.get("data-currency")
        amount_text = raw.get("data-amount")
        try:
            amount = Decimal(amount_text)
        except (InvalidOperation, TypeError):
            raise ValueError("Invalid asking-price amount")

        if amount <= 0 or not currency:
            raise ValueError("Incomplete asking-price observation")

        return {
            "price_type": "ASKING_SALE",
            "amount": float(amount),
            "currency": currency,
        }

from decimal import Decimal


def comparable(left: PriceObservation, right: PriceObservation) -> bool:
    return all([
        left.canonical_property_id == right.canonical_property_id,
        left.price_type == right.price_type,
        left.currency == right.currency,
        left.market == right.market,
    ])


def percentage_change(previous: float, current: float) -> float:
    if previous <= 0:
        raise ValueError("Previous price must be positive")
    return float(
        (Decimal(str(current)) - Decimal(str(previous)))
        / Decimal(str(previous))
        * 100
    )

def confirmed_price_change(
    observations: list[PriceObservation],
    threshold_percent: float,
) -> dict | None:
    if len(observations) < 3:
        return None

    baseline, first_new, second_new = observations[-3:]
    if not comparable(baseline, first_new):
        return None
    if not comparable(first_new, second_new):
        return None
    if first_new.amount != second_new.amount:
        return None

    change = percentage_change(baseline.amount, second_new.amount)
    if abs(change) < threshold_percent:
        return None

    return {
        "event": "CONFIRMED_PRICE_CHANGE",
        "previous": baseline.amount,
        "current": second_new.amount,
        "currency": second_new.currency,
        "change_percent": round(change, 2),
    }