""" Car search crawler -- finds used hybrids under $10K near Woodinville WA. Always runs headed on Xvfb. View live at browser.ampbox.io. """ import asyncio import json import re 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) -> 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(3000) listings = [] cards = await page.query_selector_all(".cl-search-result") print(f" Found {len(cards)} result elements") for card in cards[:50]: try: # Get the full text of the card -- title is in there text = await card.inner_text() lines = [l.strip() for l in text.split("\n") if l.strip() and l.strip() != "•"] # Get the link 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 is usually the longest non-price, non-date text line title = "" price = "" location = "" for line in lines: if line.startswith("$"): price = line elif re.match(r"^\d+/\d+$", line): continue # date like 5/23 elif re.match(r"^\d+k?\s*mi", line, re.I): continue # mileage elif not title and len(line) > 5: title = line # Fallback: extract from URL 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() # Get image 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) -> 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 = [] # Try multiple selector patterns for CarGurus 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 = [l.strip() for l in text.split("\n") if l.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) -> list[dict]: """AutoTempest -- meta-aggregator.""" 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-list-item, .listing, [class*='result']") for card in cards[:50]: try: text = await card.inner_text() if len(text.strip()) < 10: continue lines = [l.strip() for l in text.split("\n") if l.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')}") print("Looking for: Used hybrids under $10K within 50mi of 98077 (Woodinville WA)") 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() for search_fn in [search_craigslist, search_cargurus, search_autotempest]: 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(), "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[: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())