from pydantic import BaseModel from datetime import date, timedelta from uuid import UUID import uuid from app.db import ( get_conn, release_conn, get_active_conversation_for_number, get_customer_by_phone, check_dnc_conversation_for_number, ) from app.lia_rules import ( LIA_TIER_BEHAVIOUR, LIA_TONE_GUIDELINES, classify_inbound_tier, select_eot_intro, ) def normalize_phone_number(from_number: str) -> str: """ Normalize a phone number to E.164 format. Strips "whatsapp:" prefix if present. Examples: "whatsapp:+447972194025" -> "+447972194025" "+447972194025" -> "+447972194025" """ if from_number.startswith("whatsapp:"): return from_number[9:] # len("whatsapp:") = 9 return from_number VALID_PAYMENT_BANDS = (18, 12, 9, 6, 3) class LiaInbound(BaseModel): from_number: str body: str channel: str | None = "whatsapp" conversation_id: str | None = None payments_remaining: int | None = None class LiaReply(BaseModel): conversation_id: str mode: str status: str defer_until: date | None = None reply: str stage: str | None = None async def find_latest_open_conversation(conn, from_number: str, channel: str): """ Find the latest open (active) conversation for a given from_number + channel. Returns conversation row or None. """ row = await conn.fetchrow( """ SELECT conversation_id, customer_id, dealer_id, template_key, stage, status, defer_until, defer_reason, from_number, to_number, campaign FROM dp_conversations WHERE to_number = $1 AND channel = $2 AND status = 'active' ORDER BY updated_at DESC, created_at DESC LIMIT 1 """, from_number, channel, ) return row async def log_inbound_message( conn, conversation_id: UUID, channel: str, sender: str, message_body: str, template_key: str | None = None, mode: str = "inbound", ): """ Log an inbound customer message to dp_messages using existing schema. Uses: conversation_id, direction, channel, sender, message_body, template_key, mode """ await conn.execute( """ INSERT INTO dp_messages ( conversation_id, direction, channel, sender, message_body, template_key, mode ) VALUES ($1, 'inbound', $2, $3, $4, $5, $6) """, conversation_id, channel, sender, message_body, template_key, mode, ) async def log_outbound_message( conn, conversation_id: UUID, channel: str, message_body: str, template_key: str | None, mode: str = "outbound", ): """ Log an outbound Lia reply to dp_messages using existing schema. Uses: conversation_id, direction, channel, sender, message_body, template_key, mode """ 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) """, conversation_id, channel, message_body, template_key, mode, ) async def get_last_outbound_mode(conn, conversation_id: UUID) -> str | None: """ Get the mode of the most recent outbound message from LIA. Used to determine if we're in a slot selection flow. """ row = await conn.fetchrow( """ SELECT mode FROM dp_messages WHERE conversation_id = $1 AND direction = 'outbound' AND sender = 'lia' ORDER BY created_at DESC LIMIT 1 """, conversation_id, ) return row["mode"] if row else None async def update_conversation_status( conn, conversation_id: UUID, stage: str | None = None, status: str | None = None, defer_until: date | None = None, defer_reason: str | None = None, ): """ Update dp_conversations with stage and optionally status/defer info. Note: template_key is NOT updated here since it has a FK constraint on dp_templates. Template tracking is done at the message level instead. """ if status and status in ('paused', 'dnc'): await conn.execute( """ UPDATE dp_conversations SET stage = $2, status = $3, defer_until = $4, defer_reason = $5, updated_at = now() WHERE conversation_id = $1 """, conversation_id, stage, status, defer_until, defer_reason, ) else: await conn.execute( """ UPDATE dp_conversations SET stage = $2, updated_at = now() WHERE conversation_id = $1 """, conversation_id, stage, ) def select_reply_template(stage: str | None, current_template_key: str | None, body_lower: str) -> tuple[str, str]: """ Choose the next reply template based on stage, current_template_key, and customer message. Returns (reply_text, new_template_key). """ tier = classify_inbound_tier(body_lower) behaviour = LIA_TIER_BEHAVIOUR.get(tier, LIA_TIER_BEHAVIOUR["T1_GENERAL"]) if current_template_key and current_template_key.startswith("EOT_"): if stage: intro_key = f"EOT_{stage}_INTRO" intro_text = select_eot_intro_by_stage(stage, current_template_key) if intro_text: return intro_text, intro_key return behaviour["example_reply"], f"TIER_{tier}" def select_eot_intro_by_stage(stage: str, current_template_key: str) -> str | None: """ Select the appropriate EOT intro template based on stage and whether NEW or USED. """ from app.lia_rules import ( EOT_18_INTRO, EOT_18_INTRO_USED, EOT_12_INTRO_NEW, EOT_12_INTRO_USED, EOT_9_INTRO_NEW, EOT_9_INTRO_USED, EOT_6_INTRO_NEW, EOT_6_INTRO_USED, EOT_3_INTRO_NEW, EOT_3_INTRO_USED, ) is_used = "USED" in current_template_key.upper() intros = { "18": (EOT_18_INTRO, EOT_18_INTRO_USED), "12": (EOT_12_INTRO_NEW, EOT_12_INTRO_USED), "9": (EOT_9_INTRO_NEW, EOT_9_INTRO_USED), "6": (EOT_6_INTRO_NEW, EOT_6_INTRO_USED), "3": (EOT_3_INTRO_NEW, EOT_3_INTRO_USED), } if stage in intros: new_intro, used_intro = intros[stage] return used_intro if is_used else new_intro return None async def process_inbound_message(payload: LiaInbound) -> LiaReply: """ Process an inbound message from a customer via Make.com using URE. Flow: 1. Normalize phone number and find/create conversation 2. If conversation_id provided: use that conversation 3. If no conversation_id: look up by phone number using get_active_conversation_for_number 4. If still no conversation: create a new one 5. Build customer_state dict for URE 6. Call generate_lia_reply from URE 7. Optionally enhance with LLM layer (if LIA_LLM_ENABLED=true) 8. Log inbound and outbound messages 9. Update conversation with new URE state 10. Return reply Phone number handling: - Strips "whatsapp:" prefix from from_number - Looks up dp_customer by phone_mobile in E.164 format (+44...) LLM Integration: - When LLM is enabled (LIA_LLM_ENABLED=true), the URE template is passed through the LLM layer for natural language variation - tone_tag and intent_tag are added to updated_state for analytics - External response shape is UNCHANGED """ from app.lia_ure import generate_lia_reply from app.lia_outbound import DNCBlockedError from app.lia_llm import generate_llm_reply_async, is_llm_enabled DEALER_ID = "451dbf4d-ff8b-4f45-9ace-0405cf73aaff" LIA_FROM_NUMBER = "+447700000000" # Normalize phone number: strip "whatsapp:" prefix if present phone_e164 = normalize_phone_number(payload.from_number) print(f"📥 Inbound: from_number={payload.from_number} -> phone_e164={phone_e164}") conn = await get_conn() try: channel = payload.channel or "whatsapp" # ------------------------------------------------------------------------- # STEP 1: Find existing conversation # ------------------------------------------------------------------------- row = None # Path A: conversation_id explicitly provided - use it directly if payload.conversation_id: print(f" Path A: Using provided conversation_id={payload.conversation_id}") conv_uuid = UUID(payload.conversation_id) row = await conn.fetchrow( """ SELECT c.conversation_id, c.customer_id, c.dealer_id, c.template_key, c.stage, c.status, c.defer_until, c.defer_reason, c.from_number, c.to_number, c.campaign, c.soft_no_count, c.confusion_count, c.needs_human_review, c.appointment_status, c.last_reply_type, c.pending_slot_options, c.pending_slot_context, c.pending_day_display, c.pending_day_date, c.appointment_date_display, c.appointment_time_display, cust.first_name, cust.last_name FROM dp_conversations c JOIN dp_customer cust ON c.customer_id = cust.customer_id WHERE c.conversation_id = $1 """, conv_uuid, ) if not row: raise ValueError(f"Conversation {payload.conversation_id} not found") else: # Path B: No conversation_id - look up by phone number print(f" Path B: Looking up conversation by phone_e164={phone_e164}") # First check if customer has ANY DNC conversation (global DNC check) is_dnc = await check_dnc_conversation_for_number(phone_e164) if is_dnc: print(f" ❌ Customer has DNC status - blocking inbound") raise DNCBlockedError("Customer has DNC status - no further contact allowed") # Look up most recent non-DNC conversation for this phone number conv_lookup = await get_active_conversation_for_number(phone_e164) if conv_lookup: print(f" ✅ Found existing conversation: {conv_lookup['conversation_id']}, stage={conv_lookup['stage']}") # Re-fetch with same structure as Path A for consistency row = await conn.fetchrow( """ SELECT c.conversation_id, c.customer_id, c.dealer_id, c.template_key, c.stage, c.status, c.defer_until, c.defer_reason, c.from_number, c.to_number, c.campaign, c.soft_no_count, c.confusion_count, c.needs_human_review, c.appointment_status, c.last_reply_type, c.pending_slot_options, c.pending_slot_context, c.pending_day_display, c.pending_day_date, c.appointment_date_display, c.appointment_time_display, cust.first_name, cust.last_name FROM dp_conversations c JOIN dp_customer cust ON c.customer_id = cust.customer_id WHERE c.conversation_id = $1 """, conv_lookup["conversation_id"], ) # ------------------------------------------------------------------------- # STEP 2: Extract conversation state or create new conversation # ------------------------------------------------------------------------- if row: conversation_id = row["conversation_id"] customer_id = row["customer_id"] current_status = row["status"] or "active" # Default to active if None current_stage = row["stage"] soft_no_count = row["soft_no_count"] or 0 confusion_count = row["confusion_count"] if row["confusion_count"] is not None else 0 needs_human_review = row["needs_human_review"] if row["needs_human_review"] is not None else False appointment_status = row["appointment_status"] or "NONE" first_name = row["first_name"] last_name = row["last_name"] # Slot selection state fields last_reply_type = row["last_reply_type"] pending_slot_options = list(row["pending_slot_options"]) if row["pending_slot_options"] else None pending_slot_context = row["pending_slot_context"] pending_day_display = row["pending_day_display"] pending_day_date = str(row["pending_day_date"]) if row["pending_day_date"] else None appointment_date_display = row["appointment_date_display"] appointment_time_display = row["appointment_time_display"] # When reusing an existing conversation, ignore payments_remaining from payload # Use the stage already stored in the conversation active_stage = current_stage print(f" Using existing conversation state: stage={active_stage}, status={current_status}, last_reply_type={last_reply_type}") else: # No existing conversation - create a new one print(f" No existing conversation found - creating new one") # Find or create customer cust_row = await conn.fetchrow( """ SELECT customer_id, phone_mobile, first_name, last_name FROM dp_customer WHERE phone_mobile = $1 LIMIT 1 """, phone_e164, ) if cust_row: customer_id = cust_row["customer_id"] first_name = cust_row["first_name"] last_name = cust_row["last_name"] else: new_cust = await conn.fetchrow( """ INSERT INTO dp_customer (phone_mobile, first_name) VALUES ($1, 'Unknown') RETURNING customer_id, first_name, last_name """, phone_e164, ) customer_id = new_cust["customer_id"] first_name = new_cust["first_name"] last_name = new_cust["last_name"] # Determine stage for new conversation # Use payments_remaining if provided and valid, otherwise "unknown" if payload.payments_remaining is not None and payload.payments_remaining in VALID_PAYMENT_BANDS: active_stage = str(payload.payments_remaining) else: active_stage = "unknown" print(f" Creating new conversation with stage={active_stage}") new_conv = await conn.fetchrow( """ INSERT INTO dp_conversations ( customer_id, dealer_id, channel, campaign, stage, from_number, to_number, status, soft_no_count, appointment_status ) VALUES ($1, $2, $3, 'inbound_reply', $4, $5, $6, 'active', 0, 'NONE') RETURNING conversation_id """, customer_id, UUID(DEALER_ID), channel, active_stage, LIA_FROM_NUMBER, phone_e164, ) conversation_id = new_conv["conversation_id"] current_status = "active" current_stage = active_stage soft_no_count = 0 confusion_count = 0 needs_human_review = False appointment_status = "NONE" last_reply_type = None pending_slot_options = None pending_slot_context = None pending_day_display = None pending_day_date = None appointment_date_display = None appointment_time_display = None customer_name = (first_name or '').strip() or None # If last_reply_type wasn't stored in DB, fall back to fetching from messages if not last_reply_type: last_outbound_mode = await get_last_outbound_mode(conn, conversation_id) print(f" Last outbound mode (from messages): {last_outbound_mode}") else: last_outbound_mode = last_reply_type print(f" Last outbound mode (from DB): {last_outbound_mode}") customer_state = { "customer_id": str(customer_id), "customer_name": customer_name, "stage": active_stage, "soft_no_count": soft_no_count, "confusion_count": confusion_count, "needs_human_review": needs_human_review, "appointment_status": appointment_status, "opted_out": (current_status == "dnc"), "last_message": None, "last_reply_type": last_outbound_mode, "pending_slot_options": pending_slot_options, "pending_slot_context": pending_slot_context, "pending_day_display": pending_day_display, "pending_day_date": pending_day_date, "appointment_date_display": appointment_date_display, "appointment_time_display": appointment_time_display, } await log_inbound_message( conn=conn, conversation_id=conversation_id, channel=channel, sender="customer", message_body=payload.body, template_key=None, mode="inbound", ) reply_text, updated_state = generate_lia_reply(customer_state, payload.body) mode = updated_state.get("last_reply_type", "general") if is_llm_enabled() and reply_text: tier = mode if mode.startswith("T") else "T1_GENERAL" template_key = mode llm_reply, llm_state = await generate_llm_reply_async( customer_state=customer_state, inbound_text=payload.body, tier=tier, template_key=template_key, template_text=reply_text, recent_messages=[], ) if llm_reply: reply_text = llm_reply updated_state["tone_tag"] = llm_state.get("tone_tag") updated_state["intent_tag"] = llm_state.get("intent_tag") updated_state["last_reply_type"] = llm_state.get("last_reply_type", mode) print(f" LLM enhanced reply: tone={updated_state.get('tone_tag')}, intent={updated_state.get('intent_tag')}") new_soft_no_count = updated_state.get("soft_no_count", 0) new_confusion_count = updated_state.get("confusion_count", 0) new_needs_human_review = updated_state.get("needs_human_review", False) new_appointment_status = updated_state.get("appointment_status", "NONE") opted_out = updated_state.get("opted_out", False) escalation_reason = None # Slot selection state fields new_last_reply_type = updated_state.get("last_reply_type") new_pending_slot_options = updated_state.get("pending_slot_options") new_pending_slot_context = updated_state.get("pending_slot_context") new_pending_day_display = updated_state.get("pending_day_display") new_pending_day_date_raw = updated_state.get("pending_day_date") new_appointment_date_display = updated_state.get("appointment_date_display") new_appointment_time_display = updated_state.get("appointment_time_display") # Convert pending_day_date from string to date object if needed if isinstance(new_pending_day_date_raw, str): try: new_pending_day_date = date.fromisoformat(new_pending_day_date_raw) except ValueError: new_pending_day_date = None elif isinstance(new_pending_day_date_raw, date): new_pending_day_date = new_pending_day_date_raw else: new_pending_day_date = None # Read status, defer_until, defer_reason from URE updated_state if set ure_status = updated_state.get("status") ure_defer_until = updated_state.get("defer_until") ure_defer_reason = updated_state.get("defer_reason") ure_escalation_reason = updated_state.get("escalation_reason") # Set escalation_reason from URE if provided if ure_escalation_reason: escalation_reason = ure_escalation_reason # Priority: DNC > URE status > current_status if opted_out: status = "dnc" defer_until = None defer_reason = None elif ure_status: status = ure_status defer_until = ure_defer_until defer_reason = ure_defer_reason else: status = current_status defer_until = None defer_reason = None # Legacy T2_SOFT_NO handling (fallback if URE didn't set status) if "T2_SOFT_NO" in mode and not ure_status: if new_soft_no_count >= 2: status = "paused" defer_until = date.today() + timedelta(days=60) defer_reason = "Two soft objections - parked for 60 days" if new_confusion_count == 1: new_needs_human_review = True escalation_reason = "unrecognised_message" elif new_confusion_count >= 2: status = "paused" new_needs_human_review = True escalation_reason = "unrecognised_message" if reply_text: await log_outbound_message( conn=conn, conversation_id=conversation_id, channel=channel, message_body=reply_text, template_key=mode, mode=mode, ) await conn.execute( """ UPDATE dp_conversations SET stage = $1, status = $2, soft_no_count = $3, confusion_count = $4, needs_human_review = $5, escalation_reason = $6::text, escalation_at = CASE WHEN $6 IS NOT NULL THEN NOW() ELSE escalation_at END, appointment_status = $7, defer_until = $8, defer_reason = $9::text, last_reply_type = $10, pending_slot_options = $11, pending_slot_context = $12, pending_day_display = $13, pending_day_date = $14::date, appointment_date_display = $15, appointment_time_display = $16, updated_at = NOW() WHERE conversation_id = $17 """, active_stage, status, new_soft_no_count, new_confusion_count, new_needs_human_review, escalation_reason, new_appointment_status, defer_until, defer_reason, new_last_reply_type, new_pending_slot_options, new_pending_slot_context, new_pending_day_display, new_pending_day_date, new_appointment_date_display, new_appointment_time_display, conversation_id, ) return LiaReply( conversation_id=str(conversation_id), mode=mode, status=status, defer_until=defer_until, reply=reply_text, stage=active_stage, ) except Exception as e: print(f"Error processing inbound message: {e}") raise finally: await release_conn(conn)