""" Lia Universal Reply Engine (URE) ================================= Campaign-agnostic reply generator for WhatsApp conversations. Handles small talk, direct questions, appointments, objections, and tier-based responses. Brand: Lincoln Audi (UK main dealer) Channel: WhatsApp via Twilio Tone: Warm, professional, Audi-quality, UK English """ import re from datetime import datetime, timezone, timedelta, date from typing import Tuple, Dict, Any from app.lia_rules import LIA_SOFT_NO_DEFER_TEMPLATES # ============================================================================= # CONSTANTS & TEMPLATES # ============================================================================= # Small talk keywords SMALL_TALK_KEYWORDS = [ "hi lia", "hello lia", "hi, lia", "hi it's", "hi its", "morning", "good morning", "afternoon", "good afternoon", "evening", "good evening", "thanks", "thank you", "cheers", ] # Direct question keywords DIRECT_QUESTION_KEYWORDS = [ "when are you open", "what time are you open", "opening times", "opening hours", "what are your hours", "are you open today", "when can i come in", "when can i come", "when can i pop in", "what time do you close", "how long will it take", "how long does it take", "how long does a visit take", "how long is a visit", ] # Appointment confirmation keywords DAY_WORDS = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"] TIME_OF_DAY = ["morning", "afternoon", "evening"] RELATIVE_WORDS = ["tomorrow", "this weekend", "this week", "next week", "later today"] TIME_REGEXES = [ re.compile(r"\b([01]?\d|2[0-3]):[0-5]\d\b"), # 14:30, 9:05 re.compile(r"\b([1-9]|1[0-2])\s?(am|pm)\b"), # 9am, 10 pm ] # Legacy compatibility DAY_KEYWORDS = DAY_WORDS + RELATIVE_WORDS def extract_day_display(text: str) -> str: """ Extract a human-friendly day display from appointment context text. Returns a capitalized day name or relative time phrase. Examples: - "saturday" → "Saturday" - "tomorrow morning" → "Tomorrow" - "this weekend" → "This Weekend" - "next week" → "Next Week" """ text_lower = text.lower() # Check for day words first (Monday-Sunday) for day in DAY_WORDS: if day in text_lower: return day.capitalize() # Check for relative words for relative in RELATIVE_WORDS: if relative in text_lower: # Capitalize each word for display return " ".join(word.capitalize() for word in relative.split()) return "" def compute_day_date(text: str, today: date = None) -> date | None: """ Compute a concrete calendar date from a day phrase. Args: text: The customer's message containing a day reference today: Reference date (defaults to today if not provided) Returns: A date object, or None if no date could be computed Rules: - "tomorrow" → today + 1 day - "saturday" (bare day) → next occurrence (0-6 days ahead) - "next saturday" → following week's occurrence (7-13 days ahead) - "this weekend" → this Saturday - "next week" → Monday of next week """ if today is None: today = date.today() text_lower = text.lower() # Map day names to weekday numbers (0=Monday, 6=Sunday) day_to_weekday = { "monday": 0, "tuesday": 1, "wednesday": 2, "thursday": 3, "friday": 4, "saturday": 5, "sunday": 6 } # Check for "tomorrow" if "tomorrow" in text_lower: return today + timedelta(days=1) # Check for "this weekend" → this Saturday if "this weekend" in text_lower: days_until_saturday = (5 - today.weekday()) % 7 if days_until_saturday == 0: days_until_saturday = 7 # If today is Saturday, go to next week return today + timedelta(days=days_until_saturday) # Check for "next week" → Monday of next week if "next week" in text_lower: days_until_monday = (7 - today.weekday()) % 7 if days_until_monday == 0: days_until_monday = 7 # Ensure we go to NEXT week's Monday return today + timedelta(days=days_until_monday) # Check for "later today" if "later today" in text_lower: return today # Check for specific days (Monday-Sunday) for day_name, weekday in day_to_weekday.items(): if day_name in text_lower: # Check if "next" prefix is present has_next = re.search(rf"\bnext\s+{day_name}\b", text_lower) # Calculate days until this day days_ahead = (weekday - today.weekday()) % 7 if has_next: # "next Saturday" → following week (7-13 days ahead) days_ahead += 7 elif days_ahead == 0: # If it's the same day, assume they mean next week days_ahead = 7 return today + timedelta(days=days_ahead) return None def format_date_uk(d: date) -> str: """Format a date in UK style: dd/MM/yyyy""" if d is None: return "" return d.strftime("%d/%m/%Y") # ============================================================================= # CHRISTMAS / NEW YEAR DEFERRAL DETECTION # ============================================================================= CHRISTMAS_DEFER_PATTERNS = [ "after christmas", "after xmas", "after the christmas", "after new year", "after the new year", "in the new year", "next year", "in january", "after january", "post christmas", "post xmas", "post new year", "when christmas is over", "once christmas is done", "when the holidays are over", "after the holidays", "after the festive", "post holidays", ] def is_clear_defer_request(text: str) -> bool: """ Detect if the message is a clear deferral request (e.g., "after Christmas"). Returns True if the text contains Christmas/New Year deferral phrases. """ text_lower = text.lower() return any(pattern in text_lower for pattern in CHRISTMAS_DEFER_PATTERNS) def compute_defer_date(text: str, today: date = None) -> date: """ Compute a sensible defer_until date based on the deferral phrase. Args: text: The customer's message today: Reference date (defaults to today if not provided) Returns: A date object for when to re-contact the customer Rules: - "after Christmas" / "after Xmas" → January 6th (after 12th Night) - "in the new year" / "after new year" → January 6th - "next year" / "in January" → January 10th - Default: January 6th of the next year if before Dec 25, else Jan 6 current year+1 """ if today is None: today = date.today() text_lower = text.lower() # Determine the target year # If we're in late year (Oct-Dec), defer to next year # If early year, defer later in January if today.month >= 10: target_year = today.year + 1 else: target_year = today.year # "in january" or "next year" → Jan 10th gives them time to settle if "in january" in text_lower or "next year" in text_lower: return date(target_year, 1, 10) # Most Christmas/New Year references → Jan 6th (after 12th Night) return date(target_year, 1, 6) # ============================================================================= # MONEY CONCERN DETECTION # ============================================================================= MONEY_CONCERN_PATTERNS = [ "can't afford", "cant afford", "cannot afford", "too expensive", "expensive right now", "payments too high", "monthly payments", "money's tight", "money is tight", "moneys tight", "budget is tight", "tight budget", "worried about cost", "worried about the cost", "cost too much", "costs too much", "afford to change", "afford it", "afford right now", "financially", "financial situation", "struggling", "can't stretch", "cant stretch", "not in a position", "out of budget", "bills", "cost of living", ] def is_money_concern(text: str) -> bool: """ Detect if the message expresses a money/affordability concern. Returns True if the text contains money-related concern phrases. """ text_lower = text.lower() return any(pattern in text_lower for pattern in MONEY_CONCERN_PATTERNS) # ============================================================================= # SIMPLE ACKNOWLEDGEMENT DETECTION # ============================================================================= def is_simple_acknowledgement(text: str) -> bool: """ Detect if message is a simple acknowledgement like "Thanks", "Cheers", etc. Used after decisions (defer, soft-no, opt-out) to prevent re-pitching. """ t = text.lower().strip() # Short messages only (avoid matching "thanks but I'd like to book") if len(t) > 50: return False ack_patterns = [ "thanks", "thank you", "thankyou", "ty", "thx", "cheers", "ta", "perfect", "brilliant", "great", "lovely", "fab", "cool", "ok", "okay", "👍", "👌", "🙏", "will do", "no worries", "no problem", "np", "noted", "sounds good", "that's fine", "thats fine", "fine", ] return any(w in t for w in ack_patterns) # ============================================================================= # SETTLEMENT / BALANCE QUERY DETECTION # ============================================================================= SETTLEMENT_QUERY_PATTERNS = [ "settlement", "settle", "pay off", "payoff", "pay it off", "balance", "how much do i owe", "how much do i still owe", "how much is left", "how much left", "what do i owe", "final payment", "balloon", "balloon payment", "remaining balance", "outstanding", "outstanding balance", "buy it out", "buyout", "buy out", "what's the figure", "whats the figure", "what is the figure", "early repayment", "early settlement", "redemption", "how much to finish", "how much to end", ] def is_settlement_query(text: str) -> bool: """ Detect if message is asking about settlement/balance/how much they owe. Returns True if text contains settlement-related questions. Does not trigger on complaints or opt-outs. """ t = text.lower().strip() # Avoid triggering on complaints or opt-outs complaint_words = ["complaint", "complain", "unhappy", "disgusted", "terrible"] optout_words = ["stop", "unsubscribe", "remove me", "don't contact"] if any(w in t for w in complaint_words): return False if any(w in t for w in optout_words): return False return any(pattern in t for pattern in SETTLEMENT_QUERY_PATTERNS) # ============================================================================= # SAME-DAY VAGUE TIME DETECTION ("this afternoon", "free all afternoon", etc.) # ============================================================================= SAME_DAY_VAGUE_PATTERNS = [ "this afternoon", "this morning", "this evening", "later today", "today if", "free today", "free all afternoon", "free this afternoon", "free all morning", "around this afternoon", "around this morning", "available this afternoon", "available this morning", "i'm free today", "im free today", "i'm around today", "im around today", "anytime today", "any time today", "sometime today", "some time today", ] def is_same_day_vague_time(text: str) -> bool: """ Detect if message indicates same-day availability without a specific time. Examples: "I'm free all afternoon", "this afternoon works", "later today" Returns True if text contains same-day vague time patterns. """ t = text.lower().strip() return any(pattern in t for pattern in SAME_DAY_VAGUE_PATTERNS) def generate_same_day_slots(now: datetime = None) -> list: """ Generate 2 realistic same-day time slots based on current time. Rules: - If now < 14:00 → offer 15:00 and 16:00 - If now 14:00-16:00 → offer next whole hour and +30 mins - If after 16:30 → return empty list (too late) Returns list of formatted time strings like ["3:00pm", "4:00pm"] """ if now is None: now = datetime.now() current_hour = now.hour current_minute = now.minute # Too late in the day (after 16:30) - no slots available if current_hour > 16 or (current_hour == 16 and current_minute > 30): return [] # Morning or early afternoon (before 14:00) - offer 3pm and 4pm if current_hour < 14: return ["3:00pm", "4:00pm"] # Afternoon (14:00-16:00) - offer next whole hour and +30 mins # Round up to next hour next_hour = current_hour + 1 if next_hour < 17: slot1 = f"{next_hour - 12}:00pm" if next_hour > 12 else f"{next_hour}:00pm" slot2 = f"{next_hour - 12}:30pm" if next_hour > 12 else f"{next_hour}:30pm" return [slot1, slot2] # Fallback - offer 4pm and 4:30pm if still possible if current_hour < 16: return ["4:00pm", "4:30pm"] return [] def extract_time_display(text: str) -> str: """ Extract a human-friendly time display from customer's message. Returns formatted time string or raw text if no time found. Examples: - "11am please" → "11:00am" - "the 2pm slot" → "2:00pm" - "10:30" → "10:30am" - "half 9" → "9:30am" """ text_lower = text.lower().strip() # Pattern: "X:XXam/pm" (e.g., "10:30am", "2:00pm") match = re.search(r"(\d{1,2}):(\d{2})\s?(am|pm)", text_lower) if match: hour, minute, period = match.groups() return f"{hour}:{minute}{period}" # Pattern: "X am/pm" (e.g., "11am", "2 pm") match = re.search(r"\b(\d{1,2})\s?(am|pm)\b", text_lower) if match: hour, period = match.groups() return f"{hour}:00{period}" # Pattern: "half X" (e.g., "half 9" = 9:30) match = re.search(r"half\s*(\d{1,2})", text_lower) if match: hour = match.group(1) period = "am" if int(hour) < 12 else "pm" return f"{hour}:30{period}" # Pattern: "quarter to X" (e.g., "quarter to 3" = 2:45) match = re.search(r"quarter\s+to\s+(\d{1,2})", text_lower) if match: hour = int(match.group(1)) - 1 if hour < 1: hour = 12 period = "am" if hour < 12 else "pm" return f"{hour}:45{period}" # Pattern: "quarter past X" (e.g., "quarter past 10" = 10:15) match = re.search(r"quarter\s+past\s+(\d{1,2})", text_lower) if match: hour = match.group(1) period = "am" if int(hour) < 12 else "pm" return f"{hour}:15{period}" # Pattern: Just a number in context (e.g., "the 11", "11 please") match = re.search(r"\b(\d{1,2})\b", text_lower) if match: hour = int(match.group(1)) # Assume AM for morning hours (9-11), PM for afternoon (1-5) if hour >= 9 and hour <= 11: return f"{hour}:00am" elif hour >= 1 and hour <= 5: return f"{hour}:00pm" elif hour == 12: return "12:00pm" # Fallback: return "your chosen time" if we can't parse return "your chosen time" # Phrases indicating difficulty visiting CANT_VISIT_PHRASES = ["can't get there", "cant get there", "live too far", "too far away", "no time to come", "can't come in", "cant come in"] # Urgency keywords for service issues URGENT_KEYWORDS = ["broken down", "urgent", "asap", "warning light", "immediately", "straight away"] # ============================================================================= # REPLY TEMPLATES # ============================================================================= SMALL_TALK_REPLY = ( "Hi [customer_name], it's Lia at Lincoln Audi.\n\n" "Thanks for coming back to me. Right now is a good time to review things because " "your Audi holds strong value and you've still got your existing-customer savings available.\n\n" "The simplest next step is a quick visit so we can show you your options clearly.\n" "What day normally works best for you to pop in?" ) DIRECT_QUESTION_REPLY = ( "No problem, [customer_name]. We're open:\n" "• Monday–Friday: 8:30am–6:00pm\n" "• Saturday: 9:00am–5:00pm\n" "• Sunday: closed\n\n" "A quick visit is usually 10–15 minutes, depending on what you'd like to do.\n\n" "What day would suit you best to pop in?" ) APPOINTMENT_CONFIRM_REPLY = ( "Perfect, [customer_name] – I've made a note of that.\n\n" "If anything changes before then, just drop me a quick message here." ) T1_GENERAL_REPLY = ( "Great to hear from you, [customer_name].\n\n" "You're in a strong position at the moment – your Audi's value is good and you've got " "unclaimed loyalty savings reserved as an existing Audi customer.\n\n" "The easiest way to see what this actually looks like is a quick visit to Lincoln Audi so we can " "check your figures and show you the best options while everything is still available.\n\n" "What day is usually easiest for you to pop in?" ) T5_MONEY_REPLY = ( "I completely understand, [customer_name] – the monthly payment has to feel comfortable or it's not the right move.\n\n" "What often helps is looking at your Audi's actual value and the extra savings you get as an existing customer. " "Many people find the figures look better once we've checked everything properly, rather than guessing over WhatsApp.\n\n" "Could you come into the showroom one day next week so we can run through it clearly for you?" ) T5_MONEY_PHONE_FALLBACK = ( "I completely understand, [customer_name]. In that case, we can go through the options on a short call instead so you've got a clear picture.\n\n" "When is usually a good time of day for you to take a quick call?" ) T2_SOFT_NO_FIRST = ( "No problem at all, [customer_name] – I understand it might not feel urgent right now.\n\n" "Just so you know, you're in a good position at the moment because your Audi's value is strong and you've still got " "your existing-customer savings available.\n\n" "Would a weekday or a Saturday generally work better for you if you did decide to pop in?" ) T2_SOFT_NO_SECOND = ( "That's absolutely fine, [customer_name] – thanks for letting me know.\n\n" "I'll leave things with you for now. If anything changes or you'd like to look at your options later on, " "you're always welcome to message me here." ) T4_SCHEDULED_REPLY = ( "No problem, [customer_name] – I'll make a note to get in touch nearer the time you've mentioned.\n\n" "If you'd like to look at anything sooner, just send me a quick message here." ) T3_HARD_STOP_REPLY = ( "Of course, [customer_name] – I'll update your preferences now and won't contact you again about this.\n\n" "If you ever need anything from Lincoln Audi in the future, you can still message us here." ) T6_SUPPORT_REPLY = ( "Thanks for letting me know, [customer_name]. I'll pass this straight to our service team so they can help.\n\n" "While we're looking into it, if you'd also like to review your agreement or future options, I can help with that too." ) CLARITY_FIRST_REPLY = ( "Sorry, I'm not quite sure I've followed that.\n\n" "Could you say that again in a slightly different way, or let me know if you're asking about your options, a booking, or something else?" ) T6_SUPPORT_URGENT_ADDON = ( "\n\nIf it's urgent, you can also call us on 01522 540000 and the team will pick it up straight away." ) # ============================================================================= # HELPER FUNCTIONS # ============================================================================= def normalise_text(text: str) -> str: """Normalise incoming text.""" return text.strip() def personalise_reply(reply: str, customer_name: str | None) -> str: """Replace [customer_name] placeholder with actual name or fallback.""" if customer_name: first_name = customer_name.split()[0] if customer_name else "" return reply.replace("[customer_name]", first_name) else: return reply.replace("[customer_name], ", "").replace("[customer_name]", "there") def is_small_talk(body_lower: str) -> bool: """ Detect if message is small talk. Criteria: - Message is short (< 80 chars) - Contains small talk keywords - Does NOT contain strong question words - Does NOT look like appointment intent IMPORTANT: This function is CONSERVATIVE - we prioritise detecting appointment intent over small talk. A message like "How about in the morning?" should go to the appointment flow, not be treated as small talk just because it has "morning". """ if len(body_lower) >= 80: return False # Check for small talk keywords has_small_talk = any(keyword in body_lower for keyword in SMALL_TALK_KEYWORDS) if not has_small_talk: return False # Exclude if it has strong question words strong_questions = ["when", "how much", "what time", "how long", "what options"] has_strong_question = any(q in body_lower for q in strong_questions) if has_strong_question: return False # CRITICAL: Exclude if it looks like appointment scheduling intent # This prevents "How about in the morning?" from being misclassified as small talk scheduling_phrases = [ "how about", "what about", "could do", "can do", "would work", "works for", "suits", "prefer", "i'm free", "im free", "i could", "i can", "maybe", "perhaps", "ideally", ] has_scheduling_intent = any(phrase in body_lower for phrase in scheduling_phrases) if has_scheduling_intent: return False # Exclude if message has a day word (suggests appointment context) has_day_word = any(d in body_lower for d in DAY_WORDS) or any(r in body_lower for r in RELATIVE_WORDS) if has_day_word: return False # For time-of-day words (morning/afternoon/evening), only treat as small talk # if it's a pure greeting like "Good morning" or bare "Morning" with no scheduling context time_of_day_in_message = any(t in body_lower for t in TIME_OF_DAY) if time_of_day_in_message: # Allow "Good morning" style greetings pure_greetings = ["good morning", "good afternoon", "good evening"] if any(g in body_lower for g in pure_greetings): return True # Allow bare time-of-day words (1-2 words) as greetings # e.g., "Morning", "Morning!", "Hi morning", "Hello afternoon" words = body_lower.strip().split() if len(words) <= 2: greeting_prefixes = ["hi", "hello", "hey", "hiya"] non_greeting_words = [w for w in words if w.rstrip("!?.") not in greeting_prefixes and w.rstrip("!?.") not in TIME_OF_DAY] if not non_greeting_words: return True # It's a bare greeting like "Morning" or "Hi morning" # But NOT "How about in the morning?" or "Morning works" (has scheduling context) return False return True def is_direct_question(body_lower: str) -> bool: """Detect if message is a direct practical question about opening hours, visit duration, etc.""" return any(keyword in body_lower for keyword in DIRECT_QUESTION_KEYWORDS) def is_specific_time(body_lower: str) -> bool: """ Detect if message contains a SPECIFIC appointment time. A message is SPECIFIC if it contains: - A clock pattern like "10:30", "9:00", "14:15", "11:00am" - A number + am/pm (e.g. "10am", "3 pm") - A day word plus "at" followed by a number (e.g. "Saturday at 10") - Colloquial time expressions (e.g. "half 9", "half past 10", "quarter to 3") Examples: - "How about Saturday at 10:30?" → True - "9am works for me" → True - "How about 11:00am?" → True - "Tuesday at 3" → True - "half 9" → True - "about 10:30" → True - "Tuesday afternoon" → False (use is_vague_time_window) """ # Clock pattern with optional am/pm: 10:30, 9:00, 14:15, 11:00am, 9:30 pm clock_pattern = re.compile(r"\b([01]?\d|2[0-3]):[0-5]\d\s?(am|pm)?\b") if clock_pattern.search(body_lower): return True # Number + am/pm: 9am, 10 pm, 3pm, 11am ampm_pattern = re.compile(r"\b([1-9]|1[0-2])\s?(am|pm)\b") if ampm_pattern.search(body_lower): return True # Day + "at" + number: "Saturday at 10", "Monday at 3" day_at_pattern = re.compile(r"\b(monday|tuesday|wednesday|thursday|friday|saturday|sunday|tomorrow)\s+at\s+\d") if day_at_pattern.search(body_lower): return True # Colloquial time expressions: "half 9", "half past 10", "quarter to 3", "quarter past 2" colloquial_pattern = re.compile(r"\b(half\s*(past\s*)?\d{1,2}|quarter\s+(to|past)\s+\d{1,2})\b") if colloquial_pattern.search(body_lower): return True # Approximate times with explicit time marker: "about 10am", "around 11 o'clock", "say 2pm" # Must have am/pm or o'clock to avoid matching "about 10 minutes" approx_time_pattern = re.compile(r"\b(about|around|roughly|say)\s+\d{1,2}\s?(am|pm|o'?clock)\b") if approx_time_pattern.search(body_lower): return True return False def is_bare_number_time(body_lower: str) -> bool: """ Detect if message contains a bare number that could be a time (1-12). This is used ONLY when in appointment context (last_reply_type = APPOINTMENT_TIME_CHOICES) to catch replies like "Can we do 11?" or "11 would be good" that don't have am/pm. Examples that should match: - "Can we do 11?" → True - "11 would be good" → True - "11 works for me" → True - "How about 11" → True - "the 11 please" → True - "let's do 11" → True - "11 👍" → True Examples that should NOT match: - "I have 11 meetings" → False (number not in time context) - "It's been 11 years" → False - "11am" → False (use is_specific_time for this) - "at 11:00" → False (use is_specific_time for this) """ # Skip if already has am/pm or time format (those are handled by is_specific_time) if re.search(r"\b\d{1,2}\s?(am|pm)\b", body_lower): return False if re.search(r"\b\d{1,2}:\d{2}\b", body_lower): return False # Look for bare number 1-12 as a standalone word match = re.search(r"\b([1-9]|1[0-2])\b", body_lower) if not match: return False number = match.group(0) # Check if the number appears in a time-selection context # Patterns that suggest time selection: time_context_patterns = [ rf"\b(can we do|how about|let'?s do|let us do|make it|go for|prefer|suits?|works?|good|great|perfect|fine|do)\s+{number}\b", rf"\b{number}\s+(would be|works|suits|is good|is fine|is great|is perfect|please|for me|thanks|cheers|👍|✅)\b", rf"^{number}[!?\s]*$", # Just the number alone rf"(the\s+)?{number}\s*(slot|one|o'?clock|please)?\s*$", # "the 11", "11 please", "11 one" ] for pattern in time_context_patterns: if re.search(pattern, body_lower): return True # Also match if message is very short and centered on the number # e.g., "11?" or "11 pls" or "yeah 11" words = body_lower.split() if len(words) <= 4 and any(w == number or w.startswith(number) for w in words): # Check it's not a "years/months/minutes" context non_time_context = ["year", "month", "week", "day", "minute", "hour", "meeting", "time"] if not any(ctx in body_lower for ctx in non_time_context): return True return False def is_vague_time_window(body_lower: str) -> bool: """ Detect if message contains a VAGUE time window (day + morning/afternoon/evening). A message is VAGUE if: - It contains a day word (monday-sunday, tomorrow, next week, etc.) - AND a time-of-day word (morning, afternoon, evening) - BUT does NOT contain any explicit time digits Examples: - "Can I do Tuesday afternoon?" → True - "Tomorrow morning works" → True - "Saturday at 10:30" → False (this is specific) - "Maybe next week afternoon" → True """ # Must have a day word or relative word has_day = any(d in body_lower for d in DAY_WORDS) or any(r in body_lower for r in RELATIVE_WORDS) if not has_day: return False # Must have a time-of-day word has_time_of_day = any(t in body_lower for t in TIME_OF_DAY) if not has_time_of_day: return False # Must NOT have specific time digits (otherwise it's specific, not vague) if is_specific_time(body_lower): return False return True def is_day_only_appointment(body_lower: str) -> bool: """ Detect if message is a day-only appointment request (no time-of-day specified). This function is PERMISSIVE - it detects casual day mentions that imply availability, not just explicit "can I come in on Saturday" requests. Examples that should match: - "Can I pop in on Saturday?" → True - "Hi could I pop in on Saturday to have a look at options" → True - "Is Saturday ok?" → True - "Could I come in tomorrow?" → True - "What about next week?" → True - "Hi I could do Saturday ?" → True (casual availability) - "Saturday works for me" → True - "I'm free Saturday" → True - "How about Saturday" → True Examples that should NOT match: - "Saturday morning" → False (use is_vague_time_window for this) - "I worked Saturday" → False (past tense, not availability) - "Last Saturday was busy" → False (past reference) """ # Must have a day word or relative word has_day = any(d in body_lower for d in DAY_WORDS) or any(r in body_lower for r in RELATIVE_WORDS) if not has_day: return False # Must NOT have a time-of-day word (otherwise it's vague, not day-only) has_time_of_day = any(t in body_lower for t in TIME_OF_DAY) if has_time_of_day: return False # Must NOT have specific time (otherwise it's specific) if is_specific_time(body_lower): return False # Exclude past-tense references (not availability) past_indicators = ["last", "worked", "was busy", "had", "went", "did"] if any(past in body_lower for past in past_indicators): return False # Check for visit-related intent keywords OR casual availability phrases visit_intents = [ "pop in", "come in", "visit", "come by", "drop by", "drop in", "call in", "swing by", "stop by", "come over", "see you", "book", "appointment", "available", "free", "open", "can i", "could i", "is it ok", "is that ok", "would it be", "have a look", "look at options", "discuss", "chat" ] # Casual availability phrases (conversational scheduling) # Includes first-person (I could), inclusive (we could), and question forms (could we) casual_availability = [ "i could do", "could do", "i can do", "can do", "we could do", "we can do", "could we do", "can we do", "we could", "we can", "could we", "can we", "should we do", "shall we do", "should we", "shall we", "works for me", "suits me", "good for me", "fine for me", "i'm free", "im free", "i am free", "how about", "what about", "how's", "hows", "is ok", "is good", "is fine", "ok?", "good?", "would work", "might work", "should work", "i could", "i can", "could make", "can make", ] has_visit_intent = any(intent in body_lower for intent in visit_intents) has_casual_availability = any(phrase in body_lower for phrase in casual_availability) return has_visit_intent or has_casual_availability def is_time_of_day_only(body_lower: str) -> bool: """ Detect if message contains ONLY a time-of-day (morning/afternoon/evening) without a day. This happens when customer replies with just time preference after being asked about when they can visit. They may have already mentioned a day, or be clarifying timing. Examples that should match: - "How about in the morning ?" → True - "Morning would be better" → True - "Afternoon works" → True - "I prefer afternoon" → True - "Maybe evening?" → True Examples that should NOT match: - "Good morning" → False (greeting, not appointment) - "Morning" → False (bare greeting) - "Afternoon" → False (bare greeting) - "Saturday morning" → False (has day - use is_vague_time_window) - "Tomorrow afternoon" → False (has day - use is_vague_time_window) - "10am" → False (specific time - use is_specific_time) """ # Must have a time-of-day word has_time_of_day = any(t in body_lower for t in TIME_OF_DAY) if not has_time_of_day: return False # Must NOT have a day word (otherwise it's vague_time_window) has_day = any(d in body_lower for d in DAY_WORDS) or any(r in body_lower for r in RELATIVE_WORDS) if has_day: return False # Must NOT be a specific time (e.g., "10am") if is_specific_time(body_lower): return False # Exclude pure greetings - both "Good morning" AND bare "Morning", "Afternoon", "Evening" # These are conversational greetings, not scheduling requests words = body_lower.strip().split() # Bare time-of-day word alone (1-2 words max) is a greeting, not appointment intent # e.g., "Morning", "Morning!", "Hi morning", "Hello afternoon" if len(words) <= 2: # Check if the message is essentially just the time-of-day word (with optional greeting prefix) greeting_prefixes = ["hi", "hello", "hey", "hiya", "good"] non_greeting_words = [w for w in words if w.rstrip("!?.") not in greeting_prefixes and w.rstrip("!?.") not in TIME_OF_DAY] if not non_greeting_words: return False # It's just a greeting # Explicit greeting patterns pure_greeting_patterns = [ "good morning", "good afternoon", "good evening", ] if any(g in body_lower for g in pure_greeting_patterns): return False # For longer messages, look for scheduling context phrases # These indicate the customer is discussing appointment timing, not just greeting scheduling_context = [ "how about", "what about", "in the", "prefer", "would be", "works", "suits", "better", "good for", "can do", "could do", "maybe", "ideally", "rather", "if possible", "available", ] has_scheduling_context = any(phrase in body_lower for phrase in scheduling_context) return has_scheduling_context def is_weekend_only(body_lower: str) -> bool: """ Detect if message mentions "this weekend" without specifying Saturday or Sunday. This triggers a refinement flow asking which day and time-of-day. Examples that should match: - "I can do this weekend" → True - "This weekend works" → True - "How about the weekend?" → True - "Weekend is better" → True Examples that should NOT match: - "Saturday" → False (specific day) - "Saturday morning" → False (specific day + time) - "This weekend Saturday" → False (has specific day) """ weekend_phrases = [ "this weekend", "the weekend", "weekend works", "weekend is", "at the weekend", "on the weekend", "do the weekend", "do weekend", "weekend would", "weekend suits", "weekend better", "weekend please", ] has_weekend = any(phrase in body_lower for phrase in weekend_phrases) if not has_weekend: return False # If they also mention a specific day (Saturday/Sunday), it's not weekend-only has_specific_day = "saturday" in body_lower or "sunday" in body_lower if has_specific_day: return False return True def is_next_week_only(body_lower: str) -> bool: """ Detect if message mentions "next week" without specifying a day. This triggers a refinement flow asking weekdays vs weekend and time-of-day. Examples that should match: - "Next week is better" → True - "Can we do next week?" → True - "How about next week sometime" → True - "Next week works for me" → True Examples that should NOT match: - "Next Monday" → False (specific day) - "Next week Tuesday" → False (has specific day) - "Next Saturday" → False (specific day) """ next_week_phrases = [ "next week", "the following week", ] has_next_week = any(phrase in body_lower for phrase in next_week_phrases) if not has_next_week: return False # If they also mention a specific day, it's not next-week-only specific_days = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"] has_specific_day = any(day in body_lower for day in specific_days) if has_specific_day: return False return True def detect_appointment_confirmation(body_lower: str) -> bool: """ Legacy wrapper: returns True if message contains any appointment indication. For new code, prefer using is_specific_time() or is_vague_time_window() directly. """ return is_specific_time(body_lower) or is_vague_time_window(body_lower) def detect_cant_visit(body_lower: str) -> bool: """Detect if customer indicates they can't visit showroom.""" return any(phrase in body_lower for phrase in CANT_VISIT_PHRASES) def is_urgent_service(body_lower: str) -> bool: """Detect if service/support request is urgent.""" return any(keyword in body_lower for keyword in URGENT_KEYWORDS) # ============================================================================= # SOFT-NO / PAUSE / DEFER DETECTION FUNCTIONS # ============================================================================= def is_busy_now(body_lower: str) -> bool: """ Detect if customer is temporarily busy (not a soft objection). These messages should NOT increment soft_no_count and should keep status active. Examples: - "I'm at work right now" → True - "Busy at the moment" → True - "Can't talk right now" → True - "In a meeting" → True - "Not right now, maybe later" → False (this is a soft objection, not busy-now) """ busy_now_phrases = [ "at work", "im at work", "i'm at work", "in a meeting", "in meeting", "busy right now", "busy at the moment", "bit busy right now", "can't talk right now", "cant talk right now", "can't chat right now", "cant chat right now", "driving", "on the road", "give me a minute", "give me a sec", "just a sec", "one moment", "hang on", ] return any(phrase in body_lower for phrase in busy_now_phrases) def is_callback_request(body_lower: str) -> bool: """ Detect if customer explicitly asks for a phone call instead of WhatsApp. Examples: - "Just call me" → True - "Can someone phone me?" → True - "Give me a ring" → True - "Prefer a call" → True """ callback_phrases = [ "call me", "phone me", "ring me", "give me a call", "give me a ring", "can you call", "can someone call", "can someone phone", "prefer a call", "rather a call", "just call", "just ring", "just phone", "call instead", "phone instead", "ring instead", "can i get a call", "could i get a call", ] return any(phrase in body_lower for phrase in callback_phrases) def is_defer_request(body_lower: str) -> Tuple[bool, str | None]: """ Detect if customer requests to be contacted at a specific future time. Returns: Tuple of (is_defer, parsed_context) - is_defer: True if defer request detected - parsed_context: Human-readable description of when (e.g., "next month", "after Christmas") Examples: - "Try me again next month" → (True, "next month") - "After Christmas" → (True, "after Christmas") - "Contact me in February" → (True, "in February") - "Maybe in the new year" → (True, "in the new year") """ defer_patterns = [ ("try me again", "as requested"), ("get back to me", "as requested"), ("contact me in", "as requested"), ("message me in", "as requested"), ("reach out in", "as requested"), ("after christmas", "after Christmas"), ("after new year", "after New Year"), ("after the new year", "after New Year"), ("in the new year", "in the new year"), ("next month", "next month"), ("next year", "next year"), ("in a few weeks", "in a few weeks"), ("in a couple of weeks", "in a couple of weeks"), ("in a month", "in a month"), ("in january", "in January"), ("in february", "in February"), ("in march", "in March"), ("in april", "in April"), ("in may", "in May"), ("in june", "in June"), ("in july", "in July"), ("in august", "in August"), ("in september", "in September"), ("in october", "in October"), ("in november", "in November"), ("in december", "in December"), ] for pattern, context in defer_patterns: if pattern in body_lower: return (True, context) return (False, None) def is_not_interested_pause(body_lower: str) -> bool: """ Detect if customer says "not interested right now" or "leave it for now". This is NOT a T3 opt-out - customer is saying "not now" not "never". Should set status = paused, defer 60 days, but NOT set DNC. Examples: - "Not interested right now" → True - "Leave it for now" → True - "Maybe another time" → True - "Not at the moment" → True - "Stop messaging me" → False (this is T3 opt-out) """ not_interested_phrases = [ "not interested right now", "not interested at the moment", "not interested for now", "leave it for now", "leave me for now", "park it for now", "not at the moment", "maybe another time", "not for me right now", ] # Check for positive match if any(phrase in body_lower for phrase in not_interested_phrases): # Make sure it's NOT a hard opt-out hard_stop_indicators = ["stop", "unsubscribe", "remove me", "don't contact", "do not contact"] if not any(indicator in body_lower for indicator in hard_stop_indicators): return True return False def classify_inbound_tier(body_lower: str) -> str: """ Classify inbound message into tiers. Returns one of: - T3_HARD_STOP: Opt-out / stop messaging - T2_SOFT_OBJECTION: Not now / maybe later - T4_SCHEDULED: Contact me later / specific time - T5_MONEY: Cost / affordability concerns - T6_SUPPORT: Service / MOT / warranty issues - T1_GENERAL: General interest / everything else """ # T3: Hard stop (highest priority) # IMPORTANT: "not interested" alone is T3, but "not interested right now" is NOT_INTERESTED_PAUSE # Check for soft pause patterns first to exclude them from hard stop soft_pause_phrases = [ "not interested right now", "not interested at the moment", "not interested for now", "leave it for now", "park it for now", "not at the moment", "maybe another time", "not for me right now" ] is_soft_pause = any(phrase in body_lower for phrase in soft_pause_phrases) hard_stop_phrases = [ "stop", "unsubscribe", "remove me", "delete my", "opt out", "do not contact", "don't contact", "leave me alone", "take me off", "no more messages", "stop texting", "stop messaging", "don't message", "dont message" ] # Only check "not interested" if it's NOT a soft pause (e.g., "not interested right now") if not is_soft_pause and "not interested" in body_lower: return "T3_HARD_STOP" if any(phrase in body_lower for phrase in hard_stop_phrases): return "T3_HARD_STOP" # T6: Support / service issues support_phrases = [ "service", "mot", "mot due", "warranty", "repair", "broken", "fault", "warning light", "problem with", "issue with", "broken down", "mechanic", "workshop", "fix", "technician" ] if any(phrase in body_lower for phrase in support_phrases): return "T6_SUPPORT" # T5: Money / affordability money_phrases = [ "too expensive", "can't afford", "cant afford", "budget", "cost", "how much", "price", "payment", "monthly", "afford", "expensive", "cheaper", "money", "finance" ] if any(phrase in body_lower for phrase in money_phrases): return "T5_MONEY" # T4: Scheduled / contact later scheduled_phrases = [ "next month", "few weeks", "couple of months", "later on", "closer to the time", "nearer the time", "try me in", "contact me in", "get back to me in" ] if any(phrase in body_lower for phrase in scheduled_phrases): return "T4_SCHEDULED" # T2: Soft objection soft_objection_phrases = [ "not right now", "maybe later", "not yet", "not ready", "bit busy", "leave it for now", "not urgent", "thinking about it", "haven't decided", "not sure", "don't know", "need time" ] if any(phrase in body_lower for phrase in soft_objection_phrases): return "T2_SOFT_OBJECTION" # T1: General (default) return "T1_GENERAL" # ============================================================================= # MAIN REPLY GENERATOR # ============================================================================= def generate_lia_reply(customer_state: Dict[str, Any], inbound_text: str) -> Tuple[str, Dict[str, Any]]: """ Generate Lia's reply based on customer state and inbound message. Args: customer_state: Dictionary containing: - customer_id: str - customer_name: str | None - stage: str | None (e.g., "12", "9", "6", "3") - soft_no_count: int (default 0) - appointment_status: str ("NONE", "INVITED", "BOOKED") - opted_out: bool (default False) - last_message: str | None - last_reply_type: str | None inbound_text: Customer's WhatsApp message Returns: Tuple of (reply_text, updated_customer_state) """ # Normalise input body = normalise_text(inbound_text) body_lower = body.lower() # Copy state for updates new_state = customer_state.copy() new_state["last_message"] = inbound_text # Ensure confusion/escalation fields exist if "confusion_count" not in new_state: new_state["confusion_count"] = 0 if "needs_human_review" not in new_state: new_state["needs_human_review"] = False # Get customer name for personalisation customer_name = customer_state.get("customer_name") # ------------------------------------------------------------------------- # 1. EARLY EXIT: Already opted out # ------------------------------------------------------------------------- if customer_state.get("opted_out"): return ("", new_state) # ------------------------------------------------------------------------- # 1b. CLEAR APPOINTMENT INTENT OVERRIDE # If customer clearly wants to book, override needs_human_review escalation # ------------------------------------------------------------------------- in_time_selection = customer_state.get("last_reply_type") == "APPOINTMENT_TIME_CHOICES" is_clear_appointment_intent = ( is_vague_time_window(body_lower) or is_day_only_appointment(body_lower) or (is_specific_time(body_lower) and in_time_selection) or (is_bare_number_time(body_lower) and in_time_selection) # "Can we do 11?" when in slot selection ) # If customer has clear appointment intent, clear the escalation flag # This prevents "pop in on Saturday" from being stuck in AWAITING_MANAGER if is_clear_appointment_intent and customer_state.get("needs_human_review"): new_state["needs_human_review"] = False print(f" Clearing needs_human_review - clear appointment intent detected") # ------------------------------------------------------------------------- # 1c. POST-BOOKING GUARD: Appointment already confirmed # Handle acknowledgements gracefully BEFORE escalation check # This ensures "Thanks Lia!" after booking gets a friendly reply, not AWAITING_MANAGER # ------------------------------------------------------------------------- if customer_state.get("appointment_status") == "BOOKED": last_reply = customer_state.get("last_reply_type", "") # Check for simple acknowledgements (thanks, great, 👍, etc.) ack_patterns = [ "thanks", "thank you", "cheers", "ta", "thx", "ty", "great", "perfect", "lovely", "brilliant", "fab", "sounds good", "see you", "looking forward", "can't wait", "will do", "👍", "👌", "🙏", "😊", "🎉", "✅", "ok", "okay", "cool", "nice", ] # Negative/soft-no patterns that override acknowledgement detection # These indicate the customer may not be able to make the appointment soft_no_patterns = [ "not right now", "too busy", "can't make", "cant make", "won't make", "wont make", "don't think", "dont think", "might not", "may not", "no time", "no longer", "not sure", "not going to", "not gonna", "unable to", "have to cancel", "need to cancel", "cancel it", "something came up", "something's come up", "change of plan", "can't do", "cant do", "won't be able", "wont be able", "not able", "sorry but", "unfortunately", ] # Messages under 60 chars with positive vibes (expanded from 20 to catch "Thanks Lia!") has_ack_pattern = any(p in body_lower for p in ack_patterns) has_soft_no_pattern = any(p in body_lower for p in soft_no_patterns) is_short_ack = len(body_lower) < 60 and has_ack_pattern and not has_soft_no_pattern is_emoji_only = len(body_lower) <= 5 and any(c in body_lower for c in "👍👌🙏😊🎉✅") # Duplicate processing guard: if last reply was APPOINTMENT_CONFIRMED and # the current message looks like the original slot selection (contains time) is_duplicate_time_message = ( last_reply == "APPOINTMENT_CONFIRMED" and is_specific_time(body_lower) ) # POST-BOOKING MONEY CONCERN: Customer expresses affordability worry # This takes priority over soft-no patterns (e.g., "not sure I can afford" has "not sure") # Route to T5 money path with reassurance - keep the appointment as exploratory has_money_concern = is_money_concern(body) if has_money_concern: raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" # Get appointment details for reference day_display = customer_state.get("appointment_date_display", "") day_date = customer_state.get("pending_day_date") time_display = customer_state.get("appointment_time_display", "") # Format date if available date_str = "" if day_date: if isinstance(day_date, str): try: day_date = datetime.strptime(day_date, "%Y-%m-%d").date() except ValueError: day_date = None if day_date: date_str = format_date_uk(day_date) # Build appointment reference if day_display and date_str and time_display: appointment_ref = f" on {day_display} {date_str} at {time_display}" elif day_display and time_display: appointment_ref = f" on {day_display} at {time_display}" elif time_display: appointment_ref = f" at {time_display}" else: appointment_ref = "" # T5-style money concern reply - reassure, frame as exploratory, no pressure reply = ( f"I completely understand, {name} – things are expensive right now and it's sensible to be careful. " f"Your appointment{appointment_ref} is just for a chat to see what options might be available – " f"there's absolutely no pressure, and doing nothing could well be the best choice. " f"If you'd prefer to have a quick call instead, just let me know and I can arrange that." ) new_state["last_reply_type"] = "T5_MONEY_CONCERN_FIRST_POST_BOOKING" new_state["needs_human_review"] = False # We've handled it # Keep appointment_status = BOOKED - the slot stays in diary return (reply, new_state) # POST-BOOKING SOFT-NO: Customer indicates they may not make the appointment (time/busyness) # Respond with understanding and offer to reschedule (don't re-confirm) # ONLY fire once - if we already sent POST_BOOKING_SOFT_NO, let URE handle subsequent messages already_sent_soft_no = last_reply == "POST_BOOKING_SOFT_NO" if has_soft_no_pattern and not already_sent_soft_no and not has_money_concern: raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" # Get appointment details for the reschedule offer day_display = customer_state.get("appointment_date_display", "") day_date = customer_state.get("pending_day_date") time_display = customer_state.get("appointment_time_display", "") # Format date if available date_str = "" if day_date: if isinstance(day_date, str): try: day_date = datetime.strptime(day_date, "%Y-%m-%d").date() except ValueError: day_date = None if day_date: date_str = format_date_uk(day_date) # Build reschedule offer with available details if day_display and date_str and time_display: appointment_ref = f"{day_display} {date_str} at {time_display}" elif day_display and time_display: appointment_ref = f"{day_display} at {time_display}" elif time_display: appointment_ref = f"at {time_display}" else: appointment_ref = "your appointment" reply = f"No problem at all, {name} – I know things can get busy. Your slot on {appointment_ref} is still in the diary, but if you'd like to change or cancel it, just let me know what works better and I'll update it." new_state["last_reply_type"] = "POST_BOOKING_SOFT_NO" new_state["needs_human_review"] = False # Don't escalate, we've handled it return (reply, new_state) # Pure acknowledgement (no soft-no patterns) if is_short_ack or is_emoji_only: raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" # Build a warm acknowledgement with appointment details day_display = customer_state.get("appointment_date_display", "") day_date = customer_state.get("pending_day_date") time_display = customer_state.get("appointment_time_display", "") # Format date if available date_str = "" if day_date: if isinstance(day_date, str): try: day_date = datetime.strptime(day_date, "%Y-%m-%d").date() except ValueError: day_date = None if day_date: date_str = format_date_uk(day_date) # Build confirmation with available details if day_display and date_str and time_display: reply = f"You're very welcome, {name} – we'll see you on {day_display} {date_str} at {time_display} at Lincoln Audi. 👋" elif day_display and time_display: reply = f"You're very welcome, {name} – we'll see you on {day_display} at {time_display} at Lincoln Audi. 👋" elif time_display: reply = f"You're very welcome, {name} – we'll see you at {time_display} at Lincoln Audi. 👋" else: reply = f"You're very welcome, {name} – we'll see you soon at Lincoln Audi! 👋" new_state["last_reply_type"] = "POST_BOOKING_ACK" new_state["needs_human_review"] = False # Clear any escalation flag return (reply, new_state) # Suppress duplicate processing - don't send confusion for re-processed time messages if is_duplicate_time_message: # Silent return - the confirmation was already sent new_state["last_reply_type"] = "DUPLICATE_SUPPRESSED" return ("", new_state) # ------------------------------------------------------------------------- # 1d. POST-DECISION GUARD: Handle "Thanks" after defer/soft-no/opt-out # Prevents re-pitching when customer is just being polite after a decision # ------------------------------------------------------------------------- last_reply = customer_state.get("last_reply_type") or "" # Decision types that should trigger the ack guard post_decision_types = [ "T4_DEFER_CUSTOMER_REPLY", # Customer requested deferral "T2_SOFT_NO_SECOND", # Second soft no (final close-down) "T2_SOFT_NO_FINAL", # Final soft no "T6_OPT_OUT_FINAL", # Opt-out confirmed "T6_DNC_CONFIRMED", # Do not contact confirmed ] is_post_decision = last_reply and any(last_reply.startswith(dt) for dt in post_decision_types) if is_post_decision and is_simple_acknowledgement(body): raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" # Tailor reply based on what decision was made if last_reply.startswith("T4_DEFER"): # After deferral - mention we'll pick up later defer_until = customer_state.get("defer_until") if defer_until: # Format the defer period naturally if isinstance(defer_until, str): try: defer_until = datetime.strptime(defer_until, "%Y-%m-%d").date() except ValueError: defer_until = None if defer_until and defer_until.month == 1: period = "in the new year" else: period = "when the time is right" else: period = "when the time is right" reply = f"You're very welcome, {name} – we'll pick this up {period}, but if you do want to talk sooner just drop me a quick message here. 👋" new_state["last_reply_type"] = "POST_DEFER_ACK" elif last_reply.startswith("T2_SOFT_NO"): # After final soft-no - leave door open gently reply = f"You're very welcome, {name} – if you change your mind in future, you can always drop us a message here. 👋" new_state["last_reply_type"] = "POST_SOFT_NO_ACK" elif last_reply.startswith("T6"): # After opt-out - acknowledge and close gracefully reply = f"You're very welcome, {name} – we'll leave this one alone now, but if you ever want to talk about options again just get in touch. 👋" new_state["last_reply_type"] = "POST_OPT_OUT_ACK" else: # Generic fallback reply = f"You're very welcome, {name}. 👋" new_state["last_reply_type"] = "POST_DECISION_ACK" # Don't change any status fields - just acknowledge new_state["needs_human_review"] = False print(f" ✅ Post-decision ack detected after {last_reply}") return (reply, new_state) # ------------------------------------------------------------------------- # 1e. CHRISTMAS / NEW YEAR DEFERRAL: Handle "after Christmas" etc. # This runs BEFORE escalation check to avoid false AWAITING_MANAGER # ------------------------------------------------------------------------- if is_clear_defer_request(body): raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" # Compute the defer_until date based on the phrase defer_until = compute_defer_date(body) defer_display = defer_until.strftime("%d %B") # e.g., "06 January" # Build T4-style deferral reply reply = ( f"No problem at all, {name} – we can pick this up after Christmas. " f"I'll make a note to get back in touch then, but if you'd like to talk sooner " "just drop me a message here." ) # Update state: cancel any existing appointment, set defer date new_state["last_reply_type"] = "T4_DEFER_CUSTOMER_REPLY" new_state["appointment_status"] = "NONE" new_state["defer_until"] = defer_until # Pass date object directly new_state["status"] = "active" # Keep conversation active new_state["needs_human_review"] = False # Clear escalation flag # Clear appointment display fields since we're cancelling new_state["pending_day_display"] = None new_state["pending_day_date"] = None new_state["pending_slot_options"] = None new_state["appointment_date_display"] = None new_state["appointment_time_display"] = None print(f" ✅ Christmas deferral detected - defer_until={defer_until}") return (reply, new_state) # ------------------------------------------------------------------------- # 1f. SETTLEMENT / BALANCE QUERY: Customer asking about their settlement # Route to appropriate template based on whether appointment is booked # ------------------------------------------------------------------------- if is_settlement_query(body): raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" appointment_status = customer_state.get("appointment_status", "NONE") if appointment_status == "BOOKED": # Already have appointment - reassure that settlement will be covered day_display = customer_state.get("appointment_date_display", "") day_date = customer_state.get("pending_day_date") time_display = customer_state.get("appointment_time_display", "") # Format date if available date_str = "" if day_date: if isinstance(day_date, str): try: day_date = datetime.strptime(day_date, "%Y-%m-%d").date() except ValueError: day_date = None if day_date: date_str = format_date_uk(day_date) # Build appointment reference if day_display and date_str and time_display: appt_ref = f"on {day_display} {date_str} at {time_display}" elif day_display and time_display: appt_ref = f"on {day_display} at {time_display}" elif time_display: appt_ref = f"at {time_display}" else: appt_ref = "at your appointment" reply = ( f"Good question, {name}. I can't send the exact settlement in this chat, but when you're in " f"{appt_ref} we'll go through your agreement, settlement and options properly with you.\n\n" f"If you'd like, we can also give you a view on whether it's better to change, refinance or just leave things as they are." ) new_state["last_reply_type"] = "SETTLEMENT_INFO_WHEN_BOOKED" new_state["needs_human_review"] = False print(f" ✅ Settlement query detected (appointment booked)") return (reply, new_state) else: # No appointment yet - offer visit or call reply = ( f"Thanks for asking about your settlement, {name}. I can't see the exact figure here in WhatsApp " f"because we need to run a couple of security checks in our system first.\n\n" f"What I can do is have one of the team go through your agreement and figures properly and check " f"whether it's worth doing anything at all – including if the answer is 'best leave it as it is'.\n\n" f"Would you prefer to pop in for a quick visit at Lincoln Audi, or would a short phone call work better " f"to run through your settlement and options?" ) new_state["last_reply_type"] = "SETTLEMENT_INFO_FIRST" new_state["needs_human_review"] = False print(f" ✅ Settlement query detected (no appointment)") return (reply, new_state) # ------------------------------------------------------------------------- # 1g. SAME-DAY VAGUE TIME: "I'm free all afternoon", "this afternoon", etc. # Only for active conversations, not cold inbound or complaints # ------------------------------------------------------------------------- if is_same_day_vague_time(body): status = customer_state.get("status", "active") escalation_reason = customer_state.get("escalation_reason", "") last_reply = customer_state.get("last_reply_type", "") # Guard: Only trigger for active conversations (not cold inbound, complaints, DNC) # Cold inbound would have last_reply_type empty or None is_active_conversation = ( status == "active" and escalation_reason != "customer_issue" and last_reply # Has some prior conversation context ) if is_active_conversation: # Generate same-day slots same_day_slots = generate_same_day_slots() if same_day_slots: # We have valid slots - offer them raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" slot1, slot2 = same_day_slots[0], same_day_slots[1] if len(same_day_slots) > 1 else same_day_slots[0] # Determine if this is a call context (T5 money concern / phone fallback) is_call_context = last_reply and any(x in last_reply for x in ["T5_MONEY", "CALLBACK", "PHONE", "CALL"]) # Also check if message mentions call if "call" in body.lower(): is_call_context = True if is_call_context: reply = ( f"No problem, {name} – this afternoon works.\n\n" f"I can give you a quick call at {slot1} or {slot2} – which would you prefer?" ) else: reply = ( f"No problem, {name} – this afternoon works.\n\n" f"I've got {slot1} or {slot2} available to pop into Lincoln Audi – which suits you better?" ) # Update state new_state["last_reply_type"] = "APPOINTMENT_TIME_CHOICES" new_state["pending_day_display"] = "today" new_state["pending_day_date"] = datetime.now().date().isoformat() # Store as string new_state["pending_slot_options"] = same_day_slots if customer_state.get("appointment_status") != "BOOKED": new_state["appointment_status"] = "INVITED" new_state["needs_human_review"] = False print(f" ✅ Same-day vague time detected - offering slots: {same_day_slots}") return (reply, new_state) else: # Too late in the day - offer tomorrow instead raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" reply = ( f"Unfortunately it's a bit late in the day for today, {name}. " f"Would tomorrow work for you? I've got a few slots available in the morning or afternoon." ) new_state["last_reply_type"] = "SAME_DAY_TOO_LATE" new_state["needs_human_review"] = False print(f" ⚠️ Same-day too late - offering tomorrow") return (reply, new_state) # ------------------------------------------------------------------------- # 1h. EARLY EXIT: Already escalated to manager (stay silent) # But only if we didn't just clear it due to appointment intent # ------------------------------------------------------------------------- if new_state.get("needs_human_review"): new_state["last_reply_type"] = "AWAITING_MANAGER" return ("", new_state) # ------------------------------------------------------------------------- # 2. SLOT CHOICE REPLY: Customer picking from offered time slots # ------------------------------------------------------------------------- if (customer_state.get("last_reply_type") == "APPOINTMENT_TIME_CHOICES" and customer_state.get("pending_slot_options")): pending_options = customer_state.get("pending_slot_options", []) chosen_time = None # Try to match the customer's reply to one of the offered slots for option in pending_options: # Extract hour key (e.g., "9" from "9:00am", "11" from "11:00am") hour_match = re.match(r"(\d{1,2})", option) if hour_match: hour_key = hour_match.group(1) # Check if the hour appears in the customer's reply # Match patterns like "11", "11am", "11:00", "the 11" if re.search(rf"\b{hour_key}\b", body_lower): chosen_time = option break # If we matched a chosen time, confirm the booking if chosen_time: raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" # Build appointment display fields day_display = customer_state.get("pending_day_display", "") day_date = customer_state.get("pending_day_date") time_display = chosen_time # Format date if available (may be a string from DB) date_str = "" day_name = "" if day_date: if isinstance(day_date, str): try: day_date = datetime.strptime(day_date, "%Y-%m-%d").date() except ValueError: day_date = None if day_date: date_str = format_date_uk(day_date) day_name = day_date.strftime("%A") # Get actual day name (Saturday, Sunday, etc.) # Use actual day name from date if available, otherwise fall back to day_display # This prevents "This Weekend 13/12/2025" - instead shows "Saturday 13/12/2025" display_day = day_name if day_name else day_display # Create confirmation with explicit date and time (Style B: "that's ideal") if display_day and date_str: reply = ( f"Thanks {name} – that's ideal.\n\n" f"I've booked you in for {display_day} {date_str} at {time_display} at Lincoln Audi.\n\n" "If anything changes, just drop me a quick message here and I'll happily move it." ) elif display_day: reply = ( f"Thanks {name} – that's ideal.\n\n" f"I've booked you in for {display_day} at {time_display} at Lincoln Audi.\n\n" "If anything changes, just drop me a quick message here and I'll happily move it." ) else: reply = ( f"Thanks {name} – that's ideal.\n\n" f"I've booked you in for {time_display} at Lincoln Audi.\n\n" "If anything changes, just drop me a quick message here and I'll happily move it." ) new_state["appointment_status"] = "BOOKED" new_state["appointment_confirmed_text"] = inbound_text new_state["appointment_confirmed_at"] = datetime.now(timezone.utc).isoformat() new_state["appointment_date_display"] = day_display new_state["appointment_time_display"] = time_display new_state["pending_day_date"] = day_date.isoformat() if day_date else None new_state["last_reply_type"] = "APPOINTMENT_CONFIRMED" new_state["pending_slot_options"] = None new_state["pending_slot_context"] = None new_state["pending_day_display"] = None return (reply, new_state) # If no match, fall through to normal logic below # ------------------------------------------------------------------------- # 3. SPECIFIC TIME (EARLY RETURN): Customer gives exact time → BOOK IT # ONLY if LIA just offered time slots (last_reply_type = APPOINTMENT_TIME_CHOICES) # This prevents auto-booking when customer mentions times in other contexts # Also handles bare number times like "Can we do 11?" when in slot selection # ------------------------------------------------------------------------- last_reply_type = customer_state.get("last_reply_type", "") in_slot_selection = (last_reply_type == "APPOINTMENT_TIME_CHOICES" or customer_state.get("pending_slot_options") is not None) # Check for specific time (with am/pm) OR bare number time (only in slot selection context) has_time = is_specific_time(body_lower) or (in_slot_selection and is_bare_number_time(body_lower)) if has_time and in_slot_selection: raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" # Extract time display from customer's message time_display = extract_time_display(inbound_text) day_display = customer_state.get("pending_day_display", "") day_date = customer_state.get("pending_day_date") # Format date if available (may be a string from DB) date_str = "" day_name = "" if day_date: if isinstance(day_date, str): try: day_date = datetime.strptime(day_date, "%Y-%m-%d").date() except ValueError: day_date = None if day_date: date_str = format_date_uk(day_date) day_name = day_date.strftime("%A") # Get actual day name (Saturday, Sunday, etc.) # Use actual day name from date if available, otherwise fall back to day_display # This prevents "This Weekend 13/12/2025" - instead shows "Saturday 13/12/2025" display_day = day_name if day_name else day_display # Create confirmation with explicit date and time (Style B: "that's ideal") if display_day and date_str: reply = ( f"Thanks {name} – that's ideal.\n\n" f"I've booked you in for {display_day} {date_str} at {time_display} at Lincoln Audi.\n\n" "If anything changes, just drop me a quick message here and I'll happily move it." ) elif display_day: reply = ( f"Thanks {name} – that's ideal.\n\n" f"I've booked you in for {display_day} at {time_display} at Lincoln Audi.\n\n" "If anything changes, just drop me a quick message here and I'll happily move it." ) else: reply = ( f"Thanks {name} – that's ideal.\n\n" f"I've booked you in for {time_display} at Lincoln Audi.\n\n" "If anything changes, just drop me a quick message here and I'll happily move it." ) new_state["appointment_status"] = "BOOKED" new_state["appointment_confirmed_text"] = inbound_text new_state["appointment_confirmed_at"] = datetime.now(timezone.utc).isoformat() new_state["appointment_date_display"] = day_display new_state["appointment_time_display"] = time_display new_state["pending_day_date"] = day_date.isoformat() if day_date else None new_state["last_reply_type"] = "APPOINTMENT_CONFIRMED" new_state["pending_day_display"] = None return (reply, new_state) # ------------------------------------------------------------------------- # 3c. WEEKEND REFINEMENT RESPONSE: Customer replied to weekend refinement # e.g., "Saturday morning" after "Would Saturday or Sunday be better?" # ------------------------------------------------------------------------- if customer_state.get("last_reply_type") == "WEEKEND_REFINEMENT": raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" # Parse day choice (Saturday or Sunday) chosen_day = None if "saturday" in body_lower: chosen_day = "Saturday" elif "sunday" in body_lower: chosen_day = "Sunday" # Parse time-of-day preference time_preference = None if "morning" in body_lower: time_preference = "morning" elif "afternoon" in body_lower: time_preference = "afternoon" elif "evening" in body_lower: time_preference = "evening" if chosen_day and time_preference: # Both day and time specified - offer 2 slots day_date = compute_day_date(chosen_day.lower()) if time_preference == "morning": slots_text = "10:00am or 11:30am" slot_options = ["10:00am", "11:30am"] elif time_preference == "evening": slots_text = "5:30pm or 7:00pm" slot_options = ["5:30pm", "7:00pm"] else: slots_text = "1:00pm or 3:00pm" slot_options = ["1:00pm", "3:00pm"] reply = ( f"Perfect, {name} – {chosen_day} {time_preference} works.\n\n" f"I've got {slots_text} available.\n\n" "Which would suit you better?" ) new_state["appointment_status"] = "INVITED" new_state["last_reply_type"] = "APPOINTMENT_TIME_CHOICES" new_state["pending_slot_options"] = slot_options new_state["pending_slot_context"] = inbound_text new_state["pending_day_display"] = chosen_day new_state["pending_day_date"] = day_date.isoformat() if day_date else None return (reply, new_state) elif chosen_day: # Day only - ask for time preference day_date = compute_day_date(chosen_day.lower()) reply = ( f"Great, {name} – {chosen_day} is no problem.\n\n" "Would you prefer morning or afternoon?" ) new_state["last_reply_type"] = "DAY_SELECTED_NEED_TIME" new_state["pending_day_display"] = chosen_day new_state["pending_day_date"] = day_date.isoformat() if day_date else None return (reply, new_state) elif time_preference: # Time preference only - ask which weekend day (with weekday option per Style B) reply = ( f"Thanks {name} – {time_preference} suits us fine.\n\n" f"Would Saturday or Sunday be better for you, or is a weekday {time_preference} easier?" ) new_state["last_reply_type"] = "TIME_SELECTED_NEED_DAY" new_state["pending_time_preference"] = time_preference return (reply, new_state) else: # Customer gave neither day nor time-of-day # Check if they gave a specific time like "11am" - if so, ask which weekend day if is_specific_time(body_lower): # They want a specific time - need to know which weekend day time_display = extract_time_display(inbound_text) reply = ( f"Great, {name} – {time_display} works for us.\n\n" "Would that be Saturday or Sunday?" ) new_state["last_reply_type"] = "SPECIFIC_TIME_NEED_WEEKEND_DAY" new_state["pending_specific_time"] = time_display return (reply, new_state) # Otherwise re-ask the refinement question reply = ( f"No problem, {name}.\n\n" "Would Saturday or Sunday be better, and do you prefer morning or afternoon?" ) new_state["last_reply_type"] = "WEEKEND_REFINEMENT" return (reply, new_state) # ------------------------------------------------------------------------- # 3d. NEXT WEEK REFINEMENT RESPONSE: Customer replied to next week refinement # e.g., "weekdays morning" or "weekend afternoon" # ------------------------------------------------------------------------- if customer_state.get("last_reply_type") == "NEXT_WEEK_REFINEMENT": raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" # Parse weekday/weekend preference wants_weekend = "weekend" in body_lower or "saturday" in body_lower or "sunday" in body_lower wants_weekday = any(w in body_lower for w in ["weekday", "week day", "monday", "tuesday", "wednesday", "thursday", "friday"]) # Parse time-of-day preference time_preference = None if "morning" in body_lower: time_preference = "morning" elif "afternoon" in body_lower: time_preference = "afternoon" elif "evening" in body_lower: time_preference = "evening" if wants_weekend: # Redirect to weekend refinement flow reply = ( f"Great, {name} – a weekend visit works well.\n\n" "Would Saturday or Sunday be better, and do you prefer morning or afternoon?" ) new_state["last_reply_type"] = "WEEKEND_REFINEMENT" return (reply, new_state) elif wants_weekday and time_preference: # Weekday + time preference - offer 2 concrete slots today = date.today() next_monday = today + timedelta(days=(7 - today.weekday())) # Pick two weekdays next week if time_preference == "morning": slot1_day = next_monday + timedelta(days=1) # Tuesday slot2_day = next_monday + timedelta(days=3) # Thursday slot_time = "10:00am" elif time_preference == "evening": slot1_day = next_monday + timedelta(days=1) # Tuesday slot2_day = next_monday + timedelta(days=3) # Thursday slot_time = "5:30pm" else: slot1_day = next_monday + timedelta(days=1) # Tuesday slot2_day = next_monday + timedelta(days=3) # Thursday slot_time = "2:00pm" day1_name = slot1_day.strftime("%A") day2_name = slot2_day.strftime("%A") day1_date_str = format_date_uk(slot1_day) day2_date_str = format_date_uk(slot2_day) reply = ( f"Perfect, {name} – I've got two {time_preference} slots available next week:\n\n" f"• {day1_name} {day1_date_str} at {slot_time}\n" f"• {day2_name} {day2_date_str} at {slot_time}\n\n" "Which would suit you better?" ) new_state["appointment_status"] = "INVITED" new_state["last_reply_type"] = "NEXT_WEEK_SLOT_CHOICES" new_state["pending_slot_options"] = [f"{day1_name}_{slot_time}", f"{day2_name}_{slot_time}"] new_state["pending_next_week_slots"] = [ {"day": day1_name, "date": slot1_day.isoformat(), "time": slot_time}, {"day": day2_name, "date": slot2_day.isoformat(), "time": slot_time}, ] return (reply, new_state) elif wants_weekday: # Weekday only - ask for time preference (Style B: "Great, {name} – Tuesday works") reply = ( f"Great, {name} – a weekday works.\n\n" "Would you prefer morning or afternoon?" ) new_state["last_reply_type"] = "WEEKDAY_SELECTED_NEED_TIME" return (reply, new_state) elif time_preference: # Time preference only - ask weekday or weekend (Style B) reply = ( f"Thanks {name} – {time_preference} suits us fine.\n\n" "Would a weekday or the weekend be easier for you?" ) new_state["last_reply_type"] = "TIME_SELECTED_NEED_WEEKDAY_OR_WEEKEND" new_state["pending_time_preference"] = time_preference return (reply, new_state) else: # Customer gave neither weekday/weekend nor time preference # Re-ask the refinement question reply = ( f"No problem, {name}.\n\n" "Are weekdays or the weekend better, and do you prefer morning or afternoon?" ) new_state["last_reply_type"] = "NEXT_WEEK_REFINEMENT" return (reply, new_state) # ------------------------------------------------------------------------- # 3e. DAY_SELECTED_NEED_TIME: Customer chose a day, now needs time # ------------------------------------------------------------------------- if customer_state.get("last_reply_type") == "DAY_SELECTED_NEED_TIME": raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" pending_day = customer_state.get("pending_day_display", "") pending_day_date = customer_state.get("pending_day_date") # Parse time-of-day preference (Style B pattern) if "morning" in body_lower: slots_text = "10:00am or 11:30am" slot_options = ["10:00am", "11:30am"] time_of_day = "morning" elif "evening" in body_lower: slots_text = "5:30pm or 7:00pm" slot_options = ["5:30pm", "7:00pm"] time_of_day = "evening" else: slots_text = "1:00pm or 3:00pm" slot_options = ["1:00pm", "3:00pm"] time_of_day = "afternoon" reply = ( f"Perfect, {name} – {pending_day} {time_of_day} works.\n\n" f"I've got {slots_text} available.\n\n" "Which would suit you better?" ) new_state["appointment_status"] = "INVITED" new_state["last_reply_type"] = "APPOINTMENT_TIME_CHOICES" new_state["pending_slot_options"] = slot_options new_state["pending_day_display"] = pending_day new_state["pending_day_date"] = pending_day_date return (reply, new_state) # ------------------------------------------------------------------------- # 3f. TIME_SELECTED_NEED_DAY: Customer chose morning/afternoon, needs day # ------------------------------------------------------------------------- if customer_state.get("last_reply_type") == "TIME_SELECTED_NEED_DAY": raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" pending_time = customer_state.get("pending_time_preference", "") # Parse day choice chosen_day = None if "saturday" in body_lower: chosen_day = "Saturday" elif "sunday" in body_lower: chosen_day = "Sunday" if chosen_day: day_date = compute_day_date(chosen_day.lower()) if pending_time == "morning": slots_text = "10:00am or 11:30am" slot_options = ["10:00am", "11:30am"] elif pending_time == "evening": slots_text = "5:30pm or 7:00pm" slot_options = ["5:30pm", "7:00pm"] else: slots_text = "1:00pm or 3:00pm" slot_options = ["1:00pm", "3:00pm"] reply = ( f"Perfect, {name} – {chosen_day} {pending_time} works.\n\n" f"I've got {slots_text} available.\n\n" "Which would suit you better?" ) new_state["appointment_status"] = "INVITED" new_state["last_reply_type"] = "APPOINTMENT_TIME_CHOICES" new_state["pending_slot_options"] = slot_options new_state["pending_day_display"] = chosen_day new_state["pending_day_date"] = day_date.isoformat() if day_date else None return (reply, new_state) # Check if customer chose weekday instead (new path from Style B spec) wants_weekday = any(w in body_lower for w in ["weekday", "week day", "monday", "tuesday", "wednesday", "thursday", "friday"]) if wants_weekday: # Redirect to weekday slot offers with preserved time preference today = date.today() next_monday = today + timedelta(days=(7 - today.weekday())) if pending_time == "morning": slot_time = "10:00am" elif pending_time == "evening": slot_time = "5:30pm" else: slot_time = "2:00pm" slot1_day = next_monday + timedelta(days=1) # Tuesday slot2_day = next_monday + timedelta(days=3) # Thursday day1_name = slot1_day.strftime("%A") day2_name = slot2_day.strftime("%A") day1_date_str = format_date_uk(slot1_day) day2_date_str = format_date_uk(slot2_day) reply = ( f"Perfect, {name} – I've got two {pending_time} slots next week:\n\n" f"• {day1_name} {day1_date_str} at {slot_time}\n" f"• {day2_name} {day2_date_str} at {slot_time}\n\n" "Which would suit you better?" ) new_state["appointment_status"] = "INVITED" new_state["last_reply_type"] = "NEXT_WEEK_SLOT_CHOICES" new_state["pending_next_week_slots"] = [ {"day": day1_name, "date": slot1_day.isoformat(), "time": slot_time}, {"day": day2_name, "date": slot2_day.isoformat(), "time": slot_time}, ] return (reply, new_state) # ------------------------------------------------------------------------- # 3f2. SPECIFIC_TIME_NEED_WEEKEND_DAY: Customer gave specific time, needs day # e.g., "11am" after weekend refinement → asking "Saturday or Sunday?" # ------------------------------------------------------------------------- if customer_state.get("last_reply_type") == "SPECIFIC_TIME_NEED_WEEKEND_DAY": raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" pending_time = customer_state.get("pending_specific_time", "") # Parse day choice chosen_day = None if "saturday" in body_lower: chosen_day = "Saturday" elif "sunday" in body_lower: chosen_day = "Sunday" if chosen_day: day_date = compute_day_date(chosen_day.lower()) date_str = format_date_uk(day_date) if day_date else "" # Book directly with the specific time (Style B: "that's ideal") reply = ( f"Thanks {name} – that's ideal.\n\n" f"I've booked you in for {chosen_day} {date_str} at {pending_time} at Lincoln Audi.\n\n" "If anything changes, just drop me a quick message here and I'll happily move it." ) new_state["appointment_status"] = "BOOKED" new_state["appointment_confirmed_text"] = inbound_text new_state["appointment_confirmed_at"] = datetime.now(timezone.utc).isoformat() new_state["appointment_date_display"] = chosen_day new_state["appointment_time_display"] = pending_time new_state["pending_day_date"] = day_date.isoformat() if day_date else None new_state["pending_day_display"] = chosen_day new_state["last_reply_type"] = "APPOINTMENT_CONFIRMED" return (reply, new_state) else: # Re-ask for weekend day reply = ( f"Sorry {name}, which day would you prefer – Saturday or Sunday?" ) new_state["last_reply_type"] = "SPECIFIC_TIME_NEED_WEEKEND_DAY" return (reply, new_state) # ------------------------------------------------------------------------- # 3g. WEEKDAY_SELECTED_NEED_TIME: From next week flow # ------------------------------------------------------------------------- if customer_state.get("last_reply_type") == "WEEKDAY_SELECTED_NEED_TIME": raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" # Parse time-of-day preference time_preference = "afternoon" if "morning" in body_lower: time_preference = "morning" elif "evening" in body_lower: time_preference = "evening" # Offer 2 weekday slots next week today = date.today() next_monday = today + timedelta(days=(7 - today.weekday())) if time_preference == "morning": slot_time = "10:00am" elif time_preference == "evening": slot_time = "5:30pm" else: slot_time = "2:00pm" slot1_day = next_monday + timedelta(days=1) # Tuesday slot2_day = next_monday + timedelta(days=3) # Thursday day1_name = slot1_day.strftime("%A") day2_name = slot2_day.strftime("%A") day1_date_str = format_date_uk(slot1_day) day2_date_str = format_date_uk(slot2_day) reply = ( f"Perfect, {name} – I've got two {time_preference} slots available:\n\n" f"• {day1_name} {day1_date_str} at {slot_time}\n" f"• {day2_name} {day2_date_str} at {slot_time}\n\n" "Which would suit you better?" ) new_state["appointment_status"] = "INVITED" new_state["last_reply_type"] = "NEXT_WEEK_SLOT_CHOICES" new_state["pending_next_week_slots"] = [ {"day": day1_name, "date": slot1_day.isoformat(), "time": slot_time}, {"day": day2_name, "date": slot2_day.isoformat(), "time": slot_time}, ] return (reply, new_state) # ------------------------------------------------------------------------- # 3h. TIME_SELECTED_NEED_WEEKDAY_OR_WEEKEND: From next week flow # ------------------------------------------------------------------------- if customer_state.get("last_reply_type") == "TIME_SELECTED_NEED_WEEKDAY_OR_WEEKEND": raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" pending_time = customer_state.get("pending_time_preference", "afternoon") wants_weekend = "weekend" in body_lower or "saturday" in body_lower or "sunday" in body_lower if wants_weekend: reply = ( f"Great, {name} – a weekend {pending_time} works well.\n\n" "Would Saturday or Sunday be better for you?" ) new_state["last_reply_type"] = "TIME_SELECTED_NEED_DAY" new_state["pending_time_preference"] = pending_time return (reply, new_state) else: # Weekday - offer 2 slots today = date.today() next_monday = today + timedelta(days=(7 - today.weekday())) if pending_time == "morning": slot_time = "10:00am" elif pending_time == "evening": slot_time = "5:30pm" else: slot_time = "2:00pm" slot1_day = next_monday + timedelta(days=1) # Tuesday slot2_day = next_monday + timedelta(days=3) # Thursday day1_name = slot1_day.strftime("%A") day2_name = slot2_day.strftime("%A") day1_date_str = format_date_uk(slot1_day) day2_date_str = format_date_uk(slot2_day) reply = ( f"Perfect, {name} – I've got two {pending_time} slots next week:\n\n" f"• {day1_name} {day1_date_str} at {slot_time}\n" f"• {day2_name} {day2_date_str} at {slot_time}\n\n" "Which would suit you better?" ) new_state["appointment_status"] = "INVITED" new_state["last_reply_type"] = "NEXT_WEEK_SLOT_CHOICES" new_state["pending_next_week_slots"] = [ {"day": day1_name, "date": slot1_day.isoformat(), "time": slot_time}, {"day": day2_name, "date": slot2_day.isoformat(), "time": slot_time}, ] return (reply, new_state) # ------------------------------------------------------------------------- # 3i. NEXT_WEEK_SLOT_CHOICES: Customer chooses from offered weekday slots # ------------------------------------------------------------------------- if customer_state.get("last_reply_type") == "NEXT_WEEK_SLOT_CHOICES": raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" pending_slots = customer_state.get("pending_next_week_slots", []) # Try to match customer's choice to one of the offered slots chosen_slot = None for slot in pending_slots: day_name = slot.get("day", "").lower() if day_name in body_lower: chosen_slot = slot break # Also check for "first", "second", "1", "2" style choices if not chosen_slot and len(pending_slots) >= 2: if any(w in body_lower for w in ["first", "1st", "tuesday"]): chosen_slot = pending_slots[0] elif any(w in body_lower for w in ["second", "2nd", "thursday"]): chosen_slot = pending_slots[1] if chosen_slot: day_name = chosen_slot.get("day", "") day_date_str = chosen_slot.get("date", "") slot_time = chosen_slot.get("time", "") # Format the date for display if day_date_str: try: day_date = datetime.strptime(day_date_str, "%Y-%m-%d").date() formatted_date = format_date_uk(day_date) except ValueError: formatted_date = "" else: formatted_date = "" reply = ( f"Thanks {name} – that's ideal.\n\n" f"I've booked you in for {day_name} {formatted_date} at {slot_time} at Lincoln Audi.\n\n" "If anything changes, just drop me a quick message here and I'll happily move it." ) new_state["appointment_status"] = "BOOKED" new_state["appointment_confirmed_text"] = inbound_text new_state["appointment_confirmed_at"] = datetime.now(timezone.utc).isoformat() new_state["appointment_date_display"] = day_name new_state["appointment_time_display"] = slot_time new_state["pending_day_date"] = day_date_str new_state["last_reply_type"] = "APPOINTMENT_CONFIRMED" return (reply, new_state) # ------------------------------------------------------------------------- # 3j. WEEKEND-ONLY INTENT: "This weekend" → Ask Saturday/Sunday + AM/PM # Must check BEFORE is_day_only_appointment since "this weekend" is in RELATIVE_WORDS # ------------------------------------------------------------------------- if is_weekend_only(body_lower): raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" reply = ( f"No problem, {name} – a weekend visit works well.\n\n" "Would Saturday or Sunday be better for you, and do you prefer morning or afternoon?" ) new_state["appointment_status"] = "INVITED" new_state["last_reply_type"] = "WEEKEND_REFINEMENT" return (reply, new_state) # ------------------------------------------------------------------------- # 3k. NEXT-WEEK-ONLY INTENT: "Next week" → Ask weekday/weekend + AM/PM # Must check BEFORE is_day_only_appointment since "next week" is in RELATIVE_WORDS # ------------------------------------------------------------------------- if is_next_week_only(body_lower): raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" reply = ( f"That's fine, {name} – we can look at next week.\n\n" "Are weekdays or the weekend better for you, and do you prefer morning or afternoon?" ) new_state["appointment_status"] = "INVITED" new_state["last_reply_type"] = "NEXT_WEEK_REFINEMENT" return (reply, new_state) # ------------------------------------------------------------------------- # 4. VAGUE TIME WINDOW (EARLY RETURN): Day + morning/afternoon → OFFER SLOTS # ------------------------------------------------------------------------- if is_vague_time_window(body_lower): raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" # Extract day display and compute calendar date day_display = extract_day_display(body_lower) day_date = compute_day_date(body_lower) # Offer time slots that match the time of day mentioned (Style B pattern) if "morning" in body_lower: slots_text = "10:00am or 11:30am" slot_options = ["10:00am", "11:30am"] time_of_day = "morning" elif "evening" in body_lower: slots_text = "5:30pm or 7:00pm" slot_options = ["5:30pm", "7:00pm"] time_of_day = "evening" else: # Default to afternoon (including explicit "afternoon" or unspecified) slots_text = "1:00pm or 3:00pm" slot_options = ["1:00pm", "3:00pm"] time_of_day = "afternoon" reply = ( f"Perfect, {name} – {day_display} {time_of_day} works.\n\n" f"I've got {slots_text} available.\n\n" "Which would suit you better?" ) new_state["appointment_status"] = "INVITED" new_state["last_reply_type"] = "APPOINTMENT_TIME_CHOICES" new_state["pending_slot_options"] = slot_options new_state["pending_slot_context"] = inbound_text new_state["pending_day_display"] = day_display new_state["pending_day_date"] = day_date.isoformat() if day_date else None return (reply, new_state) # ------------------------------------------------------------------------- # 4b. DAY-ONLY APPOINTMENT: Just a day, no time-of-day → OFFER MORNING/AFTERNOON # ------------------------------------------------------------------------- if is_day_only_appointment(body_lower): raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" # Extract day display and compute calendar date day_display = extract_day_display(body_lower) day_date = compute_day_date(body_lower) # Offer both morning and afternoon options since no preference given slots_text = "10:00am or 2:00pm" slot_options = ["10:00am", "2:00pm"] if day_display: reply = ( f"No problem, {name} – {day_display} works well.\n\n" f"Would you prefer {slots_text}?" ) else: reply = ( f"No problem, {name} – that works well.\n\n" f"Would you prefer {slots_text}?" ) new_state["appointment_status"] = "INVITED" new_state["last_reply_type"] = "APPOINTMENT_TIME_CHOICES" new_state["pending_slot_options"] = slot_options new_state["pending_slot_context"] = inbound_text new_state["pending_day_display"] = day_display new_state["pending_day_date"] = day_date.isoformat() if day_date else None return (reply, new_state) # ------------------------------------------------------------------------- # 5. HARD STOP (T3): Immediate opt-out # ------------------------------------------------------------------------- tier = classify_inbound_tier(body_lower) if tier == "T3_HARD_STOP": new_state["opted_out"] = True new_state["last_reply_type"] = "T3_HARD_STOP" reply = personalise_reply(T3_HARD_STOP_REPLY, customer_name) return (reply, new_state) # ------------------------------------------------------------------------- # 4. DIRECT QUESTION: Opening hours, visit duration, etc. # ------------------------------------------------------------------------- if is_direct_question(body_lower): new_state["appointment_status"] = "INVITED" new_state["last_reply_type"] = "DIRECT_QUESTION" reply = personalise_reply(DIRECT_QUESTION_REPLY, customer_name) return (reply, new_state) # ------------------------------------------------------------------------- # 5a. NOT INTERESTED PAUSE: "Not interested right now" (NOT opt-out) # Check BEFORE small talk because "thanks" can trigger small talk # Pause for 60 days, but do NOT set DNC # ------------------------------------------------------------------------- if is_not_interested_pause(body_lower): new_state["last_reply_type"] = "NOT_INTERESTED_PAUSE" new_state["status"] = "paused" new_state["defer_until"] = date.today() + timedelta(days=60) new_state["defer_reason"] = "not_interested_now" reply = personalise_reply(LIA_SOFT_NO_DEFER_TEMPLATES["NOT_INTERESTED_PAUSE"], customer_name) return (reply, new_state) # ------------------------------------------------------------------------- # 5b. BUSY NOW: Customer is temporarily busy (not a soft objection) # Keep status active, do NOT increment soft_no_count # ------------------------------------------------------------------------- if is_busy_now(body_lower): new_state["last_reply_type"] = "BUSY_NOW" reply = personalise_reply(LIA_SOFT_NO_DEFER_TEMPLATES["BUSY_NOW"], customer_name) return (reply, new_state) # ------------------------------------------------------------------------- # 5c. CALLBACK REQUESTED: Customer explicitly asks for phone call # Escalate to human with callback queue # ------------------------------------------------------------------------- if is_callback_request(body_lower): new_state["last_reply_type"] = "CALLBACK_REQUESTED" new_state["needs_human_review"] = True new_state["escalation_reason"] = "customer_requested_callback" reply = personalise_reply(LIA_SOFT_NO_DEFER_TEMPLATES["CALLBACK_REQUESTED"], customer_name) return (reply, new_state) # ------------------------------------------------------------------------- # 5d. DEFER REQUEST: Customer asks to be contacted at future time # Pause with parsed defer_until date # ------------------------------------------------------------------------- is_defer, defer_context = is_defer_request(body_lower) if is_defer: new_state["last_reply_type"] = "DEFER_CUSTOMER_REQUEST" new_state["status"] = "paused" new_state["defer_reason"] = f"customer_requested_defer: {defer_context}" new_state["defer_until"] = date.today() + timedelta(days=30) reply = personalise_reply(LIA_SOFT_NO_DEFER_TEMPLATES["DEFER_CUSTOMER_REQUEST"], customer_name) return (reply, new_state) # ------------------------------------------------------------------------- # 5e. TIME-OF-DAY ONLY: Customer says "morning" / "afternoon" without a day # This catches "How about in the morning?" after an appointment invite # Ask which day they mean # ------------------------------------------------------------------------- if is_time_of_day_only(body_lower): raw_name = customer_state.get("customer_name") name = raw_name.split()[0] if raw_name else "there" # Check if we already have a pending day from previous message pending_day = customer_state.get("pending_day_display") pending_day_date = customer_state.get("pending_day_date") if pending_day: # We have a day stored - confirm with time slots (Style B pattern) if "morning" in body_lower: slots_text = "10:00am or 11:30am" slot_options = ["10:00am", "11:30am"] time_of_day = "morning" elif "evening" in body_lower: slots_text = "5:30pm or 7:00pm" slot_options = ["5:30pm", "7:00pm"] time_of_day = "evening" else: slots_text = "1:00pm or 3:00pm" slot_options = ["1:00pm", "3:00pm"] time_of_day = "afternoon" reply = ( f"Perfect, {name} – {pending_day} {time_of_day} works.\n\n" f"I've got {slots_text} available.\n\n" "Which would suit you better?" ) new_state["appointment_status"] = "INVITED" new_state["last_reply_type"] = "APPOINTMENT_TIME_CHOICES" new_state["pending_slot_options"] = slot_options new_state["pending_slot_context"] = inbound_text # Keep the pending day from state new_state["pending_day_display"] = pending_day new_state["pending_day_date"] = pending_day_date else: # No day stored - ask which day if "morning" in body_lower: time_phrase = "in the morning" elif "evening" in body_lower: time_phrase = "in the evening" else: time_phrase = "in the afternoon" reply = ( f"{time_phrase.capitalize()} works well, {name}.\n\n" f"Which day {time_phrase} would suit you best?" ) new_state["appointment_status"] = "INVITED" new_state["last_reply_type"] = "APPOINTMENT_VAGUE_TIME" new_state["pending_time_preference"] = time_phrase return (reply, new_state) # ------------------------------------------------------------------------- # 5. SMALL TALK: Greetings, thanks, short messages # ------------------------------------------------------------------------- if is_small_talk(body_lower): if customer_state.get("appointment_status") != "BOOKED": new_state["appointment_status"] = "INVITED" new_state["last_reply_type"] = "SMALL_TALK" reply = personalise_reply(SMALL_TALK_REPLY, customer_name) return (reply, new_state) # ------------------------------------------------------------------------- # 6. TIER-BASED RESPONSES # ------------------------------------------------------------------------- # T6: SERVICE/SUPPORT if tier == "T6_SUPPORT": new_state["last_reply_type"] = "T6_SUPPORT" reply = personalise_reply(T6_SUPPORT_REPLY, customer_name) # Add urgent addon if needed if is_urgent_service(body_lower): reply += T6_SUPPORT_URGENT_ADDON return (reply, new_state) # T5: MONEY/AFFORDABILITY if tier == "T5_MONEY": new_state["last_reply_type"] = "T5_MONEY" # Check if they can't visit if detect_cant_visit(body_lower): reply = personalise_reply(T5_MONEY_PHONE_FALLBACK, customer_name) else: reply = personalise_reply(T5_MONEY_REPLY, customer_name) if customer_state.get("appointment_status") != "BOOKED": new_state["appointment_status"] = "INVITED" return (reply, new_state) # T4: SCHEDULED (contact later) if tier == "T4_SCHEDULED": new_state["last_reply_type"] = "T4_SCHEDULED" reply = personalise_reply(T4_SCHEDULED_REPLY, customer_name) return (reply, new_state) # T2: SOFT OBJECTION (two soft no rule) if tier == "T2_SOFT_OBJECTION": soft_no_count = customer_state.get("soft_no_count", 0) if soft_no_count <= 0: # First soft no - keep active, use new template new_state["soft_no_count"] = 1 new_state["last_reply_type"] = "T2_SOFT_NO_FIRST" reply = personalise_reply(LIA_SOFT_NO_DEFER_TEMPLATES["T2_SOFT_NO_FIRST"], customer_name) else: # Second soft no - park for 60 days with new template new_state["soft_no_count"] = 2 new_state["last_reply_type"] = "T2_SOFT_NO_FINAL" new_state["status"] = "paused" new_state["defer_until"] = date.today() + timedelta(days=60) new_state["defer_reason"] = "Two soft objections - parked for 60 days" reply = personalise_reply(LIA_SOFT_NO_DEFER_TEMPLATES["T2_SOFT_NO_FINAL"], customer_name) return (reply, new_state) # T1: GENERAL - positive interest messages or reasonable questions if tier == "T1_GENERAL": # Check for positive engagement indicators OR reasonable question patterns positive_indicators = [ "interested", "options", "yes", "tell me", "more info", "sounds good", "okay", "ok", "what", "how", "when", "why", "can you", "could you", "please", "help", "information", "details", "explain", "know more", "car", "audi", "finance", "deal", "offer", "upgrade", "renewal" ] if any(ind in body_lower for ind in positive_indicators) or len(body_lower) > 20: new_state["last_reply_type"] = "T1_GENERAL" if customer_state.get("appointment_status") != "BOOKED": new_state["appointment_status"] = "INVITED" # Reset confusion count on positive engagement new_state["confusion_count"] = 0 reply = personalise_reply(T1_GENERAL_REPLY, customer_name) return (reply, new_state) # ------------------------------------------------------------------------- # FINAL FALLBACK: Confusion → ask once → pause for manager # ------------------------------------------------------------------------- confusion_count = new_state.get("confusion_count", 0) if confusion_count == 0: # First time we genuinely don't understand → ask for clarity once new_state["confusion_count"] = 1 new_state["last_reply_type"] = "CLARITY_REQUEST" return (CLARITY_FIRST_REPLY, new_state) else: # Second time still unclear → pause and flag for human manager new_state["confusion_count"] = confusion_count + 1 new_state["needs_human_review"] = True new_state["last_reply_type"] = "ESCALATED_TO_MANAGER" # IMPORTANT: do not send anything to the customer here return ("", new_state) # ============================================================================= # TESTING # ============================================================================= if __name__ == "__main__": print("=" * 70) print("LIA UNIVERSAL REPLY ENGINE (URE) - TEST") print("=" * 70) test_state = { "customer_id": "test-123", "customer_name": "Sarah Johnson", "stage": "12", "soft_no_count": 0, "appointment_status": "NONE", "opted_out": False, "last_message": None, "last_reply_type": None, } test_cases = [ # Small talk ("Hi Lia, thanks for the message", "Small talk greeting"), ("Good morning", "Simple greeting"), # Direct questions ("When are you open?", "Opening hours question"), ("How long does a visit take?", "Duration question"), # Appointment confirmation ("I can do Saturday morning", "Appointment confirmation"), ("Tomorrow at 2pm works for me", "Specific time confirmation"), # General interest (T1) ("Tell me more about my options", "General interest"), ("Yes I'm interested", "Positive engagement"), # Money concerns (T5) ("How much would it cost?", "Cost question"), ("I can't afford higher payments", "Affordability concern"), ("Budget is tight but I live too far away", "Money + can't visit"), # Soft objections (T2) ("Not right now, maybe later", "First soft no"), # Scheduled (T4) ("Try me next month", "Contact later request"), # Hard stop (T3) ("Stop messaging me please", "Opt-out request"), # Support (T6) ("My car needs a service", "Service request"), ("Warning light just came on, urgent!", "Urgent service"), ] for message, description in test_cases: state = test_state.copy() # Special handling for second soft no test if description == "Second soft no": state["soft_no_count"] = 1 reply, new_state = generate_lia_reply(state, message) print(f"\n{'-' * 70}") print(f"TEST: {description}") print(f">>> Customer: {message}") print(f"<<< Lia:\n{reply}") print(f" [Type: {new_state.get('last_reply_type')}]") print(f" [Status: {new_state.get('appointment_status')}]") print(f" [Soft No Count: {new_state.get('soft_no_count')}]") if new_state.get("opted_out"): print(f" [⚠️ OPTED OUT]") # Test second soft no print(f"\n{'=' * 70}") print("TESTING TWO SOFT NO RULE:") print(f"{'=' * 70}") state = test_state.copy() # First soft no print("\n[First soft objection]") reply1, state = generate_lia_reply(state, "Not right now, bit busy") print(f">>> Customer: Not right now, bit busy") print(f"<<< Lia:\n{reply1}") print(f" Soft No Count: {state['soft_no_count']}") # Second soft no print("\n[Second soft objection - should park it]") reply2, state = generate_lia_reply(state, "Still not ready, maybe in a few months") print(f">>> Customer: Still not ready, maybe in a few months") print(f"<<< Lia:\n{reply2}") print(f" Soft No Count: {state['soft_no_count']}") print(f" [Should NOT ask another question - parked]") print(f"\n{'=' * 70}") print("URE TEST COMPLETE") print(f"{'=' * 70}\n")