""" Car search crawler -- finds used hybrids under $10K near Woodinville WA. Playwright Chromium always runs headed on Xvfb. View live at browser.ampbox.io. """ import asyncio import json 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) 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_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 search_facebook(page) -> list[dict]: """Facebook Marketplace -- view at browser.ampbox.io to log in if needed.""" url = "https://www.facebook.com/marketplace/seattle/vehicles/?maxPrice=10000" print(f" Opening Facebook Marketplace...") await page.goto(url, wait_until="networkidle", timeout=60000) if "login" in page.url.lower(): print(" Facebook needs login -- open https://browser.ampbox.io/vnc.html to log in") print(" Press Enter when done...") await asyncio.get_event_loop().run_in_executor(None, input) await page.wait_for_timeout(3000) 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 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("Live browser: https://browser.ampbox.io/vnc.html") 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, search_facebook]: 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[: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())