Ken
7e19b740cb
- Search box: type anything, toggle filter pills (hybrid, price range, distance, etc.) - Backend calls Anthropic to parse query into search params - Playwright crawls Craigslist, CarGurus, AutoTempest with dynamic URLs - Results scored/ranked with car-specific scoring engine - Second Anthropic call generates a custom buying guide from actual results - Guide sections: top picks, checklist, mechanics, negotiation, budget - All visible live in the browser panel via noVNC
535 lines
19 KiB
Python
535 lines
19 KiB
Python
"""
|
|
Car-help web app -- FastAPI backend with Anthropic-powered search + guide generation.
|
|
|
|
Runs on port 8080 inside the Docker container.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
app = FastAPI(title="car-help", version="0.2.0")
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
RESULTS_DIR = Path("results")
|
|
RESULTS_DIR.mkdir(exist_ok=True)
|
|
NOVNC_UPSTREAM = "http://localhost:6080"
|
|
|
|
ANTHROPIC_BASE_URL = os.environ.get("ANTHROPIC_BASE_URL", "https://api.anthropic.com")
|
|
ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY", "")
|
|
ANTHROPIC_MODEL = "claude-sonnet-4-20250514"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Anthropic API helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def call_anthropic(prompt: str, max_tokens: int = 2000, timeout: int = 60) -> str:
|
|
"""Call the Anthropic Messages API and return the text response."""
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
f"{ANTHROPIC_BASE_URL}/v1/messages",
|
|
headers={
|
|
"x-api-key": ANTHROPIC_API_KEY,
|
|
"content-type": "application/json",
|
|
"anthropic-version": "2023-06-01",
|
|
},
|
|
json={
|
|
"model": ANTHROPIC_MODEL,
|
|
"max_tokens": max_tokens,
|
|
"messages": [{"role": "user", "content": prompt}],
|
|
},
|
|
timeout=timeout,
|
|
)
|
|
data = resp.json()
|
|
if "content" in data and data["content"]:
|
|
return data["content"][0].get("text", "")
|
|
return json.dumps(data)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Parse user query into search parameters via LLM
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def parse_query(query: str, filters: list[str]) -> dict:
|
|
"""Use the LLM to turn a natural language query + filters into search params."""
|
|
filter_str = ", ".join(filters) if filters else "none"
|
|
prompt = f"""You are a car search assistant. Parse this user query and active filters into concrete search parameters.
|
|
|
|
User query: "{query}"
|
|
Active filters: {filter_str}
|
|
|
|
Return ONLY valid JSON (no markdown, no explanation) with these keys:
|
|
- "makes": list of car makes to search (e.g. ["Toyota", "Honda"]). Empty list = any make.
|
|
- "models": list of specific models (e.g. ["Prius", "Camry Hybrid"]). Empty list = any model.
|
|
- "min_year": minimum year (integer, e.g. 2010). null if not specified.
|
|
- "max_year": maximum year (integer). null if not specified.
|
|
- "max_price": maximum price in dollars (integer). Default 10000.
|
|
- "fuel_type": one of "hybrid", "electric", "any". Default based on filters.
|
|
- "zip": ZIP code. Default "98077" (Woodinville WA).
|
|
- "radius": search radius in miles (integer). Default 50.
|
|
- "seller_type": one of "private", "dealer", "any". Default "any".
|
|
- "max_miles": maximum mileage (integer). null if not specified.
|
|
- "search_summary": one-line human-readable summary of what we're searching for.
|
|
|
|
Example: {{"makes": ["Toyota"], "models": ["Prius"], "min_year": 2012, "max_year": 2015, "max_price": 8000, "fuel_type": "hybrid", "zip": "98077", "radius": 50, "seller_type": "any", "max_miles": 100000, "search_summary": "Toyota Prius 2012-2015 under $8K within 50mi"}}"""
|
|
|
|
text = await call_anthropic(prompt, max_tokens=500, timeout=30)
|
|
# Extract JSON from response
|
|
text = text.strip()
|
|
# Try to find JSON object in the response
|
|
match = re.search(r"\{.*\}", text, re.DOTALL)
|
|
if match:
|
|
try:
|
|
return json.loads(match.group())
|
|
except json.JSONDecodeError:
|
|
pass
|
|
# Fallback defaults
|
|
return {
|
|
"makes": [],
|
|
"models": [],
|
|
"min_year": None,
|
|
"max_year": None,
|
|
"max_price": 10000,
|
|
"fuel_type": "hybrid",
|
|
"zip": "98077",
|
|
"radius": 50,
|
|
"seller_type": "any",
|
|
"max_miles": None,
|
|
"search_summary": query,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Build search URLs from parsed parameters
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def build_craigslist_url(params: dict) -> str:
|
|
"""Build Craigslist search URL from params."""
|
|
base = "https://seattle.craigslist.org/search/cta?"
|
|
parts = []
|
|
fuel_map = {"hybrid": "4", "electric": "6", "any": ""}
|
|
fuel = fuel_map.get(params.get("fuel_type", "hybrid"), "")
|
|
if fuel:
|
|
parts.append(f"auto_fuel_type={fuel}")
|
|
if params.get("max_price"):
|
|
parts.append(f"max_price={params['max_price']}")
|
|
parts.append(f"postal={params.get('zip', '98077')}")
|
|
parts.append(f"search_distance={params.get('radius', 50)}")
|
|
if params.get("min_year"):
|
|
parts.append(f"min_auto_year={params['min_year']}")
|
|
if params.get("max_year"):
|
|
parts.append(f"max_auto_year={params['max_year']}")
|
|
if params.get("max_miles"):
|
|
parts.append(f"max_auto_miles={params['max_miles']}")
|
|
if params.get("seller_type") == "private":
|
|
parts.append("purveyor=owner")
|
|
elif params.get("seller_type") == "dealer":
|
|
parts.append("purveyor=dealer")
|
|
# Add make/model as search query
|
|
query_parts = []
|
|
if params.get("makes"):
|
|
query_parts.extend(params["makes"])
|
|
if params.get("models"):
|
|
query_parts.extend(params["models"])
|
|
if query_parts:
|
|
parts.append(f"query={'+'.join(query_parts)}")
|
|
parts.append("sort=date")
|
|
return base + "&".join(parts)
|
|
|
|
|
|
def build_cargurus_url(params: dict) -> str:
|
|
"""Build CarGurus search URL from params."""
|
|
base = (
|
|
"https://www.cargurus.com/Cars/inventorylisting/"
|
|
"viewDetailsFilterViewInventoryListing.action?"
|
|
)
|
|
parts = []
|
|
parts.append(f"zip={params.get('zip', '98077')}")
|
|
if params.get("max_price"):
|
|
parts.append(f"maxPrice={params['max_price']}")
|
|
fuel_map = {"hybrid": "HYBRID", "electric": "ELECTRIC", "any": ""}
|
|
fuel = fuel_map.get(params.get("fuel_type", "hybrid"), "")
|
|
if fuel:
|
|
parts.append(f"fuelTypes={fuel}")
|
|
parts.append(f"distance={params.get('radius', 50)}")
|
|
if params.get("max_miles"):
|
|
parts.append(f"maxMileage={params['max_miles']}")
|
|
if params.get("min_year"):
|
|
parts.append(f"startYear={params['min_year']}")
|
|
if params.get("max_year"):
|
|
parts.append(f"endYear={params['max_year']}")
|
|
parts.append("sortDir=ASC&sortType=PRICE")
|
|
return base + "&".join(parts)
|
|
|
|
|
|
def build_autotempest_url(params: dict) -> str:
|
|
"""Build AutoTempest search URL from params."""
|
|
base = "https://www.autotempest.com/results?"
|
|
parts = []
|
|
parts.append(f"zip={params.get('zip', '98077')}")
|
|
if params.get("max_price"):
|
|
parts.append(f"maxprice={params['max_price']}")
|
|
fuel_map = {"hybrid": "hybrid", "electric": "electric", "any": ""}
|
|
fuel = fuel_map.get(params.get("fuel_type", "hybrid"), "")
|
|
if fuel:
|
|
parts.append(f"fuel={fuel}")
|
|
parts.append(f"radius={params.get('radius', 50)}")
|
|
if params.get("makes"):
|
|
parts.append(f"make={params['makes'][0].lower()}")
|
|
if params.get("models"):
|
|
parts.append(f"model={params['models'][0].lower()}")
|
|
if params.get("min_year"):
|
|
parts.append(f"minyear={params['min_year']}")
|
|
if params.get("max_year"):
|
|
parts.append(f"maxyear={params['max_year']}")
|
|
if params.get("max_miles"):
|
|
parts.append(f"maxmiles={params['max_miles']}")
|
|
return base + "&".join(parts)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scoring engine
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def score_listing(listing: dict, params: dict) -> int:
|
|
"""Score a single listing 0-100."""
|
|
s = 50
|
|
title = (listing.get("title", "") + " " + listing.get("url", "")).lower()
|
|
|
|
# Model bonuses
|
|
if "prius" in title:
|
|
s += 30
|
|
elif "camry hybrid" in title or "camry" in title:
|
|
s += 25
|
|
elif "insight" in title:
|
|
s += 20
|
|
elif "ioniq" in title:
|
|
s += 20
|
|
elif "civic hybrid" in title:
|
|
s += 15
|
|
elif "corolla" in title:
|
|
s += 18
|
|
elif "rav4" in title:
|
|
s += 15
|
|
|
|
# Brand bonuses
|
|
if "toyota" in title:
|
|
s += 10
|
|
elif "honda" in title:
|
|
s += 7
|
|
elif "hyundai" in title:
|
|
s += 5
|
|
elif "kia" in title:
|
|
s += 4
|
|
|
|
# Penalties
|
|
if "nissan" in title and "leaf" in title:
|
|
s -= 25
|
|
if "altima" in title:
|
|
s -= 20
|
|
if any(w in title for w in ["salvage", "rebuilt", "flood", "junk", "parts only"]):
|
|
s -= 20
|
|
|
|
# Price parsing
|
|
price_str = listing.get("price", "")
|
|
price_num = 0
|
|
m = re.search(r"\$?([\d,]+)", price_str)
|
|
if m:
|
|
price_num = int(m.group(1).replace(",", ""))
|
|
if 0 < price_num < 6000:
|
|
s += 10
|
|
elif 6000 <= price_num <= 8000:
|
|
s += 5
|
|
elif price_num > 15000:
|
|
s -= 10
|
|
|
|
# Positive signals
|
|
if "one owner" in title or "single owner" in title:
|
|
s += 8
|
|
if "clean title" in title:
|
|
s += 8
|
|
if "low miles" in title or "low mileage" in title:
|
|
s += 5
|
|
|
|
# If user searched for specific makes/models, boost matches
|
|
for make in params.get("makes", []):
|
|
if make.lower() in title:
|
|
s += 5
|
|
for model in params.get("models", []):
|
|
if model.lower() in title:
|
|
s += 5
|
|
|
|
return max(0, min(100, s))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Generate guide via LLM
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def generate_guide(query: str, listings: list[dict], params: dict) -> dict:
|
|
"""Call the Anthropic API to generate a custom buying guide."""
|
|
# Prepare top 20 listings summary
|
|
top_listings = listings[:20]
|
|
listings_text = "\n".join(
|
|
f"- {item.get('title', 'Unknown')} | {item.get('price', 'N/A')} | Score: {item.get('score', '?')} | {item.get('url', '')}"
|
|
for item in top_listings
|
|
)
|
|
if not listings_text:
|
|
listings_text = "(No listings found)"
|
|
|
|
prompt = f"""You are a car buying expert helping someone near Woodinville, WA (ZIP 98077).
|
|
|
|
The user searched for: "{query}"
|
|
Search parameters: {json.dumps(params, default=str)}
|
|
|
|
Here are the top listings found:
|
|
{listings_text}
|
|
|
|
Generate a comprehensive buying guide tailored to this specific search. Return ONLY valid JSON (no markdown fences, no explanation before/after) with these keys:
|
|
|
|
1. "top_picks": Analysis of the best listings found. Which ones look like the best deals and why? Reference specific listings by name/price. If no listings found, give general advice for this type of car. Use markdown formatting (bold, bullet points).
|
|
|
|
2. "checklist": What to inspect when looking at THESE specific car models. Be specific to the makes/models found (e.g., if mostly Prius results, talk about hybrid battery health, catalytic converter theft, etc.). Use markdown formatting.
|
|
|
|
3. "mechanics": Recommend 3-4 mechanics near Woodinville WA 98077 that would be good for pre-purchase inspections of these types of cars. Include name, approximate location, and why they're good. Use markdown formatting.
|
|
|
|
4. "negotiation": Negotiation tips specific to this price range and car type. Include what to look up beforehand (KBB, etc.), how much below asking to start, WA-specific rules (doc fees, etc.). Use markdown formatting.
|
|
|
|
5. "budget": Total budget breakdown including the car, registration, insurance, maintenance reserves. Be specific to the price range searched. Use markdown formatting.
|
|
|
|
Each value should be a string with markdown formatting for rich display. Be practical, specific, and actionable."""
|
|
|
|
text = await call_anthropic(prompt, max_tokens=4000, timeout=90)
|
|
text = text.strip()
|
|
|
|
# Try to parse JSON from response
|
|
match = re.search(r"\{.*\}", text, re.DOTALL)
|
|
if match:
|
|
try:
|
|
guide = json.loads(match.group())
|
|
# Ensure all keys exist
|
|
for key in ["top_picks", "checklist", "mechanics", "negotiation", "budget"]:
|
|
if key not in guide:
|
|
guide[key] = "No information available."
|
|
return guide
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
# Fallback: return the raw text in top_picks
|
|
return {
|
|
"top_picks": text or "Guide generation failed. Please try again.",
|
|
"checklist": "Run a search to generate a custom checklist.",
|
|
"mechanics": "Run a search to get mechanic recommendations.",
|
|
"negotiation": "Run a search to get negotiation tips.",
|
|
"budget": "Run a search to get a budget breakdown.",
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Static files + index
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def index():
|
|
html = Path("static/index.html").read_text()
|
|
return HTMLResponse(html)
|
|
|
|
|
|
app.mount("/static", StaticFiles(directory="static"), name="static")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# API routes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@app.post("/api/search")
|
|
async def search(request: Request):
|
|
"""Main search endpoint: parse query, crawl sites, score, generate guide."""
|
|
body = await request.json()
|
|
query = body.get("query", "").strip()
|
|
filters = body.get("filters", [])
|
|
|
|
if not query:
|
|
return JSONResponse({"error": "Please enter a search query"}, 400)
|
|
|
|
try:
|
|
# Step 1: Parse query into search parameters
|
|
params = await parse_query(query, filters)
|
|
search_summary = params.get("search_summary", query)
|
|
|
|
# Step 2: Build search URLs
|
|
cl_url = build_craigslist_url(params)
|
|
cg_url = build_cargurus_url(params)
|
|
at_url = build_autotempest_url(params)
|
|
|
|
# Step 3: Run Playwright crawl via search.py subprocess
|
|
search_args = json.dumps(
|
|
{
|
|
"craigslist_url": cl_url,
|
|
"cargurus_url": cg_url,
|
|
"autotempest_url": at_url,
|
|
"params": params,
|
|
}
|
|
)
|
|
|
|
proc = await asyncio.create_subprocess_exec(
|
|
"uv",
|
|
"run",
|
|
"python",
|
|
"search.py",
|
|
"--dynamic",
|
|
search_args,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
env={
|
|
"PATH": "/usr/local/bin:/usr/bin:/bin",
|
|
"HOME": "/root",
|
|
"DISPLAY": ":99",
|
|
},
|
|
)
|
|
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=180)
|
|
|
|
# Step 4: Load results
|
|
result_files = sorted(RESULTS_DIR.glob("search_*.json"))
|
|
all_listings = []
|
|
if result_files:
|
|
data = json.loads(result_files[-1].read_text())
|
|
all_listings = data.get("listings", [])
|
|
|
|
# Step 5: Score and rank
|
|
for listing in all_listings:
|
|
listing["score"] = score_listing(listing, params)
|
|
all_listings.sort(key=lambda x: x["score"], reverse=True)
|
|
|
|
# Step 6: Generate guide via LLM
|
|
guide = await generate_guide(query, all_listings, params)
|
|
|
|
return JSONResponse(
|
|
{
|
|
"status": "ok",
|
|
"search_summary": search_summary,
|
|
"total": len(all_listings),
|
|
"listings": all_listings,
|
|
"guide": guide,
|
|
"urls_searched": {
|
|
"craigslist": cl_url,
|
|
"cargurus": cg_url,
|
|
"autotempest": at_url,
|
|
},
|
|
}
|
|
)
|
|
|
|
except asyncio.TimeoutError:
|
|
return JSONResponse({"error": "Search timed out after 3 minutes"}, 504)
|
|
except Exception as e:
|
|
import traceback
|
|
|
|
return JSONResponse(
|
|
{
|
|
"error": str(e),
|
|
"traceback": traceback.format_exc(),
|
|
},
|
|
500,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Keep existing endpoints
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@app.get("/api/results")
|
|
async def list_results():
|
|
"""List all saved search results."""
|
|
RESULTS_DIR.mkdir(exist_ok=True)
|
|
files = sorted(RESULTS_DIR.glob("search_*.json"), reverse=True)
|
|
results = []
|
|
for f in files[:20]:
|
|
try:
|
|
data = json.loads(f.read_text())
|
|
results.append(
|
|
{
|
|
"filename": f.name,
|
|
"date": data.get("search_date", ""),
|
|
"total": data.get("total_listings", 0),
|
|
}
|
|
)
|
|
except Exception:
|
|
continue
|
|
return JSONResponse(results)
|
|
|
|
|
|
@app.get("/api/results/{filename}")
|
|
async def get_result(filename: str):
|
|
"""Get a specific result file."""
|
|
path = RESULTS_DIR / filename
|
|
if not path.exists():
|
|
return JSONResponse({"error": "not found"}, 404)
|
|
return JSONResponse(json.loads(path.read_text()))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# noVNC proxy (LAST -- catch-all)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
NOVNC_PROXY_PATHS = re.compile(
|
|
r"^/(vnc\.html|vnc_lite\.html|websockify|core/|vendor/|app/|include/|noVNC)"
|
|
)
|
|
|
|
|
|
@app.api_route(
|
|
"/{path:path}",
|
|
methods=["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD", "OPTIONS"],
|
|
include_in_schema=False,
|
|
)
|
|
async def catch_all_novnc_proxy(request: Request, path: str):
|
|
"""Proxy noVNC assets and websockify from port 6080."""
|
|
if not NOVNC_PROXY_PATHS.match(f"/{path}"):
|
|
return JSONResponse({"error": "not found"}, status_code=404)
|
|
|
|
url = f"{NOVNC_UPSTREAM}/{path}"
|
|
if request.url.query:
|
|
url += f"?{request.url.query}"
|
|
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
resp = await client.request(
|
|
method=request.method,
|
|
url=url,
|
|
headers={
|
|
k: v
|
|
for k, v in request.headers.items()
|
|
if k.lower() not in ("host", "connection")
|
|
},
|
|
content=await request.body(),
|
|
)
|
|
headers = {}
|
|
if "content-type" in resp.headers:
|
|
headers["content-type"] = resp.headers["content-type"]
|
|
return StreamingResponse(
|
|
iter([resp.content]),
|
|
status_code=resp.status_code,
|
|
headers=headers,
|
|
)
|