""" 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, )