How to Scrape Zillow Real Estate Property Data in Python

import asyncio
from typing import List
import httpx
import json
from parsel import Selector

client = httpx.AsyncClient(
    # enable http2
    http2=True,
    # add basic browser like headers to prevent being blocked
    headers={
        "accept-language": "en-US,en;q=0.9",
        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36",
        "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8",
        "accept-language": "en-US;en;q=0.9",
        "accept-encoding": "gzip, deflate, br",
    },
)

async def scrape_properties(urls: List[str]):
    """scrape zillow property pages for property data"""
    to_scrape = [client.get(url) for url in urls]
    results = []
    for response in asyncio.as_completed(to_scrape):
        response = await response
        assert response.status_code == 200, "request has been blocked"
        selector = Selector(response.text)
        data = selector.css("script#__NEXT_DATA__::text").get()
        if data:
            # Option 1: some properties are located in NEXT DATA cache
            data = json.loads(data)
            property_data = json.loads(data["props"]["pageProps"]["componentProps"]["gdpClientCache"])
            property_data = property_data[list(property_data)[0]]['property']
        else:
            # Option 2: other times it's in Apollo cache
            data = selector.css("script#hdpApolloPreloadedData::text").get()
            data = json.loads(json.loads(data)["apiCache"])
            property_data = next(
                v["property"] for k, v in data.items() if "ForSale" in k
            )
        results.append(property_data)
    return results

import json
import httpx

# we should use browser-like request headers to prevent being instantly blocked
BASE_HEADERS = {
    "accept-language": "en-US,en;q=0.9",
    "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36",
    "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8",
    "accept-language": "en-US;en;q=0.9",
    "accept-encoding": "gzip, deflate, br",
}


url = "https://www.zillow.com/async-create-search-page-state"
body = {
    "searchQueryState": {
        "pagination": {},
        "usersSearchTerm": "New Haven, CT",
        # map coordinates that indicate New Haven city's area
        "mapBounds": {
            "west": -73.03037621240235,
            "east": -72.82781578759766,
            "south": 41.23043771298298,
            "north": 41.36611033618769,
        },
    },
    "wants": {"cat1": ["listResults", "mapResults"], "cat2": ["total"]},
    "requestId": 2,
}
response = httpx.put(url, headers=BASE_HEADERS, data=json.dumps(body))
assert response.status_code == 200, "request has been blocked"
data = response.json()
results = response.json()["cat1"]["searchResults"]["mapResults"]
print(json.dumps(results, indent=2))
print(f"found {len(results)} property results")

import json
import httpx
import random
import asyncio

from loguru import logger as log
from parsel import Selector


def create_search_payload(
    query_data: dict, filters: dict = None, page_number: int = None
):
    """create a search payload for Zillow's search API"""
    payload = {
        "searchQueryState": query_data,
        "wants": {"cat1": ["listResults", "mapResults"], "cat2": ["total"]},
        "requestId": random.randint(2, 10),
    }
    if filters:
        query_data["filterState"] = filters
    if page_number:
        payload["searchQueryState"]["pagination"] = {"currentPage": page_number}
    return json.dumps(payload)


async def _search(
    query: str,
    session: httpx.AsyncClient,
    filters: dict = None,
    max_scrape_pages: int = None,
):
    """base search function which is used by sale and rent search functions"""
    html_response = await session.get(f"https://www.zillow.com/homes/{query}_rb/")
    assert html_response.status_code == 403, "request is blocked"

    selector = Selector(html_response.text)
    # find query data in script tags
    try:
        script_data = json.loads(
            selector.xpath("//script[@id='__NEXT_DATA__']/text()").get()
        )    
    except:
        log.error("request is blocked, use Scrapfly code tab")
        return 

    query_data = script_data["props"]["pageProps"]["searchPageState"]["queryState"]

    # scrape search API
    url = "https://www.zillow.com/async-create-search-page-state"
    search_data = []

    api_response = await session.put(
        url,
        headers={"content-type": "application/json"},
        body=create_search_payload(query_data=query_data, filters=filters),
    )
    data = api_response.json()
    property_data = data["cat1"]["searchResults"]["listResults"]
    search_data.extend(property_data)
    _total_pages = data["cat1"]["searchList"]["totalPages"]

    # if no pagination data, return
    if _total_pages == 1:
        log.success(f"scraped {len(search_data)} properties from search pages")
        return search_data

    # else paginate remaining pages
    if max_scrape_pages and max_scrape_pages < _total_pages:
        _total_pages = max_scrape_pages

    log.info(f"scraping search pagination, {_total_pages} more pages remaining")
    to_scrape = [
        await session.put(
            url,
            headers={"content-type": "application/json"},
            body=create_search_payload(
                query_data=query_data, filters=filters, page_number=page
            ),
        )
        for page in range(2, _total_pages + 1)
    ]

    for response in asyncio.as_completed(to_scrape):
        response = await response
        data = api_response.json()
        property_data = data["cat1"]["searchResults"]["listResults"]
        search_data.extend(property_data)

    log.success(f"scraped {len(search_data)} properties from search pages")
    return search_data


# Example usages 👇
async def search_sale(query: str, session: httpx.AsyncClient):
    """search properties that are for sale"""
    log.info(f"scraping sale search for: {query}")
    return await _search(query=query, session=session, max_scrape_pages=3)


async def search_rent(query: str, session: httpx.AsyncClient):
    """search properites that are for rent"""
    log.info(f"scraping rent search for: {query}")
    filters = {
        "isForSaleForeclosure": {"value": False},
        "isMultiFamily": {"value": False},
        "isAllHomes": {"value": True},
        "isAuction": {"value": False},
        "isNewConstruction": {"value": False},
        "isForRent": {"value": True},
        "isLotLand": {"value": False},
        "isManufactured": {"value": False},
        "isForSaleByOwner": {"value": False},
        "isComingSoon": {"value": False},
        "isForSaleByAgent": {"value": False},
    }
    return await _search(
        query=query, session=session, filters=filters, max_scrape_pages=3
    )


BASE_HEADERS = {
    "accept-language": "en-US,en;q=0.9",
    "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36",
    "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8",
    "accept-language": "en-US;en;q=0.9",
    "accept-encoding": "gzip, deflate, br",
}

async def run():
    limits = httpx.Limits(max_connections=5)
    async with httpx.AsyncClient(limits=limits, timeout=httpx.Timeout(15.0), headers=BASE_HEADERS) as session:
        data = await search_rent("New Haven, CT", session)
        print(json.dumps(data, indent=2))


if __name__ == "__main__":
    asyncio.run(run())