from fastapi import FastAPI, Request, UploadFile, File, HTTPException, Header, Depends, Query, Form, Body, status from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from starlette.middleware.sessions import SessionMiddleware from typing import Optional from datetime import date, timedelta from uuid import UUID from app.db import ( fetch_monthly_summary, fetch_retention_pen, fetch_campaigns, fetch_advisors, fetch_template, defer_conversation, resume_conversation, fetch_conversation_status, insert_message, fetch_messages_for_conversation, escalate_conversation, get_conversation_with_messages, list_conversations_for_customer, get_conversation_summary ) from app.config import API_KEY, INGEST_FUNCTION_URL, INGEST_FUNCTION_TOKEN, LOYALTY_DASHBOARD_PASSWORD, SESSION_SECRET_KEY from app.api.monthly_summary import router as monthly_summary_router from app.lia_inbound import LiaInbound, LiaReply, process_inbound_message from app.lia_outbound import OutboundRequest, OutboundResponse, process_outbound_message, CampaignNotActiveError from app.lia_rules import ( LIA_TIER_BEHAVIOUR, LIA_TONE_GUIDELINES, classify_inbound_tier, select_eot_intro, LIA_NUDGE_TEMPLATES, LIA_EOT_12_NUDGES, LIA_EOT_9_NUDGES, LIA_EOT_6_NUDGES, LIA_EOT_3_NUDGES, ) import httpx def require_api_key(x_api_key: str | None = Header(default=None)): if API_KEY and x_api_key != API_KEY: raise HTTPException(status_code=401, detail="Unauthorised") return True def require_auth(request: Request): """Dependency to check if user is authenticated via session""" if not request.session.get("authenticated"): raise HTTPException(status_code=401, detail="Not authenticated") return True app = FastAPI() @app.exception_handler(HTTPException) async def custom_http_exception_handler(request: Request, exc: HTTPException): """Redirect to login page for browser requests when unauthorized, return JSON for API requests""" if exc.status_code == 401: # Always return JSON for /login endpoint so error messages work if request.url.path == "/login": return JSONResponse( status_code=exc.status_code, content={"detail": exc.detail} ) # For other routes, redirect browser requests to login is_api_request = ( request.url.path.startswith("/api/") or "application/json" in request.headers.get("accept", "") ) if not is_api_request: return RedirectResponse(url="/login", status_code=302) return JSONResponse( status_code=exc.status_code, content={"detail": exc.detail} ) import os # In production/HTTPS environments, use https_only=True for security # In development (Replit preview), use https_only=False so sessions work IS_PRODUCTION = os.getenv("REPL_DEPLOYMENT") == "1" app.add_middleware( SessionMiddleware, secret_key=SESSION_SECRET_KEY, session_cookie="dp_session", max_age=7 * 24 * 60 * 60, same_site="lax", # "lax" works better with redirects https_only=IS_PRODUCTION, # Only enforce HTTPS in production ) app.mount("/static", StaticFiles(directory="app/static"), name="static") templates = Jinja2Templates(directory="app/templates") app.include_router(monthly_summary_router) @app.get("/login", response_class=HTMLResponse) async def login_page(request: Request): """Show login page if not authenticated, redirect to dashboard if already logged in""" if request.session.get("authenticated"): return RedirectResponse(url="/", status_code=302) return templates.TemplateResponse("login.html", {"request": request}) @app.post("/login", response_class=HTMLResponse) async def login(request: Request, password: str = Form(...)): """Verify password and create session""" if not LOYALTY_DASHBOARD_PASSWORD: raise HTTPException(status_code=500, detail="Dashboard password not configured") if password == LOYALTY_DASHBOARD_PASSWORD: request.session["authenticated"] = True # Use server-side redirect so session cookie is properly set return RedirectResponse(url="/v2", status_code=303) # Return login page with error message return templates.TemplateResponse( "login.html", {"request": request, "error": "Invalid password"}, status_code=401 ) @app.get("/v2", response_class=HTMLResponse, dependencies=[Depends(require_auth)]) async def dashboard_v2(request: Request): """New Exsto-style dashboard with dark sidebar""" return templates.TemplateResponse("dashboard_v2.html", {"request": request}) @app.get("/v2/uploads", response_class=HTMLResponse, dependencies=[Depends(require_auth)]) async def manager_uploads(request: Request): """Manager data uploads screen (Exsto-style)""" return templates.TemplateResponse("manager_uploads.html", {"request": request}) @app.get("/v2/reports", response_class=HTMLResponse, dependencies=[Depends(require_auth)]) async def reports_page(request: Request): """Exports & Reports page with campaign metrics""" return templates.TemplateResponse("reports.html", {"request": request}) @app.get("/v2/campaigns", response_class=HTMLResponse, dependencies=[Depends(require_auth)]) async def campaigns_page(request: Request): """Campaign Builder page""" return templates.TemplateResponse("campaigns.html", {"request": request}) @app.post("/logout") async def logout(request: Request): """Clear session and log out""" request.session.clear() return RedirectResponse(url="/login", status_code=302) @app.get("/", response_class=HTMLResponse, dependencies=[Depends(require_auth)]) async def dashboard(request: Request): data = await fetch_monthly_summary() latest = data[-1] if data else None return templates.TemplateResponse( "dashboard.html", {"request": request, "latest": latest}, ) @app.get("/api/monthly-summary", dependencies=[Depends(require_auth)]) async def api_monthly_summary( date_from: Optional[str] = Query(None), date_to: Optional[str] = Query(None), advisor: Optional[str] = Query(None), ): data = await fetch_monthly_summary(date_from, date_to, advisor) return {"data": data} @app.get("/api/retention-pen", dependencies=[Depends(require_auth)]) async def api_retention_pen( month: Optional[str] = Query(None), payments_lte: Optional[int] = Query(None), limit: int = Query(100), advisor: Optional[str] = Query(None), ): data = await fetch_retention_pen(month, payments_lte, limit, advisor) return {"data": data} @app.get("/api/advisors", dependencies=[Depends(require_auth)]) async def api_advisors(): data = await fetch_advisors() return {"data": data} @app.get("/api/campaigns", dependencies=[Depends(require_auth)]) async def api_campaigns( status: Optional[str] = Query(None), today: Optional[str] = Query(None), ): """ Return campaigns, optionally filtered by status and today's date. - status: filter by campaign status (draft, running, paused, etc.) - today: YYYY-MM-DD string; only return campaigns active on this date """ today_date = None if today: today_date = date.fromisoformat(today) data = await fetch_campaigns(status=status, today=today_date) return {"data": data} @app.get("/api/reminders", dependencies=[Depends(require_auth)]) async def api_reminders(window: int = 7, advisor: str | None = None): """ Stub endpoint returning demo appointment reminders. 'window' is days from today (e.g. 1 = today, 7 = next 7 days). """ today = date.today() base = [ { "id": "rem_1", "conversation_id": "05decb80-fd14-4d07-856b-52cd5b4ae242", "date": str(today), "time": "10:30", "customer_name": "Alex Smith", "reg": "FY23 ABC", "reason": "End-of-term options review (12 months)", "advisor": "Jamie", "channel": "WhatsApp" }, { "id": "rem_2", "conversation_id": "1a2b3c4d-5e6f-7a8b-9c0d-1e2f3a4b5c6d", "date": str(today + timedelta(days=2)), "time": "14:00", "customer_name": "Jordan Lee", "reg": "FY21 XYZ", "reason": "Service visit – good time to review upgrade options", "advisor": "Jamie", "channel": "SMS" }, { "id": "rem_3", "conversation_id": "9f8e7d6c-5b4a-3210-fedc-ba9876543210", "date": str(today + timedelta(days=5)), "time": "11:15", "customer_name": "Sam Taylor", "reg": "FY20 QWE", "reason": "End-of-term 6 month priority review", "advisor": "Taylor", "channel": "WhatsApp" }, ] # Simple filtering by advisor name if provided items = [r for r in base if not advisor or r["advisor"] == advisor] return {"data": items} @app.get("/api/tco", dependencies=[Depends(require_auth)]) async def api_tco(model: str | None = None): """ Stub endpoint returning demo TCO summaries by model. """ rows = [ { "model": "Q3", "monthly_payment": 499, "insurance_est": 70, "energy_fuel": 110, "service_plan": 25, "tyres_misc": 20, }, { "model": "Q4 e-tron", "monthly_payment": 579, "insurance_est": 75, "energy_fuel": 80, "service_plan": 30, "tyres_misc": 25, }, { "model": "A3", "monthly_payment": 399, "insurance_est": 65, "energy_fuel": 100, "service_plan": 20, "tyres_misc": 18, }, ] if model: rows = [r for r in rows if r["model"] == model] return {"data": rows} # --- Data upload stub (CSV -> storage path placeholder) --- from uuid import uuid4 async def upload_to_supabase_storage(file: UploadFile) -> str: """ Stub implementation: in future this should upload to Supabase Storage. For now we just pretend and return a generated path. """ return f"books/{uuid4()}-{file.filename}" @app.post("/upload-book", dependencies=[Depends(require_auth)]) async def upload_book( file: UploadFile = File(...), bookMonth: str = Form(...) ): """ Receive VWFS snapshot CSV + bookMonth from the dashboard, forward them securely to the Supabase Edge Function. """ if not INGEST_FUNCTION_URL or not INGEST_FUNCTION_TOKEN: raise HTTPException(status_code=500, detail="Ingest function not configured") contents = await file.read() files = { "file": (file.filename or "vwfs_snapshot.csv", contents, file.content_type or "text/csv") } data = { "bookMonth": bookMonth } headers = { "x-ingest-token": INGEST_FUNCTION_TOKEN } try: async with httpx.AsyncClient(timeout=60.0) as client: resp = await client.post( INGEST_FUNCTION_URL, headers=headers, files=files, data=data, ) except httpx.TimeoutException: raise HTTPException( status_code=504, detail="Upload timed out — please try again with a smaller file or contact support" ) except httpx.RequestError as e: raise HTTPException( status_code=500, detail=f"Failed to connect to ingest service: {str(e)}" ) try: payload = resp.json() except Exception: payload = None if resp.status_code >= 400: msg = (payload or {}).get("error") if isinstance(payload, dict) else resp.text raise HTTPException( status_code=resp.status_code, detail=msg or "Ingest function error", ) return {"detail": "Upload complete — Book imported"} # --- Renewals CSV Ingest (Direct to Supabase) --- import csv import io import logging logger = logging.getLogger(__name__) @app.post("/api/renewals/ingest", dependencies=[Depends(require_auth)]) async def renewals_ingest( file: UploadFile = File(...), bookMonth: str = Form(...) ): """ Ingest renewals CSV directly into stg_vwfs_book_raw_csv table. Uses DELETE + INSERT strategy: deletes all existing rows for the book_month, then inserts fresh data. This ensures re-uploads fully replace previous data. CSV → Column Mapping: - CSV headers are mapped 1:1 to table columns (same names) - 'Mobile' in CSV → 'Mobile' column in table - book_month is added from the form field """ from app.supabase_client import supabase logger.info(f"[INGEST] Starting renewals ingest for book month: {bookMonth}") logger.info(f"[INGEST] File: {file.filename}, Content-Type: {file.content_type}") try: contents = await file.read() text = contents.decode('utf-8-sig') reader = csv.DictReader(io.StringIO(text)) rows = list(reader) logger.info(f"[INGEST] Parsed {len(rows)} rows from CSV") if not rows: raise HTTPException(status_code=400, detail="CSV file is empty or has no data rows") csv_columns = list(rows[0].keys()) logger.info(f"[INGEST] CSV columns: {csv_columns}") if 'Mobile' in csv_columns: sample_mobiles = [r.get('Mobile', '') for r in rows[:3]] logger.info(f"[INGEST] Sample Mobile values from CSV: {sample_mobiles}") delete_result = supabase.table('stg_vwfs_book_raw_csv').delete().eq('book_month', bookMonth).execute() deleted_count = len(delete_result.data) if delete_result.data else 0 logger.info(f"[INGEST] Deleted {deleted_count} existing rows for book_month={bookMonth}") inserted_count = 0 errors = [] for i, row in enumerate(rows): try: clean_row = {} for key, value in row.items(): clean_key = key.strip() if key else key if clean_key: clean_value = value.strip() if value else None if clean_value == '': clean_value = None clean_row[clean_key] = clean_value clean_row['book_month'] = bookMonth if i == 0: logger.info(f"[INGEST] First row Mobile value: {clean_row.get('Mobile', 'NOT FOUND')}") result = supabase.table('stg_vwfs_book_raw_csv').insert(clean_row).execute() inserted_count += 1 except Exception as e: error_msg = f"Row {i+1}: {str(e)}" errors.append(error_msg) logger.warning(f"[INGEST] {error_msg}") if len(errors) >= 10: logger.error("[INGEST] Too many errors, stopping") break logger.info(f"[INGEST] Complete: {inserted_count} rows inserted, {len(errors)} errors") if inserted_count == 0: raise HTTPException( status_code=500, detail=f"No rows were inserted. Errors: {'; '.join(errors[:5])}" ) return { "success": True, "inserted": inserted_count, "deleted": deleted_count, "errors": len(errors), "detail": f"Replaced {deleted_count} existing rows with {inserted_count} new records", "error_samples": errors[:3] if errors else [] } except HTTPException: raise except Exception as e: logger.error(f"[INGEST] Fatal error: {str(e)}") raise HTTPException(status_code=500, detail=f"Ingest failed: {str(e)}") # --- Renewals Campaign Export (for Make.com) --- @app.get("/api/renewals/export", dependencies=[Depends(require_api_key)]) async def renewals_export( book_month: str = Query(..., description="Book month in YYYY-MM format, e.g. 2025-12"), campaign: str = Query(None, description="Optional campaign tag for filtering") ): """ Export renewals data for Make.com outbound campaigns. For each customer in dp_renewals_test_view: 1. Upserts into dp_customer using customer_bpid as external_key 2. Returns a list ready for /api/lia/outbound Response format matches what /api/lia/outbound expects: - customer_id: UUID from dp_customer - payments_remaining: for stage selection - to_number: WhatsApp E.164 format """ from app.supabase_client import supabase import uuid from datetime import datetime logger.info(f"[EXPORT] Starting renewals export for book_month={book_month}, campaign={campaign}") try: query = supabase.table('dp_renewals_test_view').select('*').eq('book_month', book_month) if campaign: query = query.eq('campaign', campaign) logger.info(f"[EXPORT] Filtering by campaign={campaign}") result = query.execute() if not result.data: logger.info(f"[EXPORT] No records found for book_month={book_month}") return {"success": True, "count": 0, "customers": []} logger.info(f"[EXPORT] Found {len(result.data)} records in view") export_list = [] for row in result.data: customer_bpid = row.get('customer_bpid') first_name = row.get('first_name') last_name = row.get('last_name') whatsapp_to = row.get('whatsapp_to_number', '') phone_mobile = None if whatsapp_to and whatsapp_to.startswith('whatsapp:'): phone_mobile = whatsapp_to.replace('whatsapp:', '') existing = supabase.table('dp_customer').select('customer_id').eq('external_key', customer_bpid).execute() if existing.data and len(existing.data) > 0: customer_id = existing.data[0]['customer_id'] supabase.table('dp_customer').update({ 'first_name': first_name, 'last_name': last_name, 'phone_mobile': phone_mobile, 'updated_at': datetime.utcnow().isoformat() }).eq('customer_id', customer_id).execute() logger.info(f"[EXPORT] Updated dp_customer for BPID={customer_bpid}, customer_id={customer_id}") else: new_customer_id = str(uuid.uuid4()) supabase.table('dp_customer').insert({ 'customer_id': new_customer_id, 'external_key': customer_bpid, 'first_name': first_name, 'last_name': last_name, 'phone_mobile': phone_mobile }).execute() customer_id = new_customer_id logger.info(f"[EXPORT] Created dp_customer for BPID={customer_bpid}, customer_id={customer_id}") export_list.append({ 'customer_id': customer_id, 'reg': row.get('reg'), 'model': row.get('model'), 'payments_remaining': row.get('payments_remaining'), 'to_number': whatsapp_to, 'first_name': first_name, 'last_name': last_name }) logger.info(f"[EXPORT] Complete: {len(export_list)} customers exported") return { "success": True, "count": len(export_list), "book_month": book_month, "campaign": campaign, "customers": export_list } except Exception as e: logger.error(f"[EXPORT] Error: {str(e)}") raise HTTPException(status_code=500, detail=f"Export failed: {str(e)}") # --- WhatsApp automation stub --- from pydantic import BaseModel class WhatsAppMessage(BaseModel): contact_id: str phone_number: str | None = None campaign_id: str | None = None message_preview: str class LiaNudgeRequest(BaseModel): conversation_id: str nudge_type: str = "nudge_1" # e.g. "nudge_1" or "nudge_2" payments_remaining: int | None = None # reserved for future use class LiaAdvisorBridgeRequest(BaseModel): conversation_id: str reason: str | None = None # e.g. "No reply after Nudge 3" priority: str | None = "normal" # e.g. "normal", "high" @app.post("/api/lia/advisor-bridge", dependencies=[Depends(require_api_key)]) async def lia_advisor_bridge(payload: LiaAdvisorBridgeRequest): ... @app.post("/api/whatsapp/send", dependencies=[Depends(require_auth)]) async def send_whatsapp_stub(payload: WhatsAppMessage): # Later: push to Make/n8n or another service. print("WhatsApp stub queued:", payload.dict()) return { "status": "queued", "detail": "WhatsApp message queued for external automation (stub).", } @app.get("/api/templates/{template_key}", dependencies=[Depends(require_auth)]) async def get_template(template_key: str): """ Fetch a template and its objective/guidance from dp_templates + dp_objectives. Returns 404 if not found or not active. """ try: template = await fetch_template(template_key) if not template: raise HTTPException(status_code=404, detail="Template not found") return template except HTTPException: raise except Exception as e: # Log internal error but don't expose details to client print(f"Error fetching template {template_key}: {e}") raise HTTPException(status_code=500, detail="Internal server error") @app.post("/api/conversations/{conversation_id}/defer", dependencies=[Depends(require_auth)]) async def api_defer_conversation( conversation_id: str, payload: dict = Body(...), ): """ Pause a conversation until a future date. Used by the AI layer when the customer says "not right now" or similar. """ defer_until_str = payload.get("defer_until") reason = payload.get("reason", None) if not defer_until_str: return JSONResponse({"error": "defer_until is required (YYYY-MM-DD)"}, status_code=400) try: defer_until = date.fromisoformat(defer_until_str) except ValueError: return JSONResponse({"error": "Invalid date format for defer_until"}, status_code=400) ok = await defer_conversation(conversation_id, defer_until, reason) if not ok: return JSONResponse({"error": "Conversation not found"}, status_code=404) return {"status": "ok", "conversation_id": conversation_id, "defer_until": defer_until_str} @app.post("/api/conversations/{conversation_id}/resume", dependencies=[Depends(require_auth)]) async def api_resume_conversation( conversation_id: str, ): """ Resume a paused conversation (e.g. manager override from T-card, or scheduled reactivation). """ ok = await resume_conversation(conversation_id) if not ok: return JSONResponse({"error": "Conversation not found"}, status_code=404) return {"status": "ok", "conversation_id": conversation_id} @app.post("/api/conversations/{conversation_id}/escalate", dependencies=[Depends(require_api_key)]) async def api_escalate_conversation( conversation_id: str, payload: dict = Body(...), ): """ Mark a conversation as escalated so a manager can step in. Used when Lia needs help - not for customer-requested pauses. """ reason = payload.get("reason") valid_reasons = { "unrecognised_message", "appointment_confirmation_needed", "customer_issue", } if not reason or reason not in valid_reasons: return JSONResponse( {"error": f"Invalid reason. Must be one of: {', '.join(valid_reasons)}"}, status_code=400 ) ok = await escalate_conversation(conversation_id, reason) if not ok: return JSONResponse({"error": "Conversation not found"}, status_code=404) return {"status": "ok", "conversation_id": conversation_id, "reason": reason} @app.post("/api/conversations/{conversation_id}/manager_reply", dependencies=[Depends(require_auth)]) async def api_manager_reply( conversation_id: str, payload: dict = Body(...), ): """ Manager sends a reply to resolve an escalated conversation. This: - Sets status = 'active' - Sets needs_human_review = FALSE - Clears escalation_reason - Logs the manager message to dp_messages Request body: - message_body: str (required) - template_key: str (optional) - mode: str (optional, default 'campaign') - channel: str (optional, default 'whatsapp') """ from app.db import resolve_manager_reply, fetch_conversation_status message_body = payload.get("message_body") template_key = payload.get("template_key") mode = payload.get("mode", "campaign") channel = payload.get("channel", "whatsapp") if not message_body: return JSONResponse({"error": "message_body is required"}, status_code=400) conv_status = await fetch_conversation_status(conversation_id) if not conv_status: return JSONResponse({"error": "Conversation not found"}, status_code=404) if conv_status.get("status") == "dnc": return JSONResponse( {"error": "Conversation is DNC; no further messages allowed."}, status_code=409 ) ok = await resolve_manager_reply(conversation_id) if not ok: return JSONResponse({"error": "Failed to update conversation"}, status_code=500) try: await insert_message( conversation_id=conversation_id, direction="outbound", channel=channel, sender="manager", message_body=message_body, template_key=template_key, mode=mode, ) except Exception as e: print(f"Failed to log manager message: {e}") return { "status": "ok", "conversation_id": conversation_id, "message_logged": True, "escalation_resolved": True, } @app.get("/api/conversations/{conversation_id}/status") async def api_conversation_status( conversation_id: str, user: dict = Depends(require_auth), ): """ Return the current status of a conversation: active/paused and any defer info. """ data = await fetch_conversation_status(conversation_id) if not data: return JSONResponse({"error": "Conversation not found"}, status_code=404) return { "conversation_id": conversation_id, "status": data["status"], "defer_until": data["defer_until"], "defer_reason": data["defer_reason"], "escalation_reason": data["escalation_reason"], "escalation_at": data["escalation_at"].isoformat() if data["escalation_at"] else None, } @app.post("/api/messages") async def api_log_message( payload: dict, user: dict = Depends(require_auth), ): """ Log an outbound or inbound message for a conversation. Used by the Reminders tab when an advisor uses a Lia script. """ conversation_id = payload.get("conversation_id") message_body = payload.get("message_body") mode = payload.get("mode") # 'reminder', 'missed', 'followup', 'noreply' template_key = payload.get("template_key") ab_variant = payload.get("ab_variant") channel = payload.get("channel", "whatsapp") direction = payload.get("direction", "outbound") sender = payload.get("sender", "lia") if not conversation_id or not message_body: return JSONResponse({"error": "conversation_id and message_body are required"}, status_code=400) try: await insert_message( conversation_id=conversation_id, direction=direction, channel=channel, sender=sender, message_body=message_body, template_key=template_key, mode=mode, ab_variant=ab_variant, ) except Exception as e: print(f"Failed to insert message: {e}") return JSONResponse({"error": "Failed to log message"}, status_code=500) return {"status": "ok"} import logging logger = logging.getLogger(__name__) @app.get("/api/messages") async def api_get_messages( conversation_id: str, limit: int = 50, user: dict = Depends(require_auth), ): """ Return the latest messages for a conversation. """ try: msgs = await fetch_messages_for_conversation(conversation_id, limit=limit) except Exception: logger.exception("Failed to fetch messages") return JSONResponse({"error": "Failed to fetch messages"}, status_code=500) return { "conversation_id": conversation_id, "messages": msgs, } from app.db import get_conn, release_conn @app.get("/api/conversations/{conversation_id}/messages", dependencies=[Depends(require_api_key)]) async def api_conversation_messages( conversation_id: str, limit: int = Query(100, le=500), ): """ Return the message history for a conversation ordered by created_at. Uses the new dp_messages schema with from_number, to_number, body, etc. Secured with API key for Make.com access. """ conn = await get_conn() try: from uuid import UUID conv_uuid = UUID(conversation_id) rows = await conn.fetch( """ SELECT message_id, conversation_id, direction, channel, sender, message_body, template_key, mode, created_at FROM dp_messages WHERE conversation_id = $1 ORDER BY created_at ASC LIMIT $2 """, conv_uuid, limit, ) messages = [ { "id": str(row["message_id"]), "conversation_id": str(row["conversation_id"]), "direction": row["direction"], "channel": row["channel"], "sender": row["sender"], "message_body": row["message_body"], "template_key": row["template_key"], "mode": row["mode"], "created_at": row["created_at"].isoformat() if row["created_at"] else None, } for row in rows ] return { "conversation_id": conversation_id, "message_count": len(messages), "messages": messages, } except ValueError: raise HTTPException(status_code=400, detail="Invalid conversation_id format") except Exception as e: logger.exception(f"Failed to fetch messages for conversation {conversation_id}") raise HTTPException(status_code=500, detail="Failed to fetch messages") finally: await release_conn(conn) @app.get("/api/debug/conversations/{conversation_id}", dependencies=[Depends(require_api_key)]) async def debug_get_conversation(conversation_id: UUID): """ Debug endpoint: return a single conversation and all its messages. """ result = await get_conversation_with_messages(conversation_id) if not result: raise HTTPException(status_code=404, detail="Conversation not found") return result @app.get("/api/debug/customers/{customer_id}/conversations", dependencies=[Depends(require_api_key)]) async def debug_list_customer_conversations(customer_id: UUID): """ Debug endpoint: list all conversations for a customer (no messages, just summary). """ conversations = await list_conversations_for_customer(customer_id) return { "customer_id": str(customer_id), "count": len(conversations), "conversations": conversations, } @app.get("/api/debug/summary", dependencies=[Depends(require_api_key)]) async def debug_conversation_summary(): """ Debug endpoint: return aggregate counts of conversations by status, stage, and campaign. """ summary = await get_conversation_summary() return summary @app.get("/api/tco-preview") async def api_tco_preview( conversation_id: str, user: dict = Depends(require_auth), ): """ Return a simple stubbed TCO preview for a conversation. This will later be replaced with real logic using book data + mileage/MPG. """ # For now, just return some example structure. # Later we can look up customer, vehicle, and payment data. from datetime import datetime # Basic example values – placeholder only monthly_payment = 399.00 est_fuel = 120.00 est_insurance = 70.00 est_tax = 20.00 est_service = 25.00 total_monthly = monthly_payment + est_fuel + est_insurance + est_tax + est_service return { "conversation_id": conversation_id, "generated_at": datetime.utcnow().isoformat() + "Z", "currency": "GBP", "items": [ {"label": "Finance payment", "amount": monthly_payment, "key": "finance"}, {"label": "Estimated fuel", "amount": est_fuel, "key": "fuel"}, {"label": "Estimated insurance", "amount": est_insurance, "key": "insurance"}, {"label": "Vehicle tax", "amount": est_tax, "key": "tax"}, {"label": "Service & maintenance", "amount": est_service, "key": "service"}, ], "total_monthly": total_monthly, } from fastapi import Depends, HTTPException, Request import json @app.post("/api/lia/inbound", response_model=LiaReply, dependencies=[Depends(require_api_key)]) async def lia_inbound(request: Request): """ Receive inbound customer messages from Make.com for Lia to process. Secured with x-api-key (require_api_key), not dashboard session. Conversation-aware: Looks up existing conversation by phone number if conversation_id not provided. Creates/loads conversation, logs messages, applies URE logic, returns reply. Phone number handling: - Strips "whatsapp:" prefix from from_number - Looks up dp_customer by phone_mobile in E.164 format (+44...) """ from app.lia_outbound import DNCBlockedError try: # Read raw body and log for debugging raw_bytes = await request.body() raw_text = raw_bytes.decode("utf-8", errors="ignore") print("🔵 LIA INBOUND RAW BODY:", raw_text) # Safely parse JSON try: data = json.loads(raw_text) except json.JSONDecodeError as e: raise HTTPException(status_code=422, detail=f"Invalid JSON body: {str(e)}") # Build LiaInbound model from parsed JSON payload = LiaInbound(**data) # Run your existing processing logic reply = await process_inbound_message(payload) return reply except DNCBlockedError as e: from_num = data.get('from_number', 'unknown') if 'data' in dir() else 'unknown' print(f"🚫 Inbound DNC block: from_number={from_num}") raise HTTPException(status_code=409, detail="Customer is DNC - no further contact allowed") except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) except HTTPException: raise except Exception as e: print(f"❌ Error in lia_inbound endpoint: {e}") raise HTTPException(status_code=500, detail="Internal server error processing message") @app.post("/api/lia/outbound", response_model=OutboundResponse, dependencies=[Depends(require_api_key)]) async def lia_outbound(payload: OutboundRequest): """ Generate outbound WhatsApp message content for renewal customers. Called by Make.com to prepare campaign messages. This endpoint does NOT send the message - it only prepares the content. Make.com will handle the actual sending via Twilio. Features: - DNC guard: Blocks messages to customers with DNC status - Campaign-safety gate: Only processes if an active campaign matches the request - Automatic stage selection based on payments_remaining - Support for NEW vs USED vehicle messaging (asset_type: "new" or "used") - Sandbox mode for testing without real phone numbers - Returns personalized message text with customer details Returns HTTP 400 for validation errors (invalid UUID, missing campaign, etc.) Returns HTTP 409 if customer has DNC status. """ from app.lia_outbound import ( DNCBlockedError, CampaignNotFoundError, InvalidCustomerIdError, MissingPhoneNumberError, InvalidPaymentsBandError, DuplicateConversationError ) # Log incoming request for debugging print(f"📤 Outbound request: customer_id={payload.customer_id}, campaign={payload.campaign}, " f"payments_remaining={payload.payments_remaining}, asset_type={payload.asset_type}, " f"sandbox_mode={payload.sandbox_mode}") try: response = await process_outbound_message(payload) print(f"✅ Outbound success: conversation_id={response.conversation_id}, stage={response.stage}, to_number={response.to_number}") return response except InvalidPaymentsBandError as e: print(f"⚠️ Outbound invalid payments band: {e}") raise HTTPException(status_code=400, detail=str(e)) except DuplicateConversationError as e: print(f"⚠️ Outbound duplicate conversation: conversation_id={e.conversation_id}, stage={e.stage}") raise HTTPException( status_code=409, detail={ "error": str(e), "conversation_id": e.conversation_id, "stage": e.stage, } ) except InvalidCustomerIdError as e: print(f"⚠️ Outbound validation error: {e}") raise HTTPException(status_code=400, detail=str(e)) except MissingPhoneNumberError as e: print(f"⚠️ Outbound missing phone: {e}") raise HTTPException(status_code=400, detail=str(e)) except DNCBlockedError as e: print(f"🚫 Outbound DNC block: customer_id={payload.customer_id}") raise HTTPException(status_code=409, detail="Customer is DNC - no further contact allowed") except CampaignNotFoundError as e: print(f"⚠️ Outbound campaign not found: {e}") raise HTTPException(status_code=400, detail=str(e)) except CampaignNotActiveError as e: print(f"⚠️ Outbound campaign not active: campaign={payload.campaign}") raise HTTPException(status_code=400, detail="Campaign not active for this request") except Exception as e: print(f"❌ Outbound unexpected error: customer_id={payload.customer_id}, campaign={payload.campaign}, " f"payments_remaining={payload.payments_remaining}, sandbox_mode={payload.sandbox_mode}, error={e}") raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}") @app.get("/api/lia/campaigns/active", dependencies=[Depends(require_api_key)]) async def get_active_campaigns( site_id: str = "lincoln_audi", asset_type: str | None = None, ): """ Debug endpoint to list campaigns that are currently active by production criteria. This endpoint is for debugging/ops use. It reflects the strict production criteria: - status IN ('running', 'active') - date_from <= today (or NULL) - date_to >= today (or NULL) Sandbox mode is NOT applied here - this shows what production calls will accept. Query parameters: - site_id: Filter by site (default: 'lincoln_audi') - asset_type: Optional filter for campaign type (e.g., 'PCP', 'pcp_stage_auto') """ from app.lia_outbound import list_active_campaigns campaigns = await list_active_campaigns( site_id=site_id, asset_type=asset_type, ) return { "site_id": site_id, "asset_type_filter": asset_type, "count": len(campaigns), "campaigns": campaigns, } @app.post("/api/lia/nudge", dependencies=[Depends(require_api_key)]) async def lia_nudge(payload: LiaNudgeRequest): """ Generate and log a nudge message for a conversation, without needing an inbound customer message. This is designed to be called from Make.com using the x-api-key. It returns the text Lia should send, and logs it into dp_messages. Supports both generic nudges (NUDGE_1_EOT, etc.) and stage-specific nudges (NUDGE_1_12_NEW, etc.). Safety checks: - 404 if conversation not found - 409 if conversation.status = 'dnc' - 409 if conversation.status = 'paused' - 409 if conversation.defer_until > today """ from app.lia_outbound import ( DNCBlockedError, NudgeBlockedError, ConversationNotFoundError ) from datetime import date as date_type print(f"📤 Nudge request: conversation_id={payload.conversation_id}, nudge_type={payload.nudge_type}") # Map simple nudge_type to the internal template key mapping = { "nudge_1": "NUDGE_1_EOT", "nudge_2": "NUDGE_2_EOT", "nudge_3": "NUDGE_3_EOT", "NUDGE_1_EOT": "NUDGE_1_EOT", "NUDGE_2_EOT": "NUDGE_2_EOT", "NUDGE_3_EOT": "NUDGE_3_EOT", } template_key = mapping.get(payload.nudge_type, payload.nudge_type) # Fetch the template text - try stage-specific nudges first, then generic reply_text = None # Check if it's a stage-specific nudge (e.g., NUDGE_1_12_NEW, NUDGE_2_9_USED, etc.) for stage_dict in [LIA_EOT_12_NUDGES, LIA_EOT_9_NUDGES, LIA_EOT_6_NUDGES, LIA_EOT_3_NUDGES]: if template_key in stage_dict: reply_text = stage_dict[template_key] break # Fall back to generic nudges if not reply_text and template_key in LIA_NUDGE_TEMPLATES: reply_text = LIA_NUDGE_TEMPLATES[template_key] if not reply_text: print(f"⚠️ Nudge invalid type: {payload.nudge_type}") raise HTTPException(status_code=400, detail=f"Invalid nudge_type: {payload.nudge_type}") # Fetch customer name and vehicle model from conversation customer_name = "there" vehicle_model = "Audi" conv_status = None conv_stage = None conv_channel = None defer_until = None conn = await get_conn() try: conv_uuid = UUID(payload.conversation_id) # ------------------------------------------------------------------------- # STEP 1: Load conversation and perform safety checks # ------------------------------------------------------------------------- conv_row = await conn.fetchrow( """ SELECT conversation_id, status, stage, channel, defer_until, defer_reason FROM dp_conversations WHERE conversation_id = $1 """, conv_uuid, ) # Check 1: Conversation must exist if not conv_row: raise ConversationNotFoundError(f"Conversation {payload.conversation_id} not found") conv_status = conv_row["status"] conv_stage = conv_row["stage"] or "unknown" conv_channel = conv_row["channel"] or "whatsapp" defer_until = conv_row["defer_until"] # Check 2: DNC block if conv_status == "dnc": raise DNCBlockedError("Customer is DNC - no further contact allowed") # Check 3: Paused block if conv_status == "paused": raise NudgeBlockedError("Conversation is paused - no nudge sent", reason="paused") # Check 4: Deferred block (defer_until > today) if defer_until is not None: today = date_type.today() # Handle both date and datetime types defer_date = defer_until if isinstance(defer_until, date_type) else defer_until.date() if defer_date > today: raise NudgeBlockedError( f"Conversation is deferred until {defer_date.isoformat()} - no nudge sent", reason="deferred" ) # ------------------------------------------------------------------------- # STEP 2: Fetch customer details for personalization # ------------------------------------------------------------------------- row = await conn.fetchrow( """ SELECT dc.first_name, dc.last_name, v.model as vehicle_model FROM dp_conversations conv LEFT JOIN dp_customer dc ON dc.customer_id = conv.customer_id LEFT JOIN v_retention_book_current vr ON vr.customer_id = conv.customer_id LEFT JOIN vehicle v ON v.registration = vr.reg WHERE conv.conversation_id = $1 """, conv_uuid, ) if row: first_name = row["first_name"] or "" last_name = row["last_name"] or "" customer_name = first_name.strip() or "there" vehicle_model = row["vehicle_model"] or "Audi" # ------------------------------------------------------------------------- # STEP 3: Replace placeholders in nudge text # ------------------------------------------------------------------------- # Handle both placeholder styles: [customer_name] and [customer name] reply_text = reply_text.replace("[customer_name]", customer_name) reply_text = reply_text.replace("[customer name]", customer_name) reply_text = reply_text.replace("[vehicle_model]", vehicle_model) # ------------------------------------------------------------------------- # STEP 4: Log the nudge to dp_messages # ------------------------------------------------------------------------- await conn.execute( """ INSERT INTO dp_messages ( conversation_id, direction, channel, sender, message_body, template_key, mode ) VALUES ($1, 'outbound', $2, 'lia', $3, $4, $5) """, conv_uuid, conv_channel, reply_text, template_key, f"NUDGE_{payload.nudge_type.upper()}", ) # ------------------------------------------------------------------------- # STEP 5: Update conversation updated_at timestamp # ------------------------------------------------------------------------- await conn.execute( """ UPDATE dp_conversations SET updated_at = NOW() WHERE conversation_id = $1 """, conv_uuid, ) print(f"✅ Nudge success: conversation_id={payload.conversation_id}, stage={conv_stage}, template={template_key}") return { "conversation_id": payload.conversation_id, "status": conv_status, "defer_until": defer_until.isoformat() if defer_until else None, "reply": reply_text, "stage": conv_stage, } except ConversationNotFoundError as e: print(f"⚠️ Nudge conversation not found: {e}") raise HTTPException(status_code=404, detail=str(e)) except DNCBlockedError as e: print(f"🚫 Nudge DNC block: conversation_id={payload.conversation_id}") raise HTTPException(status_code=409, detail="Customer is DNC - no further contact allowed") except NudgeBlockedError as e: print(f"⚠️ Nudge blocked ({e.reason}): conversation_id={payload.conversation_id}") raise HTTPException(status_code=409, detail=str(e)) except ValueError as e: print(f"⚠️ Nudge invalid UUID: {payload.conversation_id}") raise HTTPException(status_code=400, detail=f"Invalid conversation_id format: {payload.conversation_id}") except Exception as e: print(f"❌ Nudge unexpected error: conversation_id={payload.conversation_id}, error={e}") raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}") finally: await release_conn(conn) # ============================================================================= # LIA UI v1 - PAGE ROUTES # ============================================================================= @app.get("/lia", response_class=HTMLResponse, dependencies=[Depends(require_auth)]) async def lia_redirect(): """Redirect /lia to /lia/conversations""" return RedirectResponse(url="/lia/conversations", status_code=302) @app.get("/lia/dashboard", response_class=HTMLResponse, dependencies=[Depends(require_auth)]) async def lia_dashboard(request: Request): """LIA Dashboard - high-level KPIs""" return templates.TemplateResponse("lia_base.html", { "request": request, "active_nav": "dashboard", }) @app.get("/lia/conversations", response_class=HTMLResponse, dependencies=[Depends(require_auth)]) async def lia_conversations(request: Request): """LIA Conversations Inbox""" return templates.TemplateResponse("lia_conversations.html", { "request": request, "active_nav": "conversations", }) @app.get("/lia/conversations/{conversation_id}", response_class=HTMLResponse, dependencies=[Depends(require_auth)]) async def lia_conversation_detail(request: Request, conversation_id: str): """LIA Conversation Detail View""" return templates.TemplateResponse("lia_conversation_detail.html", { "request": request, "active_nav": "conversations", "conversation_id": conversation_id, }) @app.get("/lia/campaign-customers", response_class=HTMLResponse, dependencies=[Depends(require_auth)]) async def lia_campaign_customers(request: Request): """LIA Campaign Customers List""" return templates.TemplateResponse("lia_base.html", { "request": request, "active_nav": "campaign-customers", }) @app.get("/lia/appointments", response_class=HTMLResponse, dependencies=[Depends(require_auth)]) async def lia_appointments(request: Request): """LIA Appointments Diary""" return templates.TemplateResponse("lia_base.html", { "request": request, "active_nav": "appointments", }) @app.get("/lia/data-campaigns", response_class=HTMLResponse, dependencies=[Depends(require_auth)]) async def lia_data_campaigns(request: Request): """LIA Data & Campaigns Management""" return templates.TemplateResponse("lia_base.html", { "request": request, "active_nav": "data-campaigns", }) # ============================================================================= # LIA UI v1 - API ENDPOINTS # ============================================================================= @app.get("/api/lia/conversations", dependencies=[Depends(require_auth)]) async def api_lia_conversations( page: int = Query(1, ge=1), limit: int = Query(25, ge=1, le=100), status: str = Query("active"), stage: Optional[str] = Query(None), search: Optional[str] = Query(None), ): """List conversations with pagination and filters - uses v_lia_conversations_current view""" conn = await get_conn() try: offset = (page - 1) * limit # Build WHERE clause based on filters conditions = [] params = [] param_idx = 1 # Status filter if status == "active": conditions.append(f"v.status IN ('active', 'awaiting_customer')") elif status == "escalated": conditions.append(f"v.status = 'needs_review'") elif status == "awaiting": conditions.append(f"v.status = 'awaiting_customer'") elif status == "paused": conditions.append(f"v.status IN ('paused', 'dnc', 'opted_out')") elif status == "booked": conditions.append(f"v.appointment_status = 'BOOKED'") # Stage filter (now uses the stage column from dp_conversations) if stage: conditions.append(f"v.stage = ${param_idx}") params.append(stage) param_idx += 1 # Search filter if search: conditions.append(f"(v.first_name ILIKE ${param_idx} OR v.last_name ILIKE ${param_idx} OR v.phone_mobile ILIKE ${param_idx})") params.append(f"%{search}%") param_idx += 1 where_clause = " AND ".join(conditions) if conditions else "1=1" # Count total from the view count_query = f""" SELECT COUNT(*) FROM v_lia_conversations_current v WHERE {where_clause} """ total_row = await conn.fetchrow(count_query, *params) total_count = total_row[0] if total_row else 0 # Count escalated (for badge) escalated_row = await conn.fetchrow( "SELECT COUNT(*) FROM v_lia_conversations_current WHERE needs_human_review = TRUE OR status = 'needs_review'" ) escalated_count = escalated_row[0] if escalated_row else 0 # Fetch conversations from the view with last message query = f""" SELECT v.conversation_id, v.customer_id, v.first_name, v.last_name, v.phone_mobile, v.stage, v.status, v.appointment_status, v.campaign, v.soft_no_count, v.confusion_count, v.needs_human_review, v.defer_until, v.defer_reason, v.created_at, v.updated_at, (SELECT message_body FROM dp_messages m WHERE m.conversation_id = v.conversation_id AND (m.mode IS NULL OR m.mode != 'NOTE') ORDER BY m.created_at DESC LIMIT 1) as last_message, (SELECT c.appointment_date_display FROM dp_conversations c WHERE c.conversation_id = v.conversation_id LIMIT 1) as appointment_date_display, (SELECT c.appointment_time_display FROM dp_conversations c WHERE c.conversation_id = v.conversation_id LIMIT 1) as appointment_time_display FROM v_lia_conversations_current v WHERE {where_clause} ORDER BY v.updated_at DESC LIMIT ${param_idx} OFFSET ${param_idx + 1} """ params.extend([limit, offset]) rows = await conn.fetch(query, *params) conversations = [] for row in rows: row_dict = dict(row) first_name = row_dict.get("first_name") or "" last_name = row_dict.get("last_name") or "" customer_name = f"{first_name} {last_name}".strip() or "Unknown" # Parse stage from the stage column (e.g., "18", "12", "9", "6", "3") stage_str = row_dict.get("stage") or "" stage_value = None if stage_str and stage_str.isdigit(): stage_value = int(stage_str) # Build next_appointment if booked next_appointment = None if row_dict.get("appointment_status") == "BOOKED" and row_dict.get("appointment_date_display"): next_appointment = { "date": row_dict.get("appointment_date_display"), "time": row_dict.get("appointment_time_display") or "TBC", } conversations.append({ "id": str(row_dict["conversation_id"]), "customer_name": customer_name, "phone": row_dict.get("phone_mobile") or "", "vehicle_description": "", # Not available in current view "reg": "", # Not available in current view "stage": stage_value, "status": row_dict.get("status"), "appointment_status": row_dict.get("appointment_status"), "campaign_name": row_dict.get("campaign"), "advisor_name": None, # Not in current schema "next_appointment": next_appointment, "hot_flag": row_dict.get("needs_human_review") or False, "soft_no_count": row_dict.get("soft_no_count") or 0, "confusion_count": row_dict.get("confusion_count") or 0, "defer_until": row_dict.get("defer_until").isoformat() if row_dict.get("defer_until") else None, "last_message": (row_dict.get("last_message") or "")[:80], "last_activity": row_dict["updated_at"].isoformat() if row_dict.get("updated_at") else None, }) return { "conversations": conversations, "total": total_count, "escalated_count": escalated_count, "page": page, "limit": limit, } finally: await release_conn(conn) @app.get("/api/lia/campaign-customers", dependencies=[Depends(require_auth)]) async def api_lia_campaign_customers( status: Optional[str] = Query(None), stage: Optional[str] = Query(None), search: Optional[str] = Query(None), campaign: Optional[str] = Query(None), advisor: Optional[str] = Query(None), channel: Optional[str] = Query(None), payment_filter: Optional[str] = Query(None), months_max: Optional[int] = Query(None), ): """Get campaign customers for dashboard - uses v_lia_conversations_current view joined with VWFS data""" conn = await get_conn() try: conditions = [] params = [] param_idx = 1 # Status filter if status and status != 'all': if status == 'engaged': conditions.append("v.status = 'active'") elif status == 'appointment': conditions.append("v.appointment_status = 'BOOKED'") elif status == 'sold': conditions.append("v.status = 'sold'") elif status == 'declined': conditions.append("v.status IN ('paused', 'dnc', 'opted_out')") elif status == 'no-response': conditions.append("v.status = 'awaiting_customer'") # Stage filter if stage: conditions.append(f"v.stage = ${param_idx}") params.append(stage) param_idx += 1 # Months remaining filter (stage as integer) if months_max: conditions.append(f"CAST(v.stage AS INTEGER) <= ${param_idx}") params.append(months_max) param_idx += 1 # Search filter - now includes registration and model if search: conditions.append(f"""( v.first_name ILIKE ${param_idx} OR v.last_name ILIKE ${param_idx} OR v.phone_mobile ILIKE ${param_idx} OR vwfs."VRM" ILIKE ${param_idx} OR vwfs."Model" ILIKE ${param_idx} )""") params.append(f"%{search}%") param_idx += 1 # Campaign filter if campaign: conditions.append(f"v.campaign = ${param_idx}") params.append(campaign) param_idx += 1 # Advisor filter if advisor and advisor != 'all': conditions.append(f"vwfs.\"Assigned to email\" ILIKE ${param_idx}") params.append(f"%{advisor}%") param_idx += 1 # Channel filter if channel and channel != 'all': conditions.append(f"v.channel = ${param_idx}") params.append(channel) param_idx += 1 # Add mandatory filters to exclude old test data and inbound_reply # Also filter to December 2025 test cohort for now (can be relaxed later) mandatory_filters = [ "v.campaign IS NOT NULL", "v.campaign != 'inbound_reply'", "v.first_name IS NOT NULL", "v.first_name != ''", "v.first_name != 'Unknown'", "v.updated_at >= '2025-12-01'" ] all_conditions = mandatory_filters + conditions where_clause = " AND ".join(all_conditions) query = f""" SELECT v.conversation_id, v.customer_id, v.external_key, v.first_name, v.last_name, v.phone_mobile, v.postcode, v.stage, v.status, v.appointment_status, v.campaign, v.channel, v.soft_no_count, v.needs_human_review, v.defer_until, v.created_at, v.updated_at, (SELECT c.appointment_date_display FROM dp_conversations c WHERE c.conversation_id = v.conversation_id LIMIT 1) as appointment_date_display, (SELECT c.appointment_time_display FROM dp_conversations c WHERE c.conversation_id = v.conversation_id LIMIT 1) as appointment_time_display, vwfs."VRM" as registration, vwfs."Make" as make, vwfs."Model" as model, vwfs."Derivative" as variant, vwfs."Fuel Type" as fuel_type, vwfs."Monthly Payment" as monthly_payment, vwfs."Balloon" as balloon, vwfs."Settlement" as settlement, vwfs."ContractEndDate" as contract_end_date, vwfs."PaymentsLeft" as payments_left, vwfs."Assigned to email" as assigned_advisor, vwfs."FinancePlanNumber" as agreement_number, (SELECT COUNT(*) FROM dp_messages m WHERE m.conversation_id = v.conversation_id AND m.direction = 'outbound') as outbound_count, (SELECT COUNT(*) FROM dp_messages m WHERE m.conversation_id = v.conversation_id AND m.direction = 'inbound') as inbound_count FROM v_lia_conversations_current v LEFT JOIN stg_vwfs_book_raw_csv vwfs ON v.external_key = vwfs."CustomerBPID" WHERE {where_clause} ORDER BY v.updated_at DESC LIMIT 200 """ rows = await conn.fetch(query, *params) customers = [] for row in rows: row_dict = dict(row) first_name = row_dict.get("first_name") or "" last_name = row_dict.get("last_name") or "" customer_name = f"{first_name} {last_name}".strip() or "Unknown" # Parse stage to months remaining stage_str = row_dict.get("stage") or "" months_remaining = None if stage_str and stage_str.isdigit(): months_remaining = int(stage_str) # Get payment info current_payment = row_dict.get("monthly_payment") if current_payment is not None: try: current_payment = float(current_payment) except (ValueError, TypeError): current_payment = None # Map dp_conversations status to dashboard status # Status logic based on messages sent/received: # - APPOINTMENT_BOOKED: customer has confirmed appointment # - ENGAGED: customer has responded (has inbound messages) but no final outcome # - MESSAGE_SENT: we've sent messages but customer hasn't responded yet # - DECLINED: customer has opted out, DNC, or paused # - NOT_CONTACTED: no messages sent yet conv_status = row_dict.get("status") or "" appt_status = row_dict.get("appointment_status") or "" outbound_count = row_dict.get("outbound_count") or 0 inbound_count = row_dict.get("inbound_count") or 0 if appt_status == "BOOKED": dashboard_status = "APPOINTMENT_BOOKED" elif conv_status in ("paused", "dnc", "opted_out"): dashboard_status = "DECLINED" elif inbound_count > 0: # Customer has responded - they're engaged dashboard_status = "ENGAGED" elif outbound_count > 0: # We've sent messages but no response yet dashboard_status = "MESSAGE_SENT" else: dashboard_status = "NOT_CONTACTED" # Build next action next_action = "Follow up" next_action_due = None if appt_status == "BOOKED" and row_dict.get("appointment_date_display"): next_action = f"Appointment: {row_dict.get('appointment_date_display')} {row_dict.get('appointment_time_display') or ''}" next_action_due = row_dict.get("appointment_date_display") elif row_dict.get("defer_until"): next_action = "Deferred - check back" next_action_due = row_dict["defer_until"].isoformat() # Extract advisor name from email advisor_email = row_dict.get("assigned_advisor") or "" advisor_name = advisor_email.split("@")[0].replace(".", " ").title() if advisor_email else "Lia Bot" # Build model display string make = row_dict.get("make") or "" model = row_dict.get("model") or "" model_display = f"{make} {model}".strip() # Get postcode for town (first part) postcode = row_dict.get("postcode") or "" town = postcode.split()[0] if postcode else "" customers.append({ "id": str(row_dict["conversation_id"]), "name": customer_name, "title": "", "town": town, "reg": row_dict.get("registration") or "", "model": model_display, "variant": row_dict.get("variant") or "", "fuelType": row_dict.get("fuel_type") or "", "status": dashboard_status, "currentPayment": current_payment, "proposedPayment": None, "paymentDelta": None, "agreementNumber": row_dict.get("agreement_number") or row_dict.get("campaign") or "", "maturityDate": row_dict.get("contract_end_date"), "monthsRemaining": months_remaining, "lastContactChannel": row_dict.get("channel") or "WhatsApp", "lastContactDate": row_dict["updated_at"].strftime("%Y-%m-%d") if row_dict.get("updated_at") else None, "lastContactUser": advisor_name, "assignedAdvisor": advisor_email or "lia-bot", "nextAction": next_action, "nextActionDueDate": next_action_due, "hasServicePlan": False, "phone": row_dict.get("phone_mobile") or "", "campaign": row_dict.get("campaign") or "", "stage": stage_str, "paymentsLeft": row_dict.get("payments_left"), "balloon": row_dict.get("balloon"), "settlement": row_dict.get("settlement"), }) return {"customers": customers, "total": len(customers)} finally: await release_conn(conn) @app.get("/api/lia/stats", dependencies=[Depends(require_auth)]) async def api_lia_stats(): """Get LIA conversation statistics from v_lia_conversations_current""" conn = await get_conn() try: # Apply same mandatory filters as campaign-customers endpoint base_filter = """ v.campaign IS NOT NULL AND v.campaign != 'inbound_reply' AND v.first_name IS NOT NULL AND v.first_name != '' AND v.first_name != 'Unknown' AND v.updated_at >= '2025-12-01' """ # Get overall stats stats_row = await conn.fetchrow(f""" SELECT COUNT(*) as total_customers, COUNT(*) FILTER (WHERE v.appointment_status = 'BOOKED') as appointments_booked, COUNT(*) FILTER (WHERE v.status = 'active') as engaged, COUNT(*) FILTER (WHERE v.status = 'awaiting_customer') as awaiting_response, COUNT(*) FILTER (WHERE v.status IN ('paused', 'dnc', 'opted_out')) as declined, COUNT(*) FILTER (WHERE v.defer_until IS NOT NULL) as deferred FROM v_lia_conversations_current v WHERE {base_filter} """) # Get message counts from dp_messages for these conversations messages_row = await conn.fetchrow(f""" SELECT COUNT(*) FILTER (WHERE m.direction = 'outbound') as messages_sent, COUNT(DISTINCT v.conversation_id) FILTER (WHERE m.direction = 'inbound') as customers_responded FROM v_lia_conversations_current v LEFT JOIN dp_messages m ON m.conversation_id = v.conversation_id WHERE {base_filter} """) # Calculate response rate total = stats_row["total_customers"] or 1 customers_responded = messages_row["customers_responded"] or 0 response_rate = round((customers_responded / total) * 100) if total > 0 else 0 # Get stage breakdown with message-based response tracking stage_rows = await conn.fetch(f""" SELECT v.stage, COUNT(*) as customers, COUNT(*) FILTER (WHERE EXISTS ( SELECT 1 FROM dp_messages m WHERE m.conversation_id = v.conversation_id AND m.direction = 'outbound' )) as contacted, COUNT(*) FILTER (WHERE EXISTS ( SELECT 1 FROM dp_messages m WHERE m.conversation_id = v.conversation_id AND m.direction = 'inbound' )) as responded, COUNT(*) FILTER (WHERE v.appointment_status = 'BOOKED') as appointments, 0 as sold FROM v_lia_conversations_current v WHERE {base_filter} AND v.stage IS NOT NULL GROUP BY v.stage ORDER BY CASE v.stage WHEN '18' THEN 1 WHEN '12' THEN 2 WHEN '9' THEN 3 WHEN '6' THEN 4 WHEN '3' THEN 5 ELSE 6 END """) # Format stage data stages = [] for row in stage_rows: stage_label = f"{row['stage']} Months" if row['stage'] else "Unknown" stages.append({ "name": stage_label, "customers": row["customers"], "contacted": row["contacted"], "responded": row["responded"], "appointments": row["appointments"], "sold": row["sold"] }) # Calculate awaiting response (contacted but not responded) messages_sent = messages_row["messages_sent"] or 0 total_customers = stats_row["total_customers"] or 0 awaiting_response = total_customers - customers_responded return { "stats": { "totalCustomers": total_customers, "messagesSent": messages_sent, "responseRate": response_rate, "appointmentsBooked": stats_row["appointments_booked"] or 0, "engaged": customers_responded, "awaitingResponse": awaiting_response, "noResponse": 0, "sold": 0, "declined": stats_row["declined"] or 0, "deferred": stats_row["deferred"] or 0 }, "stages": stages } finally: await release_conn(conn) @app.get("/api/lia/export/campaign-report", dependencies=[Depends(require_auth)]) async def api_lia_export_campaign_report(): """Export campaign report as CSV""" conn = await get_conn() try: base_filter = """ v.campaign IS NOT NULL AND v.campaign != 'inbound_reply' AND v.first_name IS NOT NULL AND v.first_name != '' AND v.first_name != 'Unknown' AND v.updated_at >= '2025-12-01' """ rows = await conn.fetch(f""" SELECT v.first_name, v.last_name, v.phone_mobile, v.email, v.stage, v.campaign, v.status, v.appointment_status, v.channel, v.updated_at, (SELECT COUNT(*) FROM dp_messages m WHERE m.conversation_id = v.conversation_id AND m.direction = 'outbound') as messages_sent, (SELECT COUNT(*) FROM dp_messages m WHERE m.conversation_id = v.conversation_id AND m.direction = 'inbound') as messages_received FROM v_lia_conversations_current v WHERE {base_filter} ORDER BY v.updated_at DESC """) import csv import io output = io.StringIO() writer = csv.writer(output) writer.writerow(['First Name', 'Last Name', 'Phone', 'Email', 'Stage', 'Campaign', 'Status', 'Appointment Status', 'Channel', 'Last Updated', 'Messages Sent', 'Messages Received']) for row in rows: writer.writerow([ row['first_name'], row['last_name'], row['phone_mobile'], row['email'], row['stage'], row['campaign'], row['status'], row['appointment_status'], row['channel'], row['updated_at'].strftime('%Y-%m-%d %H:%M') if row['updated_at'] else '', row['messages_sent'], row['messages_received'] ]) from fastapi.responses import Response return Response( content=output.getvalue(), media_type="text/csv", headers={"Content-Disposition": "attachment; filename=campaign_report.csv"} ) finally: await release_conn(conn) @app.get("/api/lia/export/appointments", dependencies=[Depends(require_auth)]) async def api_lia_export_appointments(): """Export appointments list as CSV""" conn = await get_conn() try: rows = await conn.fetch(""" SELECT v.first_name, v.last_name, v.phone_mobile, v.stage, v.campaign, v.channel, c.appointment_date_display, c.appointment_time_display, v.updated_at FROM v_lia_conversations_current v JOIN dp_conversations c ON c.conversation_id = v.conversation_id WHERE v.appointment_status = 'BOOKED' AND v.campaign IS NOT NULL AND v.campaign != 'inbound_reply' AND v.first_name IS NOT NULL AND v.first_name != '' AND v.first_name != 'Unknown' ORDER BY c.appointment_date_display, c.appointment_time_display """) import csv import io output = io.StringIO() writer = csv.writer(output) writer.writerow(['First Name', 'Last Name', 'Phone', 'Stage', 'Campaign', 'Channel', 'Appointment Date', 'Appointment Time', 'Last Updated']) for row in rows: writer.writerow([ row['first_name'], row['last_name'], row['phone_mobile'], row['stage'], row['campaign'], row['channel'], row['appointment_date_display'], row['appointment_time_display'], row['updated_at'].strftime('%Y-%m-%d %H:%M') if row['updated_at'] else '' ]) from fastapi.responses import Response return Response( content=output.getvalue(), media_type="text/csv", headers={"Content-Disposition": "attachment; filename=appointments.csv"} ) finally: await release_conn(conn) @app.get("/api/lia/export/message-history", dependencies=[Depends(require_auth)]) async def api_lia_export_message_history(): """Export message history as CSV""" conn = await get_conn() try: base_filter = """ v.campaign IS NOT NULL AND v.campaign != 'inbound_reply' AND v.first_name IS NOT NULL AND v.first_name != '' AND v.first_name != 'Unknown' AND v.updated_at >= '2025-12-01' """ rows = await conn.fetch(f""" SELECT v.first_name, v.last_name, v.phone_mobile, v.campaign, v.stage, m.direction, m.message_body, m.channel, m.created_at FROM v_lia_conversations_current v JOIN dp_messages m ON m.conversation_id = v.conversation_id WHERE {base_filter} ORDER BY v.conversation_id, m.created_at """) import csv import io output = io.StringIO() writer = csv.writer(output) writer.writerow(['First Name', 'Last Name', 'Phone', 'Campaign', 'Stage', 'Direction', 'Message', 'Channel', 'Sent At']) for row in rows: writer.writerow([ row['first_name'], row['last_name'], row['phone_mobile'], row['campaign'], row['stage'], row['direction'], row['message_body'], row['channel'], row['created_at'].strftime('%Y-%m-%d %H:%M') if row['created_at'] else '' ]) from fastapi.responses import Response return Response( content=output.getvalue(), media_type="text/csv", headers={"Content-Disposition": "attachment; filename=message_history.csv"} ) finally: await release_conn(conn) @app.get("/api/lia/conversations/{conversation_id}", dependencies=[Depends(require_auth)]) async def api_lia_conversation_detail(conversation_id: str): """Get conversation detail with messages - uses dp_conversations joined with dp_customer""" conn = await get_conn() try: conv_uuid = UUID(conversation_id) # Fetch conversation with customer info from dp_customer conv_row = await conn.fetchrow(""" SELECT c.conversation_id, c.customer_id, c.status, c.stage, c.campaign, c.appointment_status, c.appointment_date_display, c.appointment_time_display, c.soft_no_count, c.confusion_count, c.needs_human_review, c.defer_until, c.defer_reason, c.from_number, c.to_number, c.created_at, c.updated_at, cu.first_name, cu.last_name, cu.phone_mobile FROM dp_conversations c LEFT JOIN dp_customer cu ON c.customer_id = cu.customer_id WHERE c.conversation_id = $1 """, conv_uuid) if not conv_row: raise HTTPException(status_code=404, detail="Conversation not found") # Convert row to dict for safe access conv_dict = dict(conv_row) first_name = conv_dict.get("first_name") or "" last_name = conv_dict.get("last_name") or "" customer_name = f"{first_name} {last_name}".strip() or "Unknown" # Parse stage from the stage column stage_str = conv_dict.get("stage") or "" stage_value = None if stage_str and stage_str.isdigit(): stage_value = int(stage_str) # Build next appointment if booked next_appointment = None if conv_dict.get("appointment_status") == "BOOKED": next_appointment = { "date": conv_dict.get("appointment_date_display") or "TBC", "time": conv_dict.get("appointment_time_display") or "TBC", "type": "Visit", "status": "Booked", } customer = { "name": customer_name, "phone": conv_dict.get("phone_mobile") or conv_dict.get("from_number") or "", "vehicle_description": "", # Not available from dp_customer "reg": "", # Not available from dp_customer "stage": stage_value, "status": conv_dict.get("status"), "appointment_status": conv_dict.get("appointment_status"), "true_equity": None, # Not available from dp_customer "current_payment": None, # Not available from dp_customer "hot_flag": conv_dict.get("needs_human_review") or False, "next_appointment": next_appointment, "campaign_name": conv_dict.get("campaign"), "advisor_name": None, # Not in current schema "tco_payment": None, # Not in current schema "soft_no_count": conv_dict.get("soft_no_count") or 0, "confusion_count": conv_dict.get("confusion_count") or 0, "defer_until": conv_dict.get("defer_until").isoformat() if conv_dict.get("defer_until") else None, "defer_reason": conv_dict.get("defer_reason"), "past_appointments": [], # TODO: Fetch from appointments table "notes": [], # Will be populated below } # Fetch messages (excluding notes) msg_rows = await conn.fetch(""" SELECT message_id as id, sender, message_body as body, created_at as timestamp FROM dp_messages WHERE conversation_id = $1 AND (mode IS NULL OR mode != 'NOTE') ORDER BY created_at ASC """, conv_uuid) messages = [] for msg in msg_rows: messages.append({ "id": str(msg["id"]), "sender": msg["sender"], "body": msg["body"] or "", "timestamp": msg["timestamp"].isoformat() if msg["timestamp"] else None, }) # Fetch notes separately note_rows = await conn.fetch(""" SELECT message_id as id, message_body as content, created_at FROM dp_messages WHERE conversation_id = $1 AND mode = 'NOTE' ORDER BY created_at DESC """, conv_uuid) customer["notes"] = [ { "id": str(note["id"]), "content": note["content"] or "", "created_at": note["created_at"].isoformat() if note["created_at"] else None, } for note in note_rows ] return { "customer": customer, "messages": messages, } except ValueError: raise HTTPException(status_code=400, detail="Invalid conversation ID format") finally: await release_conn(conn) from pydantic import BaseModel as PydanticBaseModel class ManagerReplyRequest(PydanticBaseModel): body: str pause_lia: bool = False mark_resolved: bool = False @app.post("/api/lia/conversations/{conversation_id}/reply", dependencies=[Depends(require_auth)]) async def api_lia_manager_reply(conversation_id: str, payload: ManagerReplyRequest): """Send a manager reply to a conversation""" conn = await get_conn() try: conv_uuid = UUID(conversation_id) # Use transaction to ensure both operations succeed or fail together async with conn.transaction(): # Insert manager message first await conn.execute(""" INSERT INTO dp_messages ( conversation_id, direction, channel, sender, message_body, mode ) VALUES ($1, 'outbound', 'whatsapp', 'manager', $2, 'MANAGER_REPLY') """, conv_uuid, payload.body) # Update conversation status new_status = "paused" if payload.pause_lia else "awaiting_customer" if payload.mark_resolved: new_status = "active" if not payload.pause_lia else "paused" await conn.execute(""" UPDATE dp_conversations SET status = $1, updated_at = NOW(), needs_human_review = FALSE WHERE conversation_id = $2 """, new_status, conv_uuid) return {"success": True, "new_status": new_status} except ValueError: raise HTTPException(status_code=400, detail="Invalid conversation ID format") except Exception as e: print(f"❌ Manager reply failed: {e}") raise HTTPException(status_code=500, detail="Failed to save manager reply") finally: await release_conn(conn) @app.post("/api/lia/conversations/{conversation_id}/resolve", dependencies=[Depends(require_auth)]) async def api_lia_resolve_escalation(conversation_id: str): """Resolve an escalated conversation""" conn = await get_conn() try: conv_uuid = UUID(conversation_id) await conn.execute(""" UPDATE dp_conversations SET status = 'active', needs_human_review = FALSE, updated_at = NOW() WHERE conversation_id = $1 """, conv_uuid) return {"success": True} except ValueError: raise HTTPException(status_code=400, detail="Invalid conversation ID format") finally: await release_conn(conn) @app.post("/api/lia/conversations/{conversation_id}/resume", dependencies=[Depends(require_auth)]) async def api_lia_resume_conversation(conversation_id: str): """Resume a paused conversation""" conn = await get_conn() try: conv_uuid = UUID(conversation_id) await conn.execute(""" UPDATE dp_conversations SET status = 'active', updated_at = NOW() WHERE conversation_id = $1 """, conv_uuid) return {"success": True} except ValueError: raise HTTPException(status_code=400, detail="Invalid conversation ID format") finally: await release_conn(conn) class NoteRequest(PydanticBaseModel): content: str @app.post("/api/lia/conversations/{conversation_id}/notes", dependencies=[Depends(require_auth)]) async def api_lia_add_note(conversation_id: str, payload: NoteRequest): """Add a note to a conversation""" conn = await get_conn() try: conv_uuid = UUID(conversation_id) # Check if dp_notes table exists, if not store in dp_messages as system note # For now, we'll store notes as system messages in dp_messages await conn.execute(""" INSERT INTO dp_messages ( conversation_id, direction, channel, sender, message_body, mode ) VALUES ($1, 'internal', 'system', 'manager', $2, 'NOTE') """, conv_uuid, payload.content) return {"success": True} except ValueError: raise HTTPException(status_code=400, detail="Invalid conversation ID format") except Exception as e: print(f"❌ Failed to save note: {e}") raise HTTPException(status_code=500, detail="Failed to save note") finally: await release_conn(conn)