scraper.py

#!/usr/bin/env python3
"""
Zillow Scraper - Interactive CLI Tool
Powered by RealtyAPI.io | Zillow Scraper API

Scrape Zillow property data, search listings, find agents,
analyze housing markets, and run skip traces - all from the command line.

Supports single test mode (preview) and bulk mode (from input.json).

Get your API key at: https://www.realtyapi.io
"""

import csv
import json
import os
import sys
import time

from config import ensure_api_key
from zillow_api import ZillowAPI

SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
OUTPUT_DIR = os.path.join(SCRIPT_DIR, "output")
INPUT_FILE = os.path.join(SCRIPT_DIR, "input.json")


# ─────────────────────────────────────────────
# Helpers
# ─────────────────────────────────────────────

def preview(data, max_chars=1000):
    """Show first max_chars of the JSON response as a preview."""
    text = json.dumps(data, indent=2)
    if len(text) <= max_chars:
        print(text)
    else:
        print(text[:max_chars])
        print(f"\n... ({len(text) - max_chars} more characters)")


def prompt_choice(options, prompt_text="Select an option"):
    """Display numbered options and get user's choice."""
    print()
    for i, option in enumerate(options, 1):
        print(f"  {i}. {option}")
    print()
    while True:
        try:
            choice = int(input(f"{prompt_text} (1-{len(options)}): "))
            if 1 <= choice <= len(options):
                return choice
            print(f"Please enter a number between 1 and {len(options)}")
        except ValueError:
            print("Please enter a valid number")


def prompt_output_format():
    """Ask user what output format they want."""
    options = ["JSON (.json)", "CSV (.csv)", "Both JSON and CSV"]
    choice = prompt_choice(options, "Save output as")
    return choice


def save_json(data, filepath):
    """Save data as JSON."""
    os.makedirs(os.path.dirname(filepath), exist_ok=True)
    with open(filepath, "w") as f:
        json.dump(data, f, indent=2)


def flatten_dict(d, parent_key="", sep="_"):
    """Flatten a nested dict for CSV export."""
    items = []
    for k, v in d.items():
        new_key = f"{parent_key}{sep}{k}" if parent_key else k
        if isinstance(v, dict):
            items.extend(flatten_dict(v, new_key, sep).items())
        elif isinstance(v, list):
            items.append((new_key, json.dumps(v)))
        else:
            items.append((new_key, v))
    return dict(items)


def save_csv(records, filepath):
    """Save a list of dicts as CSV."""
    os.makedirs(os.path.dirname(filepath), exist_ok=True)
    if not records:
        print("No data to save.")
        return

    # Flatten each record
    flat_records = []
    for r in records:
        if isinstance(r, dict):
            flat_records.append(flatten_dict(r))
        else:
            flat_records.append({"data": json.dumps(r)})

    # Collect all keys across records
    all_keys = []
    seen = set()
    for r in flat_records:
        for k in r.keys():
            if k not in seen:
                all_keys.append(k)
                seen.add(k)

    with open(filepath, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=all_keys, extrasaction="ignore")
        writer.writeheader()
        writer.writerows(flat_records)


def save_results(results, base_name, fmt_choice):
    """Save results based on user's format choice."""
    os.makedirs(OUTPUT_DIR, exist_ok=True)
    json_path = os.path.join(OUTPUT_DIR, f"{base_name}.json")
    csv_path = os.path.join(OUTPUT_DIR, f"{base_name}.csv")

    if fmt_choice in (1, 3):
        save_json(results, json_path)
        print(f"  Saved: {json_path}")

    if fmt_choice in (2, 3):
        csv_records = results if isinstance(results, list) else [results]
        save_csv(csv_records, csv_path)
        print(f"  Saved: {csv_path}")


def load_input_file():
    """Load the input.json file."""
    if not os.path.exists(INPUT_FILE):
        print(f"\n  \033[93m⚠  input.json not found at:\033[0m {INPUT_FILE}")
        print(f"  \033[93m   Create it using input_sample.json as a template.\033[0m")
        print(f"  \033[93m   Copy input_sample.json to input.json and fill in your data.\033[0m\n")
        return None
    with open(INPUT_FILE, "r") as f:
        return json.load(f)


def progress_bar(current, total, label=""):
    """Display a simple progress bar."""
    width = 30
    filled = int(width * current / total)
    bar = "=" * filled + "-" * (width - filled)
    pct = int(100 * current / total)
    print(f"\r  [{bar}] {pct}% ({current}/{total}) {label}", end="", flush=True)
    if current == total:
        print()


