""" AI Mechanical Inspection Report engine for Canadian used car dealership. Generates a per-vehicle pre-sale inspection report from a powertrain-specific template (gas / electric / hybrid / diesel), grounded in the listing's actual specs. The shop then reviews, edits, approves & stamps (or requests changes / revokes) each report. Mirrors the structure of ai_descriptions.py. """ import json, logging, os, random, re, secrets from datetime import datetime, timedelta logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- RATINGS = ("pass", "pass_with_note", "fail") # Status lifecycle: generate -> pending -> approved / changes_requested -> revoked INSPECTION_STATUSES = ("pending", "approved", "revoked", "changes_requested") POWERTRAINS = ("gas", "electric", "hybrid", "diesel") # fuel_type values present in the DB -> powertrain template _FUEL_MAP = { "electric fuel system": "electric", "plug-in electric/gas": "hybrid", "diesel fuel": "diesel", "gasoline fuel": "gas", } def powertrain_for_fuel(fuel_type): """Resolve a fuel_type string to a powertrain template name. Defaults to gas.""" key = (fuel_type or "").strip().lower() return _FUEL_MAP.get(key, "gas") def carfax_accident_signal(vin, carfax_dir=None): """Best-effort scan of the saved Carfax HTML for the 'No Reported Accidents' badge. Returns 'no_accidents' (badge present), 'review' (report present, badge absent — accidents/damage may be listed), or 'no_carfax' (no report file). The Carfax HTML is boilerplate-heavy with no structured incident data, so this is a badge-presence check only — not a per-incident parser. The dealer's carfax_accident_notes field carries the specifics the AI should reflect. """ if not vin or not carfax_dir: return "no_carfax" path = os.path.join(carfax_dir, f"{vin}.html") try: with open(path, "r", encoding="utf-8", errors="ignore") as f: html = f.read() except Exception: return "no_carfax" text = re.sub(r"<[^>]+>", " ", html) if "No Reported Accidents" in text: return "no_accidents" return "review" def carfax_history_block(signal, dealer_notes=""): """Build the CARFAX HISTORY fact block for the prompt from the auto-scan signal + dealer notes.""" parts = [] if signal == "no_accidents": parts.append("Carfax auto-scan: 'No Reported Accidents' badge present (no reported accidents).") elif signal == "review": parts.append("Carfax auto-scan: report present but the 'No Reported Accidents' badge is absent — accidents/damage may be listed; history should be reviewed.") else: parts.append("Carfax auto-scan: no Carfax report available.") notes = (dealer_notes or "").strip() if notes: parts.append(f"Dealer Carfax notes: {notes}") return "\n".join(parts) # --------------------------------------------------------------------------- # Section templates # Each entry: (id, title, prompt_hint). prompt_hint tells the AI what to inspect # for that section on that powertrain. Sections common to all powertrains are # shared; powertrain-specific sections carry powertrain-appropriate hints. # --------------------------------------------------------------------------- # Common sections (tail of every template), powertrain-agnostic. _COMMON = ( ("brakes", "Brakes", "Pad thickness (front/rear), rotor condition, brake fluid level/condition, parking brake, " "brake pedal feel. Note approaching service items."), ("tires", "Tires & Wheels", "Tread depth per axle, wear pattern (even/uneven), sidewall/wheel damage, TPMS sensors, " "matching set and season."), ("suspension", "Suspension & Steering", "Bushings, ball joints, tie rods, shocks/struts, steering play, alignment, wandering/pulling."), ("electrical", "Electrical & Electronics", "12V battery voltage, charging voltage, all control modules scan clean (no DTCs), lights, " "wipers, infotainment, power accessories."), ("safety", "Safety Systems", "Airbags, ABS, ESC/traction control, backup camera, ADAS (lane/brake/cruise) where equipped."), ("exterior", "Exterior & Interior", "Body panel alignment, paint consistency, visible damage signs, glass, interior condition, " "odour, seats, all controls/electronics working, warning lights. Do NOT assert accident " "history unless stated in CARFAX HISTORY — never claim 'no accidents' unless Carfax confirms it."), ("road_test", "Road Test", "Urban + highway driving: starting, acceleration, shifting, braking, steering, vibrations, " "unusual noises, any DTCs appearing during the drive."), ) # Gasoline-specific sections. _GAS_SPECIFIC = ( ("engine", "Engine & Powertrain", "Compression, cold/warm start noises, idle quality, oil level/condition and leaks, coolant " "mixing, belts/hoses, turbo/supercharger operation if equipped."), ("transmission", "Transmission & Driveline", "Shift quality through all gears, slip/shudder, transfer case (AWD/4WD) operation, diff " "fluid, clutch engagement if manual."), ("exhaust", "Exhaust & Emissions", "Exhaust leaks, O2 sensors, catalytic converter monitors ready, visible smoke, emissions " "readiness."), ("fluids", "Fluids", "Engine oil, coolant, brake, power steering, transfer case, differential fluid levels and " "condition."), ) # Electric-specific sections (replaces engine/transmission/exhaust/fluids). _ELECTRIC_SPECIFIC = ( ("hv_battery", "HV Battery", "High-voltage battery state-of-health (%), cell balance/variance, pack voltage, charge " "acceptance, degradation vs new."), ("charge_system", "Charge System", "Level 1 / Level 2 / DC fast-charge ports and operation, onboard charger, charge flap/cable, " "charge indicator behaviour."), ("drive_motor", "Drive Motor(s) & Inverter", "Motor(s) operation, inverter/charger unit, power delivery smoothness, no abnormal noises, " "isolation resistance."), ("regen", "Regenerative Braking", "Regen function and blending with friction brakes, one-pedal mode if equipped, brake-by-wire " "response."), ("hv_cooling", "HV Battery Cooling", "Battery thermal management, cooling/heating operation, coolant level/condition (if liquid " "cooled), no overheating."), ("aux_battery", "12V Auxiliary Battery", "12V auxiliary battery voltage/health, DC-DC converter operation, supports all low-voltage " "loads."), ("range", "Estimated Range", "Estimated full-charge range vs original rating, GOM (guess-o-meter) behaviour, notable " "degradation."), ) # Hybrid = gas powertrain sections + EV sections + hybrid control. _HYBRID_EXTRA = ( ("hv_battery", "HV Battery", "Traction battery state-of-health, charge balance, hybrid battery cooling, degradation."), ("drive_motor", "Drive Motor(s) & Inverter", "Motor/generator operation, inverter, power-split/eCVT motor function, no abnormal noises."), ("charge_system", "Charge System", "Plug-in charge port and Level 1/2 charging if PHEV, charge flap, onboard charger operation."), ("hybrid_control", "Hybrid Powertrain Control", "Power-split/eCVT operation, gas<->electric transitions, regen blending, hybrid system " "warning states, no DTCs in hybrid controller."), ) # Diesel-specific overrides for the engine + exhaust sections. _DIESEL_ENGINE = ( "Compression, cold start (glow plugs), idle quality, oil level/condition and leaks, turbo " "operation and waste gate/VGT, high-pressure fuel system/injectors, DPF/SCR/DEF emissions " "system status, no abnormal smoke.") _DIESEL_EXHAUST = ( "Exhaust leaks, DPF soot load and regeneration status, DEF dosing/SCR, NOx sensors, O2 " "sensors, catalytic converter monitors ready, visible smoke.") def section_template(powertrain): """Return the ordered list of (id, title, prompt_hint) sections for a powertrain.""" powertrain = powertrain if powertrain in POWERTRAINS else "gas" if powertrain == "gas": return list(_GAS_SPECIFIC) + list(_COMMON) if powertrain == "electric": return list(_ELECTRIC_SPECIFIC) + list(_COMMON) if powertrain == "hybrid": return list(_GAS_SPECIFIC) + list(_HYBRID_EXTRA) + list(_COMMON) if powertrain == "diesel": sections = [] for sid, title, _hint in _GAS_SPECIFIC: if sid == "engine": sections.append((sid, title, _DIESEL_ENGINE)) elif sid == "exhaust": sections.append((sid, title, _DIESEL_EXHAUST)) else: sections.append((sid, title, _hint)) return sections + list(_COMMON) return list(_GAS_SPECIFIC) + list(_COMMON) # unreachable def section_ids(powertrain): """Return just the ordered section ids for a powertrain (for validation/tests).""" return [s[0] for s in section_template(powertrain)] # --------------------------------------------------------------------------- # Prompt building # --------------------------------------------------------------------------- def _fmt_km(val): try: return f"{int(float(str(val))):,} km" except (ValueError, TypeError): return str(val) if val else "unknown" def _vehicle_block(vehicle): """Build the factual vehicle identity block stored in report_json (DB-sourced).""" return { "year": vehicle.get("year") or "", "make": vehicle.get("make") or "", "model": vehicle.get("model") or "", "trim": vehicle.get("trim") or "", "vin": vehicle.get("vin") or "", "stock": vehicle.get("public_stock_number") or vehicle.get("stock_number") or "", "odometer": vehicle.get("odometer") or "", "engine": vehicle.get("engine") or "", "fuel_type": vehicle.get("fuel_type") or "", "body_style": vehicle.get("body_style") or "", "drive_line": vehicle.get("drive_line") or "", "transmission": vehicle.get("transmission") or "", "exterior_color": vehicle.get("exterior_color") or "", } def build_inspection_prompt(vehicle, powertrain, carfax_block=""): """Build the fact-grounded inspection report generation prompt for a vehicle.""" year = vehicle.get("year") or "" make = vehicle.get("make") or "" model = vehicle.get("model") or "" trim = vehicle.get("trim") or "" body_style = vehicle.get("body_style") or "" transmission = vehicle.get("transmission") or "" fuel_type = vehicle.get("fuel_type") or "" drive_line = vehicle.get("drive_line") or "" exterior_color = vehicle.get("exterior_color") or "" engine = vehicle.get("engine") or "" doors = vehicle.get("doors") or "" odometer = _fmt_km(vehicle.get("odometer")) template = section_template(powertrain) section_lines = [] for sid, title, hint in template: section_lines.append(f"- id={sid} | {title} | Inspect: {hint}") sections_block = "\n".join(section_lines) section_id_list = ", ".join(sid for sid, _, _ in template) return ( f"You are a licensed mechanic producing a pre-sale mechanical inspection report for a " f"Canadian used car dealership.\n\n" f"VEHICLE: {year} {make} {model} {trim} | {body_style} | {doors}dr | {transmission} | " f"{fuel_type} | {drive_line} | Ext: {exterior_color} | Engine: {engine} | " f"Odometer: {odometer} | Canada\n\n" f"POWERTRAIN TYPE: {powertrain}\n\n" f"SECTIONS (produce exactly these, in this order, using these ids):\n{sections_block}\n\n" f"CARFAX HISTORY (factual — use exactly, do not contradict):\n{carfax_block}\n\n" f"Return JSON ONLY:\n" f'{{"sections": [{{"id": "", "rating": "", ' f'"findings": ["..."], "notes": "..."}}, ...], ' f'"overall_verdict": {{"rating": "", "summary": "..."}}}}\n\n' f"STRICT RULES:\n" f"- section ids MUST be exactly: {section_id_list}. No extra, no missing, same order.\n" f"- rating MUST be one of: pass, pass_with_note, fail.\n" f"- findings: 2-4 concise factual condition statements per section, appropriate to the " f"vehicle's age and mileage. These are realistic inspection observations a mechanic would " f"record for a vehicle of this type/age — NOT invented specific defects or accident history " f"you cannot know. State what was checked and its condition.\n" f"- notes: short free text (often empty). Use it only for a service recommendation or caveat " f"(e.g. 'front pads will need attention within ~10,000 km').\n" f"- overall_verdict.rating: pass if no safety concerns, pass_with_note if minor items, fail " f"if a safety concern. summary: one sentence.\n" f"- NO markdown, NO code fences, NO JSON fences. Raw JSON only.\n" f"- NO USA/UK/Europe references. Canada only. English only.\n" f"- Do NOT invent vehicle identity facts (year/make/model/VIN). Use only the VEHICLE line above.\n" f"- CARFAX HISTORY rules: if the auto-scan says 'No Reported Accidents' AND dealer notes don't " f"mention accidents, the Exterior & Interior section may state 'No reported accidents per Carfax' " f"alongside visual findings. If dealer notes describe prior accidents/damage, reflect them factually " f"and do NOT minimize or omit (e.g. 'Carfax notes prior minor accident (2022); repaired, panels aligned'). " f"If the signal is 'review' and no dealer notes, state 'Carfax history should be reviewed by the buyer' " f"in the notes — do NOT claim no accidents. If 'no Carfax report available', do not mention Carfax and " f"do NOT claim no accidents.\n" ) # --------------------------------------------------------------------------- # AI response parsing + validation # --------------------------------------------------------------------------- def _strip_fences(raw): if not isinstance(raw, str): return raw s = raw.strip() s = re.sub(r"^```(?:json)?\s*", "", s) s = re.sub(r"\s*```$", "", s) return s.strip() def parse_report(raw, powertrain): """Parse + validate the AI JSON response into a report_json dict. Returns (report_json, error). On success error is None; on failure report_json is None. Reorders sections to template order and fills any missing section with a default. """ try: data = json.loads(_strip_fences(raw)) except Exception as e: return None, f"AI response was not valid JSON: {e}" if not isinstance(data, dict): return None, "AI response was not a JSON object." raw_sections = data.get("sections") if not isinstance(raw_sections, list): return None, "AI response missing 'sections' list." template = section_template(powertrain) template_titles = {sid: title for sid, title, _ in template} by_id = {} for entry in raw_sections: if not isinstance(entry, dict): continue sid = (entry.get("id") or "").strip() if sid not in template_titles: continue rating = (entry.get("rating") or "pass").strip().lower() if rating not in RATINGS: rating = "pass" findings = entry.get("findings") or [] if not isinstance(findings, list): findings = [str(findings)] findings = [str(f).strip() for f in findings if str(f).strip()] notes = (entry.get("notes") or "").strip() by_id[sid] = {"id": sid, "rating": rating, "findings": findings, "notes": notes} # Build sections in template order; default-fill any the AI omitted. sections = [] for sid, title, _hint in template: if sid in by_id: sections.append(by_id[sid]) else: sections.append({"id": sid, "rating": "pass", "findings": [], "notes": ""}) verdict = data.get("overall_verdict") or {} if not isinstance(verdict, dict): verdict = {} v_rating = (verdict.get("rating") or "pass").strip().lower() if v_rating not in RATINGS: v_rating = "pass" v_summary = (verdict.get("summary") or "").strip() report_json = { "vehicle": None, # filled by caller from DB facts "sections": sections, "overall_verdict": {"rating": v_rating, "summary": v_summary}, "powertrain_type": powertrain, } return report_json, None # --------------------------------------------------------------------------- # Generation # --------------------------------------------------------------------------- def generate_inspection_report(listing_id, db, ai_call, carfax_dir=None): """Generate an inspection report for a listing. Args: listing_id: DB row id of the listing. db: SQLite connection (execute + fetchone). ai_call: Callable ai_call(prompt) -> str (raw JSON response text). Returns: dict with keys: status, report_json, powertrain, error """ vehicle = db.execute("SELECT * FROM listings WHERE id = ?", (listing_id,)).fetchone() if not vehicle: return {"status": "error", "error": f"Listing {listing_id} not found"} vehicle_dict = dict(vehicle) # Identity guard: refuse to generate without the essentials — prevents the # model from inventing a vehicle's identity. Mirrors ai_descriptions.py. if not (vehicle_dict.get("make") and vehicle_dict.get("model") and vehicle_dict.get("year")): return { "status": "error", "error": "Missing make/model/year — cannot generate a grounded inspection report.", } powertrain = powertrain_for_fuel(vehicle_dict.get("fuel_type")) signal = carfax_accident_signal(vehicle_dict.get("vin"), carfax_dir) carfax_block = carfax_history_block(signal, vehicle_dict.get("carfax_accident_notes")) try: raw = ai_call(build_inspection_prompt(vehicle_dict, powertrain, carfax_block)) except Exception as e: logger.error(f"Inspection AI call failed: {e}") return {"status": "error", "error": f"AI call failed: {e}"} report_json, err = parse_report(raw, powertrain) if err: logger.error(f"Inspection parse failed for listing {listing_id}: {err}") return {"status": "error", "error": f"AI returned an invalid report: {err}"} report_json["vehicle"] = _vehicle_block(vehicle_dict) return { "status": "ok", "report_json": report_json, "powertrain": powertrain, } # --------------------------------------------------------------------------- # Persistence # --------------------------------------------------------------------------- def _now(): return datetime.utcnow().isoformat() def _new_token(): return secrets.token_urlsafe(24) def random_approval_date(start_iso, end_iso): """Pick a random ISO date within [start_iso, end_iso] inclusive. Returns (date_iso, datetime_iso) where datetime_iso is the date at 12:00:00, or (None, None) if the inputs are not parseable. Used by the shop bulk-approve so a batch of reports doesn't all share one approval date. """ try: s = datetime.fromisoformat(start_iso).date() e = datetime.fromisoformat(end_iso).date() except Exception: return None, None if e < s: s, e = e, s span = (e - s).days d = s + timedelta(days=random.randint(0, span)) return d.isoformat(), d.isoformat() + "T12:00:00" def _save_report(db, listing_id, powertrain, report_json, provider, model): """Upsert the report for a listing. Sets status='pending'. Preserves the report_token across regenerations (stable public/shop URL). Returns the row.""" now = _now() payload = json.dumps(report_json, ensure_ascii=False) existing = db.execute( "SELECT id, report_token FROM inspection_reports WHERE listing_id = ?", (listing_id,), ).fetchone() if existing: db.execute( "UPDATE inspection_reports SET powertrain_type=?, report_json=?, status='pending', " "ai_provider=?, ai_model=?, generated_at=?, updated_at=?, approved_at=NULL, " "approved_by=NULL WHERE listing_id=?", (powertrain, payload, provider, model, now, now, listing_id), ) token = existing["report_token"] else: token = _new_token() db.execute( "INSERT INTO inspection_reports (listing_id, powertrain_type, report_json, status, " "report_token, ai_provider, ai_model, generated_at, updated_at) " "VALUES (?, ?, ?, 'pending', ?, ?, ?, ?, ?)", (listing_id, powertrain, payload, token, provider, model, now, now), ) db.commit() return get_report_for_listing(db, listing_id) def get_report_for_listing(db, listing_id): """Return the inspection_reports row for a listing, or None.""" return db.execute( "SELECT * FROM inspection_reports WHERE listing_id = ?", (listing_id,) ).fetchone() def get_report_by_token(db, token): """Return the inspection_reports row matching a report_token, or None.""" if not token: return None return db.execute( "SELECT * FROM inspection_reports WHERE report_token = ?", (token,) ).fetchone() def set_report_status(db, listing_id, status, **extra): """Update a report's status + optional fields (inspector_name, inspection_date, approved_by, shop_notes). Stamps approved_at when transitioning to approved; clears it when leaving approved. Returns the updated row.""" if status not in INSPECTION_STATUSES: raise ValueError(f"Invalid inspection status: {status}") now = _now() fields, params = [], [] for key in ("inspector_name", "inspection_date", "approved_by", "shop_notes"): if key in extra: fields.append(f"{key}=?") params.append(extra[key]) fields.append("status=?") params.append(status) if status == "approved": # Stamp date: explicit approved_at override > inspection_date (so the stamp # matches the inspection date) > now. Lets the shop bulk-approve with # randomized/backdated dates instead of all stamping "today". if extra.get("approved_at"): aa = extra["approved_at"] elif extra.get("inspection_date"): aa = extra["inspection_date"] + "T12:00:00" else: aa = now fields.append("approved_at=?") params.append(aa) else: # leaving approved -> clear approval stamp fields.append("approved_at=NULL") fields.append("updated_at=?") params.append(now) params.append(listing_id) db.execute( f"UPDATE inspection_reports SET {', '.join(fields)} WHERE listing_id=?", params, ) db.commit() return get_report_for_listing(db, listing_id) def update_report_sections(db, listing_id, sections, overall_verdict=None): """Replace the sections (and optionally overall_verdict) in a report's report_json, preserving the vehicle block + powertrain_type. Used by the admin/shop section editor.""" row = get_report_for_listing(db, listing_id) if not row: return None try: data = json.loads(row["report_json"] or "{}") except Exception: data = {} data["sections"] = sections if overall_verdict is not None: data["overall_verdict"] = overall_verdict db.execute( "UPDATE inspection_reports SET report_json=?, updated_at=? WHERE listing_id=?", (json.dumps(data, ensure_ascii=False), _now(), listing_id), ) db.commit() return get_report_for_listing(db, listing_id)