Ken
f30af4aefd
- Parse .cl-search-result inner text for title, price, location - Fallback to URL slug extraction for missing titles - Capture listing images for display in UI - AutoTempest scraper also improved with better selectors
252 lines
8.2 KiB
Python
252 lines
8.2 KiB
Python
"""
|
|
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())
|