def run_bulk(items, label, fetch_fn, base_name):
    """
    Run a bulk operation from a list of input items.
    fetch_fn(item) -> dict (API response)
    Shows progress, does not display responses, saves all at end.
    """
    total = len(items)
    if total == 0:
        print(f"  No {label} items found in input.json")
        return

    print(f"\n  Found {total} {label} item(s) in input.json")
    fmt_choice = prompt_output_format()

    results = []
    errors = []

    print()
    for i, item in enumerate(items, 1):
        item_label = str(item) if not isinstance(item, dict) else json.dumps(item)[:60]
        progress_bar(i, total, item_label)
        try:
            data = fetch_fn(item)
            results.append({"input": item, "result": data})
        except Exception as e:
            errors.append({"input": item, "error": str(e)})
            results.append({"input": item, "result": None, "error": str(e)})
        # Small delay between requests to be polite
        if i < total:
            time.sleep(0.3)

    print(f"\n  Done: {total - len(errors)} succeeded, {len(errors)} failed")
    save_results(results, base_name, fmt_choice)


def ask_mode():
    """Ask whether user wants single test or bulk mode."""
    options = [
        "Test one (enter input manually, see preview)",
        "Bulk run (read from input.json, save all results)",
    ]
    return prompt_choice(options, "Select mode")


# ─────────────────────────────────────────────
# Property Lookup
# ─────────────────────────────────────────────

def handle_property_lookup(api):
    print("\n--- Property Lookup ---")
    options = [
        "Lookup by Address",
        "Lookup by ZPID (Zillow Property ID)",
        "Lookup by Zillow URL",
    ]
    lookup_choice = prompt_choice(options)
    mode = ask_mode()

    if mode == 1:
        # Single test
        if lookup_choice == 1:
            val = input("\nEnter property address: ").strip()
            if not val: return print("Address required.")
            print("\nFetching...")
            data = api.get_property_by_address(val)
        elif lookup_choice == 2:
            val = input("\nEnter ZPID: ").strip()
            if not val: return print("ZPID required.")
            print("\nFetching...")
            data = api.get_property_by_zpid(val)
        else:
            val = input("\nEnter Zillow URL: ").strip()
            if not val: return print("URL required.")
            print("\nFetching...")
            data = api.get_property_by_url(val)

        print("\n--- Preview (first 1000 chars) ---")
        preview(data)

        save = input("\nSave full output? (y/n): ").strip().lower()
        if save == "y":
            fmt = prompt_output_format()
            save_results(data, "property_data", fmt)

    else:
        # Bulk mode
        input_data = load_input_file()
        if not input_data:
            return

        key_map = {1: "property_addresses", 2: "property_zpids", 3: "property_urls"}
        key = key_map[lookup_choice]
        items = input_data.get(key, [])

        fn_map = {
            1: lambda addr: api.get_property_by_address(addr),
            2: lambda zpid: api.get_property_by_zpid(zpid),
            3: lambda url: api.get_property_by_url(url),
        }

        label_map = {1: "address", 2: "ZPID", 3: "URL"}
        run_bulk(items, label_map[lookup_choice], fn_map[lookup_choice],
                 f"property_by_{label_map[lookup_choice].lower()}")


# ─────────────────────────────────────────────
# Search
# ─────────────────────────────────────────────

def handle_search(api):
    print("\n--- Search Zillow ---")
    options = [
        "Search by Location (address/city/ZIP)",
        "Search by Coordinates (lat/lng/radius)",
        "Search by Zillow URL",
        "Search by MLS Number",
        "Autocomplete Suggestions",
    ]
    search_choice = prompt_choice(options)
    mode = ask_mode()

    if mode == 1:
        # Single test
        data = _search_single(api, search_choice)
        if data is None:
            return
        print("\n--- Preview (first 1000 chars) ---")
        preview(data)
        save = input("\nSave full output? (y/n): ").strip().lower()
        if save == "y":
            fmt = prompt_output_format()
            save_results(data, "search_results", fmt)

    else:
        # Bulk mode
        input_data = load_input_file()
        if not input_data:
            return

        key_map = {
            1: "search_locations",
            2: "search_coordinates",
            3: "search_urls",
            4: "search_mls_numbers",
            5: "autocomplete_queries",
        }
        key = key_map[search_choice]
        items = input_data.get(key, [])

        fn_map = {
            1: lambda item: api.search_by_address(
                location=item.get("location", item) if isinstance(item, dict) else item,
                listing_status=item.get("listing_status", "For_Sale") if isinstance(item, dict) else "For_Sale",
                page=item.get("page", 1) if isinstance(item, dict) else 1,
                sort_order=item.get("sort_order") if isinstance(item, dict) else None,
                bed_min=item.get("bed_min") if isinstance(item, dict) else None,
                bed_max=item.get("bed_max") if isinstance(item, dict) else None,
                bathrooms=item.get("bathrooms") if isinstance(item, dict) else None,
                list_price_range=item.get("list_price_range") if isinstance(item, dict) else None,
                home_type=item.get("home_type") if isinstance(item, dict) else None,
                keywords=item.get("keywords") if isinstance(item, dict) else None,
            ),
            2: lambda item: api.search_by_coordinates(
                latitude=item["latitude"],
                longitude=item["longitude"],
                radius=item.get("radius", "1"),
                listing_status=item.get("listing_status", "For_Sale"),
            ),
            3: lambda url: api.search_by_url(url),
            4: lambda mls: api.search_by_mls(mls),
            5: lambda q: api.autocomplete(q),
        }

        label_map = {1: "location", 2: "coordinate", 3: "URL", 4: "MLS", 5: "autocomplete"}
        run_bulk(items, label_map[search_choice], fn_map[search_choice],
                 f"search_{label_map[search_choice].lower()}")


