scraper.py

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

Scrape Redfin property data, search listings, get valuations,
walk scores, agent info, and more - 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 redfin_api import RedfinAPI

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):
    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="_"):
    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):
    os.makedirs(os.path.dirname(filepath), exist_ok=True)
    if not records:
        print("No data to save.")
        return
    flat_records = []
    for r in records:
        if isinstance(r, dict):
            flat_records.append(flatten_dict(r))
        else:
            flat_records.append({"data": json.dumps(r)})
    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):
    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():
    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=""):
    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):
    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)})
        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():
    options = [
        "Test one (enter input manually, see preview)",
        "Bulk run (read from input.json, save all results)",
    ]
    return prompt_choice(options, "Select mode")


# ─────────────────────────────────────────────
# Property Search
# ─────────────────────────────────────────────

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

    if mode == 1:
        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:
        input_data = load_input_file()
        if not input_data:
            return
        key_map = {
            1: "search_locations",
            2: "search_coordinates",
            3: "search_urls",
            4: "search_region_ids",
            5: "autocomplete_queries",
        }
        key = key_map[search_choice]
        items = input_data.get(key, [])

        fn_map = {
            1: lambda item: api.search_by_location(
                location_name=item.get("location", item) if isinstance(item, dict) else item,
                search_type=item.get("search_type") if isinstance(item, dict) else None,
                page=item.get("page", 1) if isinstance(item, dict) else 1,
                result_count=item.get("result_count") if isinstance(item, dict) else None,
                sort_order=item.get("sort_order") if isinstance(item, dict) else None,
                min_price=item.get("min_price") if isinstance(item, dict) else None,
                max_price=item.get("max_price") if isinstance(item, dict) else None,
                min_beds=item.get("min_beds") if isinstance(item, dict) else None,
                max_beds=item.get("max_beds") if isinstance(item, dict) else None,
                baths=item.get("baths") if isinstance(item, dict) else None,
                home_type=item.get("home_type") if isinstance(item, dict) else None,
                keyword=item.get("keyword") if isinstance(item, dict) else None,
            ),
            2: lambda item: api.search_by_coordinates(
                latitude=item["latitude"],
                longitude=item["longitude"],
                radius=item.get("radius", "1"),
                search_type=item.get("search_type"),
                result_count=item.get("result_count"),
            ),
            3: lambda item: api.search_by_url(
                search_url=item.get("url", item) if isinstance(item, dict) else item,
                result_count=item.get("result_count") if isinstance(item, dict) else None,
            ),
            4: lambda item: api.search_by_region_id(
                region_id=item.get("region_id", item) if isinstance(item, dict) else item,
                search_type=item.get("search_type") if isinstance(item, dict) else None,
            ),
            5: lambda q: api.autocomplete(q),
        }
        label_map = {1: "location", 2: "coordinate", 3: "URL", 4: "region ID", 5: "autocomplete"}
        run_bulk(items, label_map[search_choice], fn_map[search_choice],
                 f"search_{label_map[search_choice].lower().replace(' ', '_')}")


