feat: initial car search setup with Playwright crawler and sites list

This commit is contained in:
Ken
2026-05-24 22:56:12 +00:00
commit 290678635e
7 changed files with 431 additions and 0 deletions
+196
View File
@@ -0,0 +1,196 @@
"""
Car search crawler -- finds used hybrids under $10K near Woodinville WA.
Uses Playwright (headless Chromium) to handle JS-heavy car sites.
"""
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: {url}")
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 main():
print(f"Car Search -- {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=True)
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: {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())