def _search_single(api, search_choice):
    """Handle single-test search input."""
    if search_choice == 1:
        location = input("\nEnter location (city, ZIP, or address): ").strip()
        if not location:
            print("Location required.")
            return None
        status_options = ["For_Sale", "For_Rent", "Sold"]
        print("\nListing status:")
        sc = prompt_choice(status_options, "Select listing status")
        listing_status = status_options[sc - 1]

        print("\nOptional filters (press Enter to skip):")
        sort_input = input("Sort order (Newest/Price_High_to_Low/Price_Low_to_High): ").strip() or None
        bed_min = input("Min bedrooms (1-5 or Studio): ").strip() or None
        bed_max = input("Max bedrooms (1-5): ").strip() or None
        bathrooms = input("Bathrooms (OnePlus/TwoPlus/ThreePlus/FourPlus): ").strip() or None
        price_range = input("Price range (e.g., min:100000, max:500000): ").strip() or None
        home_type = input("Home type (Houses, Townhomes, Condos/Co-ops, etc.): ").strip() or None
        keywords = input("Keywords (pool, fireplace, etc.): ").strip() or None
        page = input("Page number (1-5, default 1): ").strip()
        page = int(page) if page else 1

        print("\nSearching...")
        return api.search_by_address(
            location=location, listing_status=listing_status, page=page,
            sort_order=sort_input, bed_min=bed_min, bed_max=bed_max,
            bathrooms=bathrooms, list_price_range=price_range,
            home_type=home_type, keywords=keywords,
        )

    elif search_choice == 2:
        lat = input("\nLatitude (e.g., 40.599283): ").strip()
        lng = input("Longitude (e.g., -74.129194): ").strip()
        radius = input("Radius in miles (e.g., 0.5): ").strip()
        if not all([lat, lng, radius]):
            print("Latitude, longitude, and radius required.")
            return None
        status_options = ["For_Sale", "For_Rent", "Sold"]
        print("\nListing status:")
        sc = prompt_choice(status_options, "Select listing status")
        print("\nSearching...")
        return api.search_by_coordinates(lat, lng, radius, status_options[sc - 1])

    elif search_choice == 3:
        url = input("\nEnter Zillow search URL: ").strip()
        if not url:
            print("URL required.")
            return None
        print("\nSearching...")
        return api.search_by_url(url)

    elif search_choice == 4:
        mls = input("\nEnter MLS number: ").strip()
        if not mls:
            print("MLS number required.")
            return None
        print("\nSearching...")
        return api.search_by_mls(mls)

    elif search_choice == 5:
        query = input("\nEnter search text: ").strip()
        if not query:
            print("Query required.")
            return None
        print("\nFetching suggestions...")
        return api.autocomplete(query)


# ─────────────────────────────────────────────
# Property Details
# ─────────────────────────────────────────────

DETAIL_OPTIONS = [
    ("Property Images",               "get_property_images",            "property_images"),
    ("Price History",                  "get_price_history",              "price_history"),
    ("Comparable Homes",              "get_comparable_homes",           "comparable_homes"),
    ("Similar Properties",            "get_similar_properties",         "similar_properties"),
    ("Nearby Properties",             "get_nearby_properties",          "nearby_properties"),
    ("Walk/Transit/Bike Scores",      "get_walk_transit_bike_scores",   "walk_transit_bike"),
    ("Climate Risk Data",             "get_climate_data",               "climate_data"),
    ("Tax History",                   "get_tax_history",                "tax_history"),
    ("Owner/Agent Info",              "get_owner_agent_info",           "owner_agent"),
    ("Zestimate History (10 years)",  "get_zestimate_history",          "zestimate_history"),
]