def _search_single(api, search_choice):
    if search_choice == 1:
        location = input("\nEnter location (city, state, ZIP): ").strip()
        if not location:
            print("Location required.")
            return None
        type_options = ["For_Sale", "For_Rent", "Sold"]
        print("\nSearch type:")
        tc = prompt_choice(type_options, "Select search type")
        search_type = type_options[tc - 1]

        print("\nOptional filters (press Enter to skip):")
        sort = input("Sort (Newest/Price_High_to_Low/Price_Low_to_High): ").strip() or None
        min_price = input("Min price (e.g., 100000): ").strip() or None
        max_price = input("Max price (e.g., 500000): ").strip() or None
        min_beds = input("Min beds (Any/Studio/One/Two/Three/Four/FivePlus): ").strip() or None
        baths = input("Baths (Any/OnePlus/TwoPointFivePlus/ThreePlus): ").strip() or None
        home_type = input("Home type (House,Condo,Townhouse,Apartment,Land): ").strip() or None
        keyword = input("Keyword (pool, view, etc.): ").strip() or None
        page = input("Page (default 1): ").strip()
        page = int(page) if page else 1

        print("\nSearching...")
        return api.search_by_location(
            location_name=location, search_type=search_type, page=page,
            sort_order=sort, min_price=min_price, max_price=max_price,
            min_beds=min_beds, baths=baths, home_type=home_type, keyword=keyword,
        )

    elif search_choice == 2:
        lat = input("\nLatitude (e.g., 40.748817): ").strip()
        lng = input("Longitude (e.g., -73.985428): ").strip()
        radius = input("Radius in miles (e.g., 1.5): ").strip()
        if not all([lat, lng, radius]):
            print("Latitude, longitude, and radius required.")
            return None
        type_options = ["For_Sale", "For_Rent", "Sold"]
        print("\nSearch type:")
        tc = prompt_choice(type_options, "Select search type")
        print("\nSearching...")
        return api.search_by_coordinates(lat, lng, radius, search_type=type_options[tc - 1])

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

    elif search_choice == 4:
        rid = input("\nEnter Region ID (e.g., 4_3244): ").strip()
        if not rid:
            print("Region ID required.")
            return None
        print("\nSearching...")
        return api.search_by_region_id(rid)

    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
# ─────────────────────────────────────────────

def handle_property_details(api):
    print("\n--- Property Details ---")
    options = [
        "Details by Address",
        "Details by Property ID + Listing ID",
        "Details by Redfin URL",
    ]
    detail_choice = prompt_choice(options)
    mode = ask_mode()

    if mode == 1:
        if detail_choice == 1:
            addr = input("\nEnter property address: ").strip()
            if not addr: return print("Address required.")
            print("\nFetching...")
            data = api.get_details_by_address(addr)
        elif detail_choice == 2:
            pid = input("\nEnter property ID: ").strip()
            lid = input("Enter listing ID: ").strip()
            if not pid or not lid: return print("Both IDs required.")
            print("\nFetching...")
            data = api.get_details_by_id(pid, lid)
        else:
            url = input("\nEnter Redfin property URL: ").strip()
            if not url: return print("URL required.")
            print("\nFetching...")
            data = api.get_details_by_url(url)

        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_details", fmt)
    else:
        input_data = load_input_file()
        if not input_data:
            return
        key_map = {1: "property_addresses", 2: "property_ids", 3: "property_urls"}
        key = key_map[detail_choice]
        items = input_data.get(key, [])

        fn_map = {
            1: lambda addr: api.get_details_by_address(addr),
            2: lambda item: api.get_details_by_id(item["property_id"], item["listing_id"]),
            3: lambda url: api.get_details_by_url(url),
        }
        label_map = {1: "address", 2: "ID pair", 3: "URL"}
        run_bulk(items, label_map[detail_choice], fn_map[detail_choice],
                 f"details_by_{label_map[detail_choice].lower().replace(' ', '_')}")


# ─────────────────────────────────────────────
# Individual Property Data
# ─────────────────────────────────────────────

INDIVIDUAL_OPTIONS = [
    ("Agent Info",              "get_agent_info",              "agent_info",       "id_pair"),
    ("Amenities",               "get_amenities",               "amenities",        "id_pair"),
    ("AVM Estimate (Valuation)","get_avm_estimate",            "avm_estimate",     "id_pair"),
    ("Basic Details (by URL)",  "get_basic_details",           "basic_details",    "url"),
    ("Customer Conversion Info","get_customer_conversion_info","customer_conv",    "id_pair"),
    ("Flood Risk Info",         "get_flood_info",              "flood_info",       "flood"),
    ("Tour Date Picker",        "get_date_picker_data",        "date_picker",      "listing_id"),
    ("Hot Market Info",         "get_hot_market_info",         "hot_market",       "property_id"),
    ("Insights",                "get_insights",                "insights",         "id_pair"),
    ("Main House Info Panel",   "get_main_house_info",         "main_house_info",  "id_pair"),
    ("Mortgage Calculator",     "get_mortgage_calculator",     "mortgage_calc",    "id_pair"),
    ("Overview",                "get_overview",                "overview",         "id_pair"),
    ("Popularity Info",         "get_popularity_info",         "popularity",       "listing_id"),
    ("Price Drop Info",         "get_price_drop_info",         "price_drop",       "listing_id"),
    ("Tour Insights",           "get_tour_insights",           "tour_insights",    "id_pair"),
    ("Walk/Transit/Bike Score", "get_walk_score",              "walk_score",       "id_pair"),
]


