""" Car search crawler -- finds used hybrids under $10K near Woodinville WA. Uses Playwright (headless Chromium) to handle JS-heavy car sites. Two modes: headless (default): uv run python search.py headed via noVNC: docker run ... (see README) When running in the Docker container, the browser is visible at http://localhost:6080/vnc.html -- use this to handle CAPTCHAs, log into Facebook Marketplace, or visually inspect results. """ import asyncio import json import os 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) # If DISPLAY is set (noVNC container), run headed HEADED = bool(os.environ.get("DISPLAY")) async def pause_for_human(page, reason: str): """Pause and prompt for human interaction via noVNC.""" if not HEADED: print(f" SKIP (headless): {reason}") return False print(f"\n{'=' * 60}") print(f" HUMAN NEEDED: {reason}") print(f" Open http://localhost:6080/vnc.html to interact") print(f" Press Enter here when done...") print(f"{'=' * 60}") # Wait for user to press Enter in the terminal await asyncio.get_event_loop().run_in_executor(None, input) return True async def search_craigslist(page) -> list[dict]: """Craigslist Seattle -- simple HTML, most reliable to scrape.""" url = ( "https://seattle.craigslist.org/search/cta" "?auto_fuel_type=4" "&max_price=10000" "&postal=98077" "&search_distance=50" "&sort=date" ) print(f" Searching Craigslist...") await page.goto(url, wait_until="domcontentloaded", timeout=30000) await page.wait_for_timeout(2000) listings = [] cards = await page.query_selector_all( ".cl-static-search-result, .result-row, .cl-search-result" ) for card in cards[:50]: try: title_el = await card.query_selector(".titlestring, .result-title, a") price_el = await card.query_selector(".priceinfo, .result-price") title = await title_el.inner_text() if title_el else "Unknown" price = await price_el.inner_text() if price_el else "N/A" link_el = await card.query_selector("a[href]") link = await link_el.get_attribute("href") if link_el else "" listings.append({ "source": "craigslist", "title": title.strip(), "price": price.strip(), "url": link, }) except Exception: continue print(f" Found {len(listings)} Craigslist listings") return listings async def search_cargurus(page) -> list[dict]: """CarGurus -- JS-heavy but good data.""" url = ( "https://www.cargurus.com/Cars/inventorylisting/" "viewDetailsFilterViewInventoryListing.action" "?zip=98077&maxPrice=10000&fuelTypes=HYBRID" "&distance=50&sortDir=ASC&sortType=PRICE" ) print(f" Searching CarGurus...") await page.goto(url, wait_until="networkidle", timeout=60000) await page.wait_for_timeout(3000) listings = [] cards = await page.query_selector_all( "[data-cg-ft='car-blade'], .pazLpc, article" ) for card in cards[:50]: try: title_el = await card.query_selector( "h4, .pazLpc a, [data-cg-ft='car-blade-link']" ) price_el = await card.query_selector( "[data-cg-ft='car-blade-price'], .pazLpc .JzvSTe" ) title = await title_el.inner_text() if title_el else "Unknown" price = await price_el.inner_text() if price_el else "N/A" link_el = await card.query_selector("a[href*='/Cars/']") 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}" listings.append({ "source": "cargurus", "title": title.strip(), "price": price.strip(), "url": link, }) except Exception: continue print(f" Found {len(listings)} CarGurus listings") return listings async def search_facebook(page) -> list[dict]: """Facebook Marketplace -- needs login (human via noVNC).""" if not HEADED: print(" SKIP Facebook Marketplace (requires login -- run in Docker with noVNC)") return [] url = "https://www.facebook.com/marketplace/seattle/vehicles/?maxPrice=10000" print(f" Opening Facebook Marketplace...") await page.goto(url, wait_until="networkidle", timeout=60000) # Check if login is needed if "login" in page.url.lower(): await pause_for_human(page, "Log into Facebook, then press Enter") await page.wait_for_timeout(3000) await pause_for_human(page, "Apply filters (hybrid, $10K max, 50mi) and press Enter when ready") # Scrape whatever is visible listings = [] cards = await page.query_selector_all("[class*='marketplace']") for card in cards[:30]: try: text = await card.inner_text() link_el = await card.query_selector("a[href*='/marketplace/']") link = await link_el.get_attribute("href") if link_el else "" listings.append({ "source": "facebook", "title": text[:100].strip(), "price": "see listing", "url": f"https://www.facebook.com{link}" if link else "", }) except Exception: continue print(f" Found {len(listings)} Facebook listings") return listings async def search_autotempest(page) -> list[dict]: """AutoTempest -- meta-aggregator, searches multiple sites.""" url = ( "https://www.autotempest.com/results" "?zip=98077&maxprice=10000&fuel=hybrid&radius=50" ) print(f" Searching AutoTempest...") await page.goto(url, wait_until="networkidle", timeout=60000) await page.wait_for_timeout(5000) listings = [] cards = await page.query_selector_all( ".result-row, .listing-row, [class*='result']" ) for card in cards[:50]: try: title_el = await card.query_selector(".title, h3, a") price_el = await card.query_selector(".price, [class*='price']") title = await title_el.inner_text() if title_el else "Unknown" price = await price_el.inner_text() if price_el else "N/A" link_el = await card.query_selector("a[href]") link = await link_el.get_attribute("href") if link_el else "" listings.append({ "source": "autotempest", "title": title.strip(), "price": price.strip(), "url": link, }) except Exception: continue print(f" Found {len(listings)} AutoTempest listings") return listings async def main(): mode = "HEADED (noVNC)" if HEADED else "HEADLESS" print(f"Car Search [{mode}] -- {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}") print( "Looking for: Used hybrids under $10K " "within 50 miles of 98077 (Woodinville WA)" ) print("=" * 70) all_listings: list[dict] = [] async with async_playwright() as p: browser = await p.chromium.launch(headless=not HEADED) 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() searches = [search_craigslist, search_cargurus, search_autotempest] if HEADED: searches.append(search_facebook) for search_fn in searches: try: results = await search_fn(page) all_listings.extend(results) except Exception as e: print(f" ERROR in {search_fn.__name__}: {e}") await browser.close() # Save results timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M") output_file = RESULTS_DIR / f"search_{timestamp}.json" with open(output_file, "w") as f: json.dump( { "search_date": datetime.now(timezone.utc).isoformat(), "mode": mode, "params": { "zip": "98077", "max_price": 10000, "fuel": "hybrid", "radius_miles": 50, }, "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[:20]: print(f" [{listing['source']}] {listing['price']:>8s} {listing['title']}") if len(all_listings) > 20: print(f" ... and {len(all_listings) - 20} more (see {output_file})") if __name__ == "__main__": asyncio.run(main())