def handle_property_details(api):
    print("\n--- Property Details ---")
    labels = [opt[0] for opt in DETAIL_OPTIONS]
    detail_choice = prompt_choice(labels)
    _, method_name, base_name = DETAIL_OPTIONS[detail_choice - 1]
    method = getattr(api, method_name)

    mode = ask_mode()

    if mode == 1:
        # Single test - pick identifier type
        print("\nIdentify property by:")
        id_options = ["Address", "ZPID", "Zillow URL"]
        id_choice = prompt_choice(id_options, "Select identifier")

        byzpid, byurl, byaddress = None, None, None
        if id_choice == 1:
            byaddress = input("\nEnter address: ").strip()
        elif id_choice == 2:
            byzpid = input("\nEnter ZPID: ").strip()
        else:
            byurl = input("\nEnter Zillow URL: ").strip()

        if not any([byzpid, byurl, byaddress]):
            return print("Identifier required.")

        print("\nFetching...")
        data = method(byzpid=byzpid, byurl=byurl, byaddress=byaddress)

        print("\n--- Preview (first 1000 chars) ---")
        preview(data)

        save = input("\nSave full output? (y/n): ").strip().lower()
        if save == "y":
            fmt = prompt_output_format()
            save_results(data, base_name, fmt)

    else:
        # Bulk mode - reads property_details from input.json
        input_data = load_input_file()
        if not input_data:
            return
        items = input_data.get("property_details", [])

        def fetch(item):
            return method(
                byzpid=item.get("zpid"),
                byurl=item.get("url"),
                byaddress=item.get("address"),
            )

        run_bulk(items, base_name, fetch, base_name)


# ─────────────────────────────────────────────
# Market Analytics
# ─────────────────────────────────────────────

def handle_market_analytics(api):
    print("\n--- Housing Market Analytics ---")
    mode = ask_mode()

    if mode == 1:
        query = input("Enter city/state/ZIP (or 'USA' for nationwide): ").strip()
        if not query:
            return print("Search query required.")
        home_type = input("Home type (All_Homes/Single_Family/Condo, Enter to skip): ").strip() or None
        rental = input("Include rental market data? (y/n, default n): ").strip().lower()

        print("\nFetching...")
        data = api.get_housing_market(
            search_query=query, home_type=home_type,
            exclude_rental_trends=False if rental == "y" else None,
        )
        print("\n--- Preview (first 1000 chars) ---")
        preview(data)

        save = input("\nSave full output? (y/n): ").strip().lower()
        if save == "y":
            fmt = prompt_output_format()
            save_results(data, "housing_market", fmt)
    else:
        input_data = load_input_file()
        if not input_data:
            return
        items = input_data.get("market_queries", [])

        def fetch(item):
            if isinstance(item, str):
                return api.get_housing_market(search_query=item)
            return api.get_housing_market(
                search_query=item["search_query"],
                home_type=item.get("home_type"),
                exclude_rental_trends=item.get("exclude_rental_trends"),
            )

        run_bulk(items, "market query", fetch, "housing_market")


# ─────────────────────────────────────────────
# Agent Search
# ─────────────────────────────────────────────

def handle_agent_search(api):
    print("\n--- Find an Agent ---")
    options = [
        "Search Agents",
        "Agent Details (by URL or username)",
        "Agent's For-Sale Listings",
        "Agent's Rental Listings",
        "Agent's Sold Properties",
        "Agent Reviews",
    ]
    agent_choice = prompt_choice(options)
    mode = ask_mode()

    if mode == 1:
        data = _agent_single(api, agent_choice)
        if data is None:
            return

        print("\n--- Preview (first 1000 chars) ---")
        preview(data)

        save = input("\nSave full output? (y/n): ").strip().lower()
        if save == "y":
            fmt = prompt_output_format()
            name_map = {
                1: "agent_search", 2: "agent_details", 3: "agent_for_sale",
                4: "agent_for_rent", 5: "agent_sold", 6: "agent_reviews",
            }
            save_results(data, name_map[agent_choice], fmt)
    else:
        input_data = load_input_file()
        if not input_data:
            return

        if agent_choice == 1:
            items = input_data.get("agent_searches", [])
            def fetch(item):
                if isinstance(item, str):
                    return api.search_agents(location=item)
                return api.search_agents(
                    location=item.get("location"),
                    agent_name=item.get("agent_name"),
                    is_top_agent=item.get("is_top_agent"),
                    specialties=item.get("specialties"),
                    languages=item.get("languages"),
                    page=item.get("page", 1),
                )
            run_bulk(items, "agent search", fetch, "agent_search")
        else:
            items = input_data.get("agent_profiles", [])
            method_map = {
                2: "get_agent_details", 3: "get_agent_for_sale",
                4: "get_agent_for_rent", 5: "get_agent_sold",
                6: "get_agent_reviews",
            }
            method = getattr(api, method_map[agent_choice])
            def fetch(item):
                if isinstance(item, str):
                    if item.startswith("http"):
                        return method(agent_link=item)
                    return method(username=item)
                return method(
                    agent_link=item.get("agent_link"),
                    username=item.get("username"),
                )
            name_map = {
                2: "agent_details", 3: "agent_for_sale",
                4: "agent_for_rent", 5: "agent_sold", 6: "agent_reviews",
            }
            run_bulk(items, "agent profile", fetch, name_map[agent_choice])


