research-workbench/search.py

"""
Car search crawler -- finds used cars based on dynamic parameters.
Always runs headed on Xvfb. View live at research.ampbox.io.

Usage:
  python search.py                   # runs default hardcoded search
  python search.py --dynamic '{...}' # runs with dynamic URLs/params from app.py
"""

import asyncio
import json
import re
import sys
from datetime import datetime, timezone
from pathlib import Path

from playwright.async_api import async_playwright

RESULTS_DIR = Path("results")
RESULTS_DIR.mkdir(exist_ok=True)


def extract_title_from_url(url: str) -> str:
    """Extract a readable title from a Craigslist URL slug."""
    m = re.search(r"/d/([^/]+)/", url)
    if m:
        return m.group(1).replace("-", " ").title()
    return ""


async def search_craigslist(page, url: str) -> list[dict]:
    """Craigslist -- simple HTML, most reliable to scrape."""
    print(f"  Searching Craigslist: {url}")
    await page.goto(url, wait_until="domcontentloaded", timeout=30000)
    await page.wait_for_timeout(3000)

    listings = []
    cards = await page.query_selector_all(".cl-search-result")
    print(f"  Found {len(cards)} result elements")

    for card in cards[:50]:
        try:
            text = await card.inner_text()
            lines = [
                ln.strip()
                for ln in text.split("\n")
                if ln.strip() and ln.strip() != "\u2022"
            ]

            link_el = await card.query_selector(
                "a.main, a.cl-app-anchor, a[href*='/d/']"
            )
            link = await link_el.get_attribute("href") if link_el else ""

            title = ""
            price = ""
            for line in lines:
                if line.startswith("$"):
                    price = line
                elif re.match(r"^\d+/\d+$", line):
                    continue
                elif re.match(r"^\d+k?\s*mi", line, re.I):
                    continue
                elif not title and len(line) > 5:
                    title = line

            if not title and link:
                title = extract_title_from_url(link)

            if not price:
                price_el = await card.query_selector(".priceinfo")
                if price_el:
                    price = await price_el.inner_text()

            img_el = await card.query_selector("img[src]")
            img = await img_el.get_attribute("src") if img_el else ""

            listings.append(
                {
                    "source": "craigslist",
                    "title": title.strip(),
                    "price": price.strip(),
                    "url": link,
                    "image": img,
                }
            )
        except Exception:
            continue

    print(f"  Extracted {len(listings)} Craigslist listings")
    return listings


async def search_cargurus(page, url: str) -> list[dict]:
    """CarGurus -- JS-heavy but good data."""
    print(f"  Searching CarGurus: {url}")
    await page.goto(url, wait_until="networkidle", timeout=60000)
    await page.wait_for_timeout(3000)

    listings = []
    cards = await page.query_selector_all(
        "article, [data-cg-ft='car-blade'], a[href*='/Cars/']"
    )
    for card in cards[:50]:
        try:
            text = await card.inner_text()
            lines = [ln.strip() for ln in text.split("\n") if ln.strip()]

            title = lines[0] if lines else "Unknown"
            price = ""
            for line in lines:
                if "$" in line and any(c.isdigit() for c in line):
                    price = line
                    break

            link_el = await card.query_selector("a[href*='/Cars/']")
            if not link_el:
                link_el = card if await card.get_attribute("href") else None
            link = await link_el.get_attribute("href") if link_el else ""
            if link and not link.startswith("http"):
                link = f"https://www.cargurus.com{link}"

            img_el = await card.query_selector("img[src*='cargurus']")
            img = await img_el.get_attribute("src") if img_el else ""

            listings.append(
                {
                    "source": "cargurus",
                    "title": title[:100].strip(),
                    "price": price.strip(),
                    "url": link,
                    "image": img,
                }
            )
        except Exception:
            continue

    print(f"  Found {len(listings)} CarGurus listings")
    return listings


