""" Car search crawler -- finds used cars based on dynamic parameters. Always runs headed on Xvfb. View live at browser.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())