How to Scrape Zillow Property Listings

"""
KnoxProxy scraper for Zillow property listings.
Extracts price, address, Zestimate, bed/bath count, and photo URLs from the
embedded JSON state on a property detail page.
"""
import json
import random
import re
import time

import requests

PROXY_HOST = "proxy.knoxproxy.com"
PROXY_PORT = 10000  # residential rotating
PROXY_USER = "your_username"
PROXY_PASS = "your_password"

HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
        "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
    ),
    "Accept-Language": "en-US,en;q=0.9",
}

NEXT_DATA_RE = re.compile(
    r'<script id="__NEXT_DATA__" type="application/json">(.*?)</script>', re.DOTALL
)


def proxy_dict(state: str = "ca") -> dict:
    user = f"{PROXY_USER}-country-us-state-{state}"
    proxy_url = f"http://{user}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}"
    return {"http": proxy_url, "https": proxy_url}


def fetch_zillow_property(zpid: str, url_slug: str, max_retries: int = 3) -> dict:
    url = f"https://www.zillow.com/homedetails/{url_slug}/{zpid}_zpid/"

    for attempt in range(1, max_retries + 1):
        try:
            resp = requests.get(url, headers=HEADERS, proxies=proxy_dict(), timeout=20)
            if resp.status_code == 200:
                return parse_zillow_property(resp.text, zpid)
            print(f"Attempt {attempt}: Zillow returned status {resp.status_code}")
        except requests.exceptions.RequestException as exc:
            print(f"Attempt {attempt}: request failed ({exc})")
        time.sleep(random.uniform(4, 7))

    raise RuntimeError(f"Failed to fetch Zillow property {zpid} after {max_retries} attempts")


def parse_zillow_property(html: str, zpid: str) -> dict:
    match = NEXT_DATA_RE.search(html)
    if not match:
        return {"zpid": zpid, "error": "property JSON not found"}

    try:
        data = json.loads(match.group(1))
    except json.JSONDecodeError:
        return {"zpid": zpid, "error": "could not decode property state"}

    # Zillow changes this embedded JSON structure periodically. Print
    # data.keys() and inspect the result before trusting any hardcoded
    # path for price, address, Zestimate, or photo URLs.
    return {"zpid": zpid, "raw_keys": list(data.keys())[:5]}


if __name__ == "__main__":
    print(fetch_zillow_property(zpid="12345678", url_slug="123-Main-St-Springfield-CA-90000"))

"""
Append parsed Zillow listings to a CSV file for market analysis over time (proxy scraper companion).
Run alongside zillow_scraper.py, feeding it each fetch_zillow_property()
result so price and inventory changes can be tracked across a metro area.
"""
import csv
import os
from datetime import datetime, timezone

FIELDS = ["zpid", "address", "price", "zestimate", "bedrooms", "bathrooms", "scraped_at"]


def append_listing(row: dict, path: str = "zillow_listings.csv") -> None:
    row = {**row, "scraped_at": datetime.now(timezone.utc).isoformat()}
    file_exists = os.path.exists(path)

    with open(path, "a", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDS, extrasaction="ignore")
        if not file_exists:
            writer.writeheader()
        writer.writerow(row)


if __name__ == "__main__":
    sample = {
        "zpid": "12345678",
        "address": "123 Main St, Springfield, CA 90000",
        "price": "$612,000",
        "zestimate": "$608,400",
        "bedrooms": 3,
        "bathrooms": 2,
    }
    append_listing(sample)

{
  "zpid": "12345678",
  "address": "123 Main St, Springfield, CA 90000",
  "price": "$612,000",
  "zestimate": "$608,400",
  "bedrooms": 3,
  "bathrooms": 2,
  "photos": [
    "https://photos.zillowstatic.com/fp/example-uncropped_scaled.jpg"
  ]
}