async def search_autotempest(page, url: str) -> list[dict]:
    """AutoTempest -- meta-aggregator."""
    print(f"  Searching AutoTempest: {url}")
    await page.goto(url, wait_until="networkidle", timeout=60000)
    await page.wait_for_timeout(5000)

    listings = []
    cards = await page.query_selector_all(
        ".result-list-item, .listing, [class*='result']"
    )
    for card in cards[:50]:
        try:
            text = await card.inner_text()
            if len(text.strip()) < 10:
                continue

            lines = [ln.strip() for ln in text.split("\n") if ln.strip()]
            title = lines[0] if lines else "Unknown"
            price = ""
            for line in lines:
                if "$" in line and any(c.isdigit() for c in line):
                    price = line
                    break

            link_el = await card.query_selector("a[href]")
            link = await link_el.get_attribute("href") if link_el else ""

            img_el = await card.query_selector("img[src]")
            img = await img_el.get_attribute("src") if img_el else ""

            listings.append(
                {
                    "source": "autotempest",
                    "title": title[:100].strip(),
                    "price": price.strip(),
                    "url": link,
                    "image": img,
                }
            )
        except Exception:
            continue

    print(f"  Found {len(listings)} AutoTempest listings")
    return listings


async def main():
    print(f"Car Search -- {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}")

    # Parse dynamic args if provided
    dynamic_config = None
    if len(sys.argv) >= 3 and sys.argv[1] == "--dynamic":
        dynamic_config = json.loads(sys.argv[2])
        params = dynamic_config.get("params", {})
        search_summary = params.get("search_summary", "dynamic search")
        print(f"Dynamic search: {search_summary}")
    else:
        print("Default search: Used hybrids under $10K within 50mi of 98077")

    print("=" * 70)

    all_listings: list[dict] = []

    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=False)
        context = await browser.new_context(
            user_agent=(
                "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
                "(KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36"
            ),
            viewport={"width": 1280, "height": 720},
        )
        page = await context.new_page()

        if dynamic_config:
            # Dynamic search with provided URLs
            searches = [
                (search_craigslist, dynamic_config.get("craigslist_url", "")),
                (search_cargurus, dynamic_config.get("cargurus_url", "")),
                (search_autotempest, dynamic_config.get("autotempest_url", "")),
            ]
            for search_fn, url in searches:
                if url:
                    try:
                        results = await search_fn(page, url)
                        all_listings.extend(results)
                    except Exception as e:
                        print(f"  ERROR in {search_fn.__name__}: {e}")
        else:
            # Default hardcoded search
            default_urls = {
                "craigslist": (
                    "https://seattle.craigslist.org/search/cta"
                    "?auto_fuel_type=4&max_price=10000&postal=98077"
                    "&search_distance=50&sort=date"
                ),
                "cargurus": (
                    "https://www.cargurus.com/Cars/inventorylisting/"
                    "viewDetailsFilterViewInventoryListing.action"
                    "?zip=98077&maxPrice=10000&fuelTypes=HYBRID"
                    "&distance=50&sortDir=ASC&sortType=PRICE"
                ),
                "autotempest": (
                    "https://www.autotempest.com/results"
                    "?zip=98077&maxprice=10000&fuel=hybrid&radius=50"
                ),
            }
            for name, url in default_urls.items():
                search_fn = {
                    "craigslist": search_craigslist,
                    "cargurus": search_cargurus,
                    "autotempest": search_autotempest,
                }[name]
                try:
                    results = await search_fn(page, url)
                    all_listings.extend(results)
                except Exception as e:
                    print(f"  ERROR in {name}: {e}")

        await browser.close()

    # Save results
    timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M")
    search_params = (
        dynamic_config.get("params", {})
        if dynamic_config
        else {"zip": "98077", "max_price": 10000, "fuel": "hybrid", "radius_miles": 50}
    )
    output_file = RESULTS_DIR / f"search_{timestamp}.json"
    with open(output_file, "w") as f:
        json.dump(
            {
                "search_date": datetime.now(timezone.utc).isoformat(),
                "params": search_params,
                "total_listings": len(all_listings),
                "listings": all_listings,
            },
            f,
            indent=2,
        )

    print("=" * 70)
    print(f"Total: {len(all_listings)} listings found")
    print(f"Saved to: {output_file}")

    for listing in all_listings[:10]:
        print(
            f"  [{listing['source']}] {listing['price']:>8s}  {listing['title'][:60]}"
        )

    if len(all_listings) > 10:
        print(f"  ... and {len(all_listings) - 10} more")


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