def handle_individual_details(api):
    print("\n--- Individual Property Data ---")
    labels = [opt[0] for opt in INDIVIDUAL_OPTIONS]
    choice = prompt_choice(labels)
    _, method_name, base_name, input_type = INDIVIDUAL_OPTIONS[choice - 1]
    method = getattr(api, method_name)
    mode = ask_mode()

    if mode == 1:
        data = _individual_single(method, input_type)
        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, base_name, fmt)
    else:
        input_data = load_input_file()
        if not input_data:
            return

        # For bulk, most individual endpoints use "property_ids" (id pairs)
        if input_type == "id_pair":
            items = input_data.get("property_ids", [])
            fetch = lambda item: method(item["property_id"], item["listing_id"])
        elif input_type == "url":
            items = input_data.get("property_urls", [])
            fetch = lambda url: method(url)
        elif input_type == "listing_id":
            items = input_data.get("listing_ids", [])
            fetch = lambda lid: method(lid)
        elif input_type == "property_id":
            items = input_data.get("property_id_only", [])
            fetch = lambda pid: method(pid)
        elif input_type == "flood":
            items = input_data.get("flood_queries", [])
            fetch = lambda item: method(item["fips_code"], item["apn"], item["lat"], item["lng"])
        else:
            print("Unsupported bulk input type.")
            return

        run_bulk(items, base_name, fetch, base_name)


def _individual_single(method, input_type):
    if input_type == "id_pair":
        pid = input("\nEnter property ID: ").strip()
        lid = input("Enter listing ID: ").strip()
        if not pid or not lid:
            print("Both property ID and listing ID required.")
            return None
        print("\nFetching...")
        return method(pid, lid)

    elif input_type == "url":
        url = input("\nEnter Redfin property URL: ").strip()
        if not url:
            print("URL required.")
            return None
        print("\nFetching...")
        return method(url)

    elif input_type == "listing_id":
        lid = input("\nEnter listing ID: ").strip()
        if not lid:
            print("Listing ID required.")
            return None
        print("\nFetching...")
        return method(lid)

    elif input_type == "property_id":
        pid = input("\nEnter property ID: ").strip()
        if not pid:
            print("Property ID required.")
            return None
        print("\nFetching...")
        return method(pid)

    elif input_type == "flood":
        fips = input("\nEnter FIPS county code (e.g., 36081): ").strip()
        apn = input("Enter Assessor's Parcel Number (e.g., 4109420115): ").strip()
        lat = input("Enter latitude: ").strip()
        lng = input("Enter longitude: ").strip()
        if not all([fips, apn, lat, lng]):
            print("All flood info fields required.")
            return None
        print("\nFetching...")
        return method(fips, apn, lat, lng)


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

def main():
    print("=" * 60)
    print("  Redfin 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 = RedfinAPI(api_key)

    while True:
        print("\n" + "-" * 50)
        print("  What would you like to do?")
        print("-" * 50)
        options = [
            "Search Redfin Listings",
            "Property Details (by address, ID, or URL)",
            "Individual Property Data (agent, AVM, scores, flood, etc.)",
            "Exit",
        ]
        choice = prompt_choice(options)

        if choice == 1:
            handle_search(api)
        elif choice == 2:
            handle_property_details(api)
        elif choice == 3:
            handle_individual_details(api)
        elif choice == 4:
            print("\nGoodbye!")
            break


if __name__ == "__main__":
    main()