def _agent_single(api, agent_choice):
    if agent_choice == 1:
        location = input("\nLocation (city/ZIP, Enter to skip): ").strip() or None
        name = input("Agent name (Enter to skip): ").strip() or None
        top_only = input("Top agents only? (y/n, default n): ").strip().lower()
        specialties = input("Specialties (e.g., luxury-homes,new-construction): ").strip() or None
        print("\nSearching agents...")
        return api.search_agents(
            location=location, agent_name=name,
            is_top_agent=True if top_only == "y" else None,
            specialties=specialties,
        )
    else:
        agent_link = input("\nAgent Zillow URL (Enter to skip): ").strip() or None
        username = input("Agent username (Enter to skip): ").strip() or None
        if not agent_link and not username:
            print("Provide either agent URL or username.")
            return None
        method_map = {
            2: "get_agent_details", 3: "get_agent_for_sale",
            4: "get_agent_for_rent", 5: "get_agent_sold",
            6: "get_agent_reviews",
        }
        method = getattr(api, method_map[agent_choice])
        print("\nFetching...")
        return method(agent_link=agent_link, username=username)


# ─────────────────────────────────────────────
# Skip Tracing
# ─────────────────────────────────────────────

def handle_skip_trace(api):
    print("\n--- Skip Tracing ---")
    print("Note: Each skip trace call costs 10 API requests.\n")
    mode = ask_mode()

    if mode == 1:
        street = input("Street address (e.g., 3828 Double Oak Ln): ").strip()
        citystatezip = input("City, State ZIP (e.g., Irving, TX 75061): ").strip()
        if not street or not citystatezip:
            return print("Both street and city/state/zip required.")

        print("\nRunning skip trace...")
        data = api.skip_trace_by_address(street, citystatezip)

        print("\n--- Preview (first 1000 chars) ---")
        preview(data)

        save = input("\nSave full output? (y/n): ").strip().lower()
        if save == "y":
            fmt = prompt_output_format()
            save_results(data, "skip_trace", fmt)
    else:
        input_data = load_input_file()
        if not input_data:
            return
        items = input_data.get("skip_traces", [])

        def fetch(item):
            return api.skip_trace_by_address(
                street=item["street"],
                citystatezip=item["citystatezip"],
                page=item.get("page", 1),
            )

        run_bulk(items, "skip trace", fetch, "skip_trace")


# ─────────────────────────────────────────────
# Main
# ─────────────────────────────────────────────

def main():
    print("=" * 60)
    print("  Zillow Scraper - Powered by RealtyAPI.io")
    print("  https://www.realtyapi.io")
    print("=" * 60)

    api_key = ensure_api_key()
    if not api_key:
        sys.exit(1)

    api = ZillowAPI(api_key)

    while True:
        print("\n" + "-" * 50)
        print("  What would you like to do?")
        print("-" * 50)
        options = [
            "Property Lookup (by address, ZPID, or URL)",
            "Search Zillow Listings",
            "Property Details (images, comps, scores, history)",
            "Housing Market Analytics (ZHVI)",
            "Find a Real Estate Agent",
            "Skip Tracing",
            "Exit",
        ]
        choice = prompt_choice(options)

        handlers = {
            1: handle_property_lookup,
            2: handle_search,
            3: handle_property_details,
            4: handle_market_analytics,
            5: handle_agent_search,
            6: handle_skip_trace,
        }

        if choice == 7:
            print("\nGoodbye!")
            break

        handlers[choice](api)


if __name__ == "__main__":
    main()