from __future__ import annotations from dataclasses import dataclass from datetime import date, datetime, time, timedelta from difflib import SequenceMatcher import json import re import unicodedata from zoneinfo import ZoneInfo import httpx from fastapi import HTTPException from pydantic import BaseModel, Field, ValidationError from sqlalchemy import select from sqlalchemy.orm import Session, joinedload from app.core.config import get_settings from app.models.booking import WhatsAppBookingSession from app.models.customer import Customer from app.models.reservation import Reservation, ReservationSource, ReservationStatus from app.models.room import Room from app.models.table import Table from app.schemas.reservation import ReservationCreate, ReservationUpdate from app.services.assignment import ( RoomAssignmentCandidate, combine_date_time, list_available_room_candidates, reservations_for_assignment, resolve_preferred_room_ids, select_best_candidate_across_rooms, ) from app.services.contact_validation import normalize_customer_phone from app.services.reservation_service import delete_reservation, get_reservation_or_404, update_reservation from app.services.time_context_service import current_time_context ROME_TZ = ZoneInfo("Europe/Rome") SESSION_TIMEOUT_HOURS = 12 ALTERNATIVE_WINDOW_MINUTES = 180 ALTERNATIVE_LIMIT = 3 ALTERNATIVE_STEP_MINUTES = 15 BOOKING_MODEL_LABEL = "booking-automation" CANCELLATION_SESSION_STATUSES = {"cancellation_collecting", "cancellation_disambiguation"} MODIFICATION_SESSION_STATUSES = {"modification_lookup", "modification_update"} ACTIVE_SESSION_STATUSES = {"collecting", "awaiting_confirmation", *CANCELLATION_SESSION_STATUSES, *MODIFICATION_SESSION_STATUSES} FLOW_ABORT_PATTERNS = ( "lascia stare", "non importa", "stop", "ferma tutto", "annulla richiesta", ) def build_internal_llm_headers() -> dict[str, str]: settings = get_settings() headers = {"Content-Type": "application/json"} token = (settings.llm_proxy_internal_token or "").strip() if token: headers["X-Internal-API-Token"] = token return headers CANCELLATION_PATTERNS = ("cancell", "annull", "elimin", "disdic", "rimuov") MODIFICATION_PATTERNS = ( "modific", "cambia", "sposta", "aggiung", "togli", "riduc", "aument", "posticip", "anticip", "aggiorn", ) NEW_BOOKING_PATTERNS = ( "nuova prenotazione", "prenotazione nuova", "voglio prenotare", "vorrei prenotare", "devo fare una prenotazione", "voglio fare una prenotazione", "fare una prenotazione", "ne voglio fare una nuova", "voglio farne una nuova", "prenotare un tavolo", "voglio un tavolo", ) KEEP_SENDER_REFERENCE_PATTERNS = ( "va bene il mio numero", "il mio numero va bene", "usa il mio numero", "lascia il mio numero", "resta il mio numero", "resta questo numero", "va bene questo numero", "contattate me", "contatta me", "scrivete a me", "chiamate me", ) OTHER_CONTACT_REFERENCE_PATTERNS = ( "numero di paolo", "numero di lui", "numero di lei", "contattate lui", "contattate lei", "contattate paolo", "chiamate lui", "chiamate lei", "scrivete a lui", "scrivete a lei", ) PHONE_REFERENCE_KEYWORDS = ("numero", "telefono", "cell", "cellulare", "contatto", "whatsapp") GREETING_ONLY_PATTERNS = { "ciao", "ciao ciao", "salve", "buongiorno", "buonasera", "buon pomeriggio", "ehi", "hey", } OTHER_CUSTOMER_BOOKING_INFO_PATTERNS = ( re.compile(r"\bci\s+sono\s+prenotazion(?:i)?\s+per\s+(?:oggi|domani|dopodomani|stasera|staser[a]?|pranzo)\b", re.IGNORECASE), re.compile(r"\bci\s+sono\s+prenotazion(?:i)?\s+per\s+(?:lunedi|martedi|mercoledi|giovedi|venerdi|sabato|domenica)\b", re.IGNORECASE), re.compile(r"\bci\s+sono\s+altre?\s+prenotazion", re.IGNORECASE), re.compile(r"\bquante?\s+prenotazion(?:i)?\s+(?:ci\s+sono|avete|ci\s+avete)\b", re.IGNORECASE), re.compile(r"\bchi(?:\s+altro)?\s+ha\s+prenotat", re.IGNORECASE), re.compile(r"\ba\s+nome\s+di\s+chi\b", re.IGNORECASE), re.compile(r"\baltri?\s+client", re.IGNORECASE), re.compile(r"\baltre?\s+persone\b.*\bhanno\s+prenotat", re.IGNORECASE), ) SELF_RESERVATION_LOOKUP_PATTERNS = ( re.compile(r"\bho\s+gia\s+prenotazion(?:i)?\b", re.IGNORECASE), re.compile(r"\bho\s+prenotazion(?:i)?\s+attive?\b", re.IGNORECASE), re.compile(r"\bho\s+prenotat", re.IGNORECASE), re.compile(r"\brisulta\b.*\bprenotazion", re.IGNORECASE), re.compile(r"\bquali\b.*\bprenotazion(?:i)?\b.*\bho\b", re.IGNORECASE), re.compile(r"\bci\s+sono\s+prenotazion(?:i)?\s+a\s+mio\s+nome\b", re.IGNORECASE), ) ACTIVE_RESERVATION_STATUSES = ( ReservationStatus.pending, ReservationStatus.confirmed, ReservationStatus.seated, ) CONFIRM_PATTERNS = ( "confermo", "ok confermo", "va bene", "perfetto", "procedi", "prenota", "fissa", ) DECLINE_PATTERNS = ( "annulla", "cancella", "lascia stare", "non confermo", "non va bene", ) BOOKING_KEYWORD_STEMS = ( "prenot", "tavol", "posto", "copert", "person", "cancell", "annull", "elimin", "disdic", "rimuov", "modific", "cambia", "sposta", "aggiung", "togli", "riduc", "aument", "posticip", "anticip", "aggiorn", ) BOOKING_TEMPORAL_KEYWORDS = ( "domani", "dopodomani", "oggi", "stasera", "questa sera", ) BOOKING_KEYWORDS = re.compile( r"\b(prenot|tavol|posto|copert|persona|persone|domani|oggi|stasera|staser|sabato|domenica|lunedi|martedi|mercoledi|giovedi|venerdi|cancell|annull|elimin|disdic|rimuov|modific|cambia|sposta|aggiung|togli|riduc|aument|posticip|anticip|aggiorn|\d{1,2}:\d{2})", re.IGNORECASE, ) ITALIAN_SMALL_NUMBERS = { "zero": 0, "un": 1, "uno": 1, "una": 1, "due": 2, "tre": 3, "quattro": 4, "cinque": 5, "sei": 6, "sette": 7, "otto": 8, "nove": 9, "dieci": 10, } WEEKDAY_NAMES = ["lunedi", "martedi", "mercoledi", "giovedi", "venerdi", "sabato", "domenica"] MONTH_NAMES = [ "gennaio", "febbraio", "marzo", "aprile", "maggio", "giugno", "luglio", "agosto", "settembre", "ottobre", "novembre", "dicembre", ] BOOKING_TEMPORAL_KEYWORDS = (*BOOKING_TEMPORAL_KEYWORDS, *WEEKDAY_NAMES) class BookingExtraction(BaseModel): booking_related: bool = False intent: str = "other" reservation_date: str | None = None start_time: str | None = None guests: int | None = Field(default=None, ge=1, le=120) customer_name: str | None = None customer_phone: str | None = None area_preference: str | None = None notes: str | None = None confidence: str = "low" @dataclass class BookingAutomationResult: reply: str assistant_model: str route: str = "booking-automation" trace: dict[str, object] | None = None @dataclass class AvailabilitySnapshot: requested_options: list[RoomAssignmentCandidate] fallback_options: list[RoomAssignmentCandidate] preferred_room_ids: set[int] | None @property def has_requested_availability(self) -> bool: return bool(self.requested_options) @property def requested_room_names(self) -> list[str]: return [option.room_name for option in self.requested_options] @property def fallback_room_names(self) -> list[str]: return [option.room_name for option in self.fallback_options] class ReservationNoLongerAvailable(Exception): pass def venue_has_active_tables(*, venue_id: int, db: Session) -> bool: stmt = ( select(Table.id) .join(Room, Room.id == Table.room_id) .where(Room.venue_id == venue_id, Table.is_active.is_(True)) .limit(1) ) return db.execute(stmt).first() is not None def normalize_match_text(value: str) -> str: normalized = normalize_temporal_text(value) normalized = re.sub(r"[^a-z0-9]+", " ", normalized) return re.sub(r"\s+", " ", normalized).strip() def tokenize_for_matching(value: str) -> list[str]: normalized = normalize_match_text(value) if not normalized: return [] return normalized.split() def is_adjacent_transposition_variant(left: str, right: str) -> bool: if len(left) != len(right): return False mismatches = [index for index, (lval, rval) in enumerate(zip(left, right)) if lval != rval] if len(mismatches) != 2 or mismatches[1] != mismatches[0] + 1: return False first = mismatches[0] second = mismatches[1] return left[first] == right[second] and left[second] == right[first] def has_shared_prefix(left: str, right: str) -> bool: shortest = min(len(left), len(right)) if shortest <= 1: return left == right prefix_length = 2 if shortest >= 4 else 1 return left[:prefix_length] == right[:prefix_length] def fuzzy_token_match(token: str, expected: str, *, min_ratio: float = 0.86) -> bool: if token == expected: return True if not token or not expected: return False if len(expected) >= 4 and token.startswith(expected) and len(token) - len(expected) <= 1: return True if len(token) >= max(3, len(expected) - 1) and len(expected) >= len(token) and expected.startswith(token): return True if abs(len(token) - len(expected)) > 2: return False if not has_shared_prefix(token, expected): return False if is_adjacent_transposition_variant(token, expected): return True return SequenceMatcher(None, token, expected).ratio() >= min_ratio def text_contains_phrase_variant(text: str, phrase: str, *, min_ratio: float = 0.86) -> bool: message_tokens = tokenize_for_matching(text) phrase_tokens = tokenize_for_matching(phrase) if not message_tokens or not phrase_tokens: return False if len(phrase_tokens) == 1: return any(fuzzy_token_match(token, phrase_tokens[0], min_ratio=min_ratio) for token in message_tokens) window_size = len(phrase_tokens) for start in range(len(message_tokens) - window_size + 1): window = message_tokens[start : start + window_size] if all( fuzzy_token_match(window[index], phrase_tokens[index], min_ratio=min_ratio) for index in range(window_size) ): return True return False def text_contains_any_phrase_variant(text: str, phrases: tuple[str, ...], *, min_ratio: float = 0.86) -> bool: return any(text_contains_phrase_variant(text, phrase, min_ratio=min_ratio) for phrase in phrases) def text_contains_stem_variant(text: str, stem: str, *, min_ratio: float = 0.84) -> bool: for token in tokenize_for_matching(text): if token.startswith(stem): return True candidate = token[: min(len(token), len(stem) + 2)] if fuzzy_token_match(candidate, stem, min_ratio=min_ratio): return True return False def text_contains_any_stem_variant(text: str, stems: tuple[str, ...], *, min_ratio: float = 0.84) -> bool: return any(text_contains_stem_variant(text, stem, min_ratio=min_ratio) for stem in stems) def message_has_booking_keywords(incoming_text: str) -> bool: if BOOKING_KEYWORDS.search(incoming_text): return True normalized = normalize_match_text(incoming_text) if re.search(r"\b\d{1,2}:\d{2}\b", normalized): return True if text_contains_any_stem_variant(normalized, BOOKING_KEYWORD_STEMS): return True return text_contains_any_phrase_variant(normalized, BOOKING_TEMPORAL_KEYWORDS) def maybe_handle_booking_message( *, db: Session, sender_phone: str, sender_name: str | None, incoming_text: str, venue_id: int, turn_duration_minutes: int, ) -> BookingAutomationResult | None: session = load_active_session(sender_phone, venue_id=venue_id, db=db) if session is not None and session.status in {"confirmed", "cancelled"}: reset_session(session) db.flush() if session is not None and should_restart_as_new_booking(session, incoming_text): reset_session(session) db.flush() active_session = session if should_use_session_for_extraction(session) else None has_open_session = active_session is not None extraction, extraction_model = extract_booking_message( incoming_text=incoming_text, session=active_session, sender_name=sender_name, ) is_booking_flow = has_open_session or extraction.booking_related or message_has_booking_keywords(incoming_text) if not is_booking_flow: return None if is_self_reservation_lookup_request(incoming_text): reservations = find_reservations_for_self_lookup(venue_id=venue_id, sender_phone=sender_phone, db=db) return BookingAutomationResult( reply=build_self_reservation_lookup_reply(reservations), assistant_model=build_booking_model_label(extraction_model), ) session = get_or_create_session(sender_phone=sender_phone, venue_id=venue_id, db=db, session=session) if should_restart_as_new_booking(session, incoming_text): reset_session(session) if should_handle_modification_flow(session, extraction, incoming_text): if session.status not in MODIFICATION_SESSION_STATUSES: session.status = "modification_lookup" session.draft = {} modification_draft = normalize_draft(session.draft) if session.reservation_id is not None and not modification_draft.get("target_reservation_id"): modification_draft["target_reservation_id"] = session.reservation_id if is_meaningful_name(sender_name) and not modification_draft.get("customer_name"): modification_draft["customer_name"] = sender_name.strip() apply_modification_extraction_to_draft( modification_draft, extraction, incoming_text, session.status, ) if is_abort_flow_intent(incoming_text) and not extraction_has_modification_updates(extraction, incoming_text): session.status = "cancelled" session.draft = {} session.reservation_id = None db.flush() return BookingAutomationResult( reply="Va bene, interrompo la modifica. Se vuoi, scrivimi pure come posso aiutarti.", assistant_model=build_booking_model_label(extraction_model), ) target_reservation = resolve_modification_target( venue_id=venue_id, sender_phone=sender_phone, sender_name=sender_name, session=session, draft=modification_draft, db=db, ) if isinstance(target_reservation, str): session.status = "modification_lookup" session.draft = modification_draft db.flush() return BookingAutomationResult( reply=target_reservation, assistant_model=build_booking_model_label(extraction_model), ) modification_draft["target_reservation_id"] = target_reservation.id session.status = "modification_update" session.reservation_id = target_reservation.id modification_result = maybe_apply_modification( draft=modification_draft, reservation=target_reservation, db=db, ) if modification_result is not None: reply_text, is_completed = modification_result if is_completed: session.status = "confirmed" session.draft = {} session.reservation_id = target_reservation.id else: session.status = "modification_update" session.draft = modification_draft db.flush() return BookingAutomationResult( reply=reply_text, assistant_model=build_booking_model_label(extraction_model), ) session.draft = modification_draft db.flush() return BookingAutomationResult( reply=build_modification_prompt_reply(target_reservation), assistant_model=build_booking_model_label(extraction_model), ) if should_handle_cancellation_flow(session, extraction, incoming_text): if session.status not in CANCELLATION_SESSION_STATUSES: reset_session(session) session.status = "cancellation_collecting" session.draft = {} cancellation_draft = normalize_draft(session.draft) if is_meaningful_name(sender_name) and not cancellation_draft.get("customer_name"): cancellation_draft["customer_name"] = sender_name.strip() apply_cancellation_extraction_to_draft(cancellation_draft, extraction) if is_abort_flow_intent(incoming_text) and not extraction_has_cancellation_updates(extraction): session.status = "cancelled" session.draft = {} session.reservation_id = None db.flush() return BookingAutomationResult( reply="Va bene, interrompo la richiesta di cancellazione. Se ti serve altro, scrivimi pure.", assistant_model=build_booking_model_label(extraction_model), ) missing_cancellation_fields = compute_cancellation_missing_fields(cancellation_draft) if missing_cancellation_fields: session.status = "cancellation_collecting" session.draft = cancellation_draft db.flush() return BookingAutomationResult( reply=build_cancellation_missing_fields_reply(missing_cancellation_fields), assistant_model=build_booking_model_label(extraction_model), ) candidates = find_reservations_for_cancellation( venue_id=venue_id, draft=cancellation_draft, db=db, ) if not candidates: session.status = "cancellation_collecting" session.draft = cancellation_draft db.flush() return BookingAutomationResult( reply=build_cancellation_not_found_reply(cancellation_draft), assistant_model=build_booking_model_label(extraction_model), ) if len(candidates) > 1: session.status = "cancellation_disambiguation" session.draft = cancellation_draft db.flush() return BookingAutomationResult( reply=build_cancellation_disambiguation_reply(candidates, cancellation_draft), assistant_model=build_booking_model_label(extraction_model), ) deleted_snapshot = delete_reservation(candidates[0].id, db) session.status = "cancelled" session.draft = {} session.reservation_id = None db.flush() return BookingAutomationResult( reply=build_cancellation_success_reply(deleted_snapshot), assistant_model=build_booking_model_label(extraction_model), ) draft = normalize_draft(session.draft) draft["customer_phone"] = sender_phone draft["sender_phone"] = sender_phone if is_meaningful_name(sender_name) and not draft.get("customer_name"): draft["customer_name"] = sender_name.strip() if is_meaningful_name(sender_name): draft["sender_name"] = sender_name.strip() apply_extraction_to_draft(draft, extraction) draft["duration_minutes"] = turn_duration_minutes maybe_apply_area_preference_from_message( draft=draft, incoming_text=incoming_text, venue_id=venue_id, db=db, ) if is_greeting_only_message(incoming_text): if draft_has_meaningful_booking_context(draft): session.status = "collecting" session.draft = draft db.flush() return BookingAutomationResult( reply=build_greeting_booking_context_reply(compute_missing_fields(draft)), assistant_model=build_booking_model_label(extraction_model), ) return None if is_other_customer_booking_info_request(incoming_text): session.draft = draft db.flush() return BookingAutomationResult( reply=build_other_customer_booking_info_reply(draft), assistant_model=build_booking_model_label(extraction_model), ) if is_decline_intent(extraction, incoming_text) and not message_has_explicit_booking_updates(incoming_text): session.status = "cancelled" session.draft = {} session.reservation_id = None db.flush() return BookingAutomationResult( reply="Va bene, nessun problema: non blocco nulla. Se vuoi, posso aiutarti con un altro orario o con una nuova prenotazione.", assistant_model=build_booking_model_label(extraction_model), ) if session.status == "awaiting_confirmation" and is_confirmation_intent(extraction, incoming_text): missing_fields = compute_missing_fields(draft) if missing_fields: session.status = "collecting" session.draft = draft db.flush() return BookingAutomationResult( reply=build_missing_fields_reply(missing_fields), assistant_model=build_booking_model_label(extraction_model), ) if not venue_has_active_tables(venue_id=venue_id, db=db): session.status = "collecting" session.draft = draft db.flush() return BookingAutomationResult( reply=build_capacity_configuration_missing_reply(draft), assistant_model=build_booking_model_label(extraction_model), ) availability = build_availability_snapshot( venue_id=venue_id, reservation_date=parse_iso_date(draft["reservation_date"]), start_time=parse_hhmm_time(draft["start_time"]), guests=int(draft["guests"]), duration_minutes=int(draft["duration_minutes"]), area_preference=draft.get("area_preference"), exclude_reservation_id=None, db=db, ) if availability.has_requested_availability: reservation = create_confirmed_whatsapp_reservation( draft=draft, venue_id=venue_id, db=db, ) if reservation is not None: session.status = "confirmed" session.reservation_id = reservation.id session.draft = {} db.flush() return BookingAutomationResult( reply=build_confirmed_reply(draft, get_assigned_room_name(reservation)), assistant_model=build_booking_model_label(extraction_model), ) alternatives = suggest_alternative_slots( venue_id=venue_id, reservation_date=parse_iso_date(draft["reservation_date"]), requested_time=parse_hhmm_time(draft["start_time"]), guests=int(draft["guests"]), duration_minutes=int(draft["duration_minutes"]), area_preference=draft.get("area_preference"), db=db, ) draft["alternative_times"] = [value.strftime("%H:%M") for value in alternatives] session.status = "collecting" session.draft = draft db.flush() return BookingAutomationResult( reply=build_unavailable_reply( draft, alternatives, is_recheck=True, requested_area=str(draft.get("area_preference") or "").strip() or None, ), assistant_model=build_booking_model_label(extraction_model), ) missing_fields = compute_missing_fields(draft) if missing_fields: session.status = "collecting" session.draft = draft db.flush() return BookingAutomationResult( reply=build_missing_fields_reply(missing_fields), assistant_model=build_booking_model_label(extraction_model), ) reference_phone_reply = maybe_resolve_reference_phone_for_other_person( draft=draft, incoming_text=incoming_text, ) if reference_phone_reply is not None: session.status = "collecting" session.draft = draft db.flush() return BookingAutomationResult( reply=reference_phone_reply, assistant_model=build_booking_model_label(extraction_model), ) requested_date = parse_iso_date(draft["reservation_date"]) requested_time = parse_hhmm_time(draft["start_time"]) if not venue_has_active_tables(venue_id=venue_id, db=db): draft["alternative_times"] = [] session.status = "collecting" session.draft = draft db.flush() return BookingAutomationResult( reply=build_capacity_configuration_missing_reply(draft), assistant_model=build_booking_model_label(extraction_model), ) availability = build_availability_snapshot( venue_id=venue_id, reservation_date=requested_date, start_time=requested_time, guests=int(draft["guests"]), duration_minutes=int(draft["duration_minutes"]), area_preference=draft.get("area_preference"), exclude_reservation_id=None, db=db, ) if availability.preferred_room_ids and not availability.requested_options and availability.fallback_options: if len(availability.fallback_room_names) == 1: draft["area_preference"] = availability.fallback_room_names[0] draft["alternative_times"] = [] session.status = "awaiting_confirmation" if len(availability.fallback_room_names) == 1 else "collecting" session.draft = draft db.flush() return BookingAutomationResult( reply=build_preferred_area_unavailable_reply( draft, requested_area=str(draft.get("area_preference") or "").strip(), available_room_names=availability.fallback_room_names, ), assistant_model=build_booking_model_label(extraction_model), ) if len(availability.requested_room_names) > 1 and not availability.preferred_room_ids: draft["alternative_times"] = [] session.status = "collecting" session.draft = draft db.flush() return BookingAutomationResult( reply=build_area_preference_prompt(draft, availability.requested_room_names), assistant_model=build_booking_model_label(extraction_model), ) if availability.has_requested_availability: draft["alternative_times"] = [] session.status = "awaiting_confirmation" session.draft = draft db.flush() assigned_room_name = availability.requested_room_names[0] if availability.requested_room_names else None return BookingAutomationResult( reply=build_confirmation_reply(draft, assigned_room_name), assistant_model=build_booking_model_label(extraction_model), ) alternatives = suggest_alternative_slots( venue_id=venue_id, reservation_date=requested_date, requested_time=requested_time, guests=int(draft["guests"]), duration_minutes=int(draft["duration_minutes"]), area_preference=draft.get("area_preference"), db=db, ) draft["alternative_times"] = [value.strftime("%H:%M") for value in alternatives] session.status = "collecting" session.draft = draft db.flush() return BookingAutomationResult( reply=build_unavailable_reply( draft, alternatives, requested_area=str(draft.get("area_preference") or "").strip() or None, ), assistant_model=build_booking_model_label(extraction_model), ) def build_booking_model_label(extraction_model: str | None) -> str: if extraction_model: return f"{BOOKING_MODEL_LABEL}:{extraction_model}" return BOOKING_MODEL_LABEL def should_restart_as_new_booking(session: WhatsAppBookingSession, incoming_text: str) -> bool: if session.status not in (*CANCELLATION_SESSION_STATUSES, *MODIFICATION_SESSION_STATUSES, "collecting", "awaiting_confirmation"): return False if text_contains_any_stem_variant(incoming_text, CANCELLATION_PATTERNS): return False if text_contains_any_stem_variant(incoming_text, MODIFICATION_PATTERNS): return False return text_contains_any_phrase_variant(incoming_text, NEW_BOOKING_PATTERNS) def draft_has_meaningful_booking_context(draft: dict | None) -> bool: if not isinstance(draft, dict): return False meaningful_keys = ( "reservation_date", "start_time", "guests", "alternative_times", "lookup_reservation_date", "lookup_start_time", "new_reservation_date", "new_start_time", "new_guests", "pending_new_guests", "guest_delta", "target_reservation_id", ) return any(draft.get(key) not in (None, "", [], 0) for key in meaningful_keys) def should_use_session_for_extraction(session: WhatsAppBookingSession | None) -> bool: if session is None or session.status not in ACTIVE_SESSION_STATUSES: return False if session.status in CANCELLATION_SESSION_STATUSES or session.status in MODIFICATION_SESSION_STATUSES: return True return draft_has_meaningful_booking_context(session.draft) def should_handle_modification_flow( session: WhatsAppBookingSession, extraction: BookingExtraction, incoming_text: str, ) -> bool: if session.status in MODIFICATION_SESSION_STATUSES: return True has_modify_keyword = text_contains_any_stem_variant(incoming_text, MODIFICATION_PATTERNS) if not has_modify_keyword and extraction.intent != "modify": return False has_open_booking_draft = bool(session.draft) or session.status == "awaiting_confirmation" if has_open_booking_draft and session.status in {"collecting", "awaiting_confirmation"} and session.reservation_id is None: return False return True def should_handle_cancellation_flow( session: WhatsAppBookingSession, extraction: BookingExtraction, incoming_text: str, ) -> bool: if session.status in CANCELLATION_SESSION_STATUSES: return True has_cancel_keyword = text_contains_any_stem_variant(incoming_text, CANCELLATION_PATTERNS) if not has_cancel_keyword and extraction.intent != "cancel": return False has_open_booking_draft = bool(session.draft) or session.reservation_id is not None or session.status == "awaiting_confirmation" if has_open_booking_draft and session.status in {"collecting", "awaiting_confirmation"} and not message_has_explicit_cancellation_updates(incoming_text): return False return True def is_abort_flow_intent(incoming_text: str) -> bool: return text_contains_any_phrase_variant(incoming_text, FLOW_ABORT_PATTERNS) def normalize_lookup_phone(value: str | None) -> str | None: if not value: return None digits = "".join(character for character in value if character.isdigit()) return digits or None def normalize_lookup_name(value: str | None) -> str: return re.sub(r"\s+", " ", (value or "")).strip().casefold() def reservation_name_matches(reservation_name: str, lookup_name: str) -> bool: normalized_reservation = normalize_lookup_name(reservation_name) normalized_lookup = normalize_lookup_name(lookup_name) if not normalized_lookup: return False return ( normalized_reservation == normalized_lookup or normalized_lookup in normalized_reservation or normalized_reservation in normalized_lookup ) def get_or_create_session( *, sender_phone: str, venue_id: int, db: Session, session: WhatsAppBookingSession | None = None, ) -> WhatsAppBookingSession: if session is not None: return session legacy_session = load_latest_session_by_phone(sender_phone, db=db) if legacy_session is not None: if legacy_session.venue_id != venue_id: legacy_session.venue_id = venue_id reset_session(legacy_session) db.flush() return legacy_session session = WhatsAppBookingSession( contact_phone=sender_phone, venue_id=venue_id, status="collecting", draft={}, ) db.add(session) db.flush() return session def load_active_session(sender_phone: str, *, venue_id: int, db: Session) -> WhatsAppBookingSession | None: session = db.scalar( select(WhatsAppBookingSession).where( WhatsAppBookingSession.contact_phone == sender_phone, WhatsAppBookingSession.venue_id == venue_id, ) ) if session is None: return None stale_before = datetime.now(ROME_TZ) - timedelta(hours=SESSION_TIMEOUT_HOURS) updated_at = session.updated_at if updated_at is not None and updated_at.tzinfo is None: updated_at = updated_at.replace(tzinfo=ROME_TZ) if updated_at is not None and updated_at < stale_before and session.status not in {"confirmed", "cancelled"}: reset_session(session) db.flush() return session def load_latest_session_by_phone(sender_phone: str, *, db: Session) -> WhatsAppBookingSession | None: return db.scalar( select(WhatsAppBookingSession) .where(WhatsAppBookingSession.contact_phone == sender_phone) .order_by(WhatsAppBookingSession.updated_at.desc(), WhatsAppBookingSession.id.desc()) ) def reset_session(session: WhatsAppBookingSession) -> None: session.status = "collecting" session.draft = {} session.reservation_id = None def normalize_draft(raw_draft: dict | None) -> dict: if not isinstance(raw_draft, dict): return {} return dict(raw_draft) def apply_extraction_to_draft(draft: dict, extraction: BookingExtraction) -> None: if extraction.customer_name: draft["customer_name"] = extraction.customer_name.strip() if extraction.customer_phone: draft["customer_phone"] = extraction.customer_phone.strip() draft["reference_phone_confirmed"] = True if extraction.reservation_date and is_valid_iso_date(extraction.reservation_date): draft["reservation_date"] = extraction.reservation_date if extraction.start_time and is_valid_hhmm_time(extraction.start_time): draft["start_time"] = extraction.start_time if extraction.guests is not None: draft["guests"] = int(extraction.guests) if extraction.area_preference: draft["area_preference"] = extraction.area_preference.strip() if extraction.notes: draft["notes"] = extraction.notes.strip() def maybe_apply_area_preference_from_message( *, draft: dict, incoming_text: str, venue_id: int, db: Session, ) -> None: existing_preference = str(draft.get("area_preference") or "").strip() if existing_preference: matching_ids = resolve_preferred_room_ids(venue_id, existing_preference, db) if matching_ids: room_options = list(db.scalars(select(Room).where(Room.id.in_(matching_ids)).order_by(Room.name))) if len(room_options) == 1: draft["area_preference"] = room_options[0].name return inferred_ids = resolve_preferred_room_ids(venue_id, incoming_text, db) if not inferred_ids: return room_options = list(db.scalars(select(Room).where(Room.id.in_(inferred_ids)).order_by(Room.name))) if len(room_options) == 1: draft["area_preference"] = room_options[0].name def build_availability_snapshot( *, venue_id: int, reservation_date: date, start_time: time, guests: int, duration_minutes: int, area_preference: str | None, exclude_reservation_id: int | None, db: Session, ) -> AvailabilitySnapshot: scheduled = reservations_for_assignment(reservation_date, venue_id, db) if exclude_reservation_id is not None: scheduled = [reservation for reservation in scheduled if reservation.id != exclude_reservation_id] probe = Reservation( venue_id=venue_id, customer_id=0, reservation_date=reservation_date, start_time=start_time, duration_minutes=duration_minutes, guests=guests, status=ReservationStatus.confirmed, source=ReservationSource.whatsapp, notes=None, area_preference=area_preference, ) preferred_room_ids = resolve_preferred_room_ids(venue_id, area_preference, db) requested_options = list_available_room_candidates( probe, venue_id, scheduled, db, preferred_room_ids=preferred_room_ids, ) fallback_options = list_available_room_candidates( probe, venue_id, scheduled, db, ) return AvailabilitySnapshot( requested_options=requested_options, fallback_options=fallback_options, preferred_room_ids=preferred_room_ids, ) def apply_cancellation_extraction_to_draft(draft: dict, extraction: BookingExtraction) -> None: if extraction.customer_name: draft["customer_name"] = extraction.customer_name.strip() if extraction.customer_phone: draft["customer_phone"] = extraction.customer_phone.strip() if extraction.reservation_date and is_valid_iso_date(extraction.reservation_date): draft["reservation_date"] = extraction.reservation_date if extraction.start_time and is_valid_hhmm_time(extraction.start_time): draft["start_time"] = extraction.start_time def apply_modification_extraction_to_draft( draft: dict, extraction: BookingExtraction, incoming_text: str, session_status: str, ) -> None: explicit_date = extract_explicit_date(incoming_text) explicit_time = extract_explicit_time(incoming_text) explicit_guests = extract_explicit_guest_count(incoming_text) if extraction.customer_name: draft["customer_name"] = extraction.customer_name.strip() if extraction.customer_phone: draft["customer_phone"] = extraction.customer_phone.strip() guest_delta = extract_guest_delta(incoming_text) if guest_delta is not None: draft["guest_delta"] = guest_delta draft.pop("new_guests", None) draft.pop("pending_new_guests", None) elif explicit_guests is not None: draft.pop("guest_delta", None) if session_status == "modification_update" or draft.get("target_reservation_id"): if explicit_date and is_valid_iso_date(explicit_date): draft["new_reservation_date"] = explicit_date if explicit_time and is_valid_hhmm_time(explicit_time): draft["new_start_time"] = explicit_time if explicit_guests is not None and guest_delta is None: draft["new_guests"] = int(explicit_guests) return if explicit_date and is_valid_iso_date(explicit_date): draft["lookup_reservation_date"] = explicit_date if explicit_time and is_valid_hhmm_time(explicit_time): draft["lookup_start_time"] = explicit_time if explicit_guests is not None and guest_delta is None: draft["pending_new_guests"] = int(explicit_guests) def extraction_has_cancellation_updates(extraction: BookingExtraction) -> bool: return any( value not in (None, "", []) for value in [ extraction.reservation_date, extraction.start_time, extraction.customer_name, extraction.customer_phone, ] ) def extraction_has_modification_updates(extraction: BookingExtraction, incoming_text: str) -> bool: return any( value not in (None, "", []) for value in [ extract_explicit_date(incoming_text), extract_explicit_time(incoming_text), extract_explicit_guest_count(incoming_text), extraction.customer_name, extraction.customer_phone, extract_guest_delta(incoming_text), ] ) def extraction_has_booking_updates(extraction: BookingExtraction) -> bool: return any( value not in (None, "", []) for value in [ extraction.reservation_date, extraction.start_time, extraction.guests, extraction.customer_name, extraction.customer_phone, extraction.area_preference, extraction.notes, ] ) def message_has_explicit_cancellation_updates(incoming_text: str) -> bool: return any( value not in (None, "", []) for value in [ extract_explicit_date(incoming_text), extract_explicit_time(incoming_text), extract_explicit_name(incoming_text), extract_explicit_phone(incoming_text), ] ) def message_has_explicit_booking_updates(incoming_text: str) -> bool: return any( value not in (None, "", []) for value in [ extract_explicit_date(incoming_text), extract_explicit_time(incoming_text), extract_explicit_guest_count(incoming_text), extract_explicit_name(incoming_text), extract_explicit_phone(incoming_text), ] ) def is_confirmation_intent(extraction: BookingExtraction, incoming_text: str) -> bool: if extraction.intent == "confirm": return True return text_contains_any_phrase_variant(incoming_text, CONFIRM_PATTERNS) and not text_contains_any_phrase_variant( incoming_text, DECLINE_PATTERNS, ) def is_decline_intent(extraction: BookingExtraction, incoming_text: str) -> bool: if extraction.intent in {"decline", "cancel"}: return True return text_contains_any_phrase_variant(incoming_text, DECLINE_PATTERNS) def compute_missing_fields(draft: dict) -> list[str]: missing: list[str] = [] if not draft.get("customer_name") or not is_meaningful_name(str(draft.get("customer_name"))): missing.append("nome") if not draft.get("reservation_date"): missing.append("data") if not draft.get("start_time"): missing.append("orario") if draft.get("guests") in (None, "", 0): missing.append("numero di persone") return missing def build_missing_fields_reply(missing_fields: list[str]) -> str: if {"data", "orario", "numero di persone"}.issubset(set(missing_fields)): return ( "Scusami, non ho capito bene la richiesta. " "Puoi riscrivermela indicando giorno, orario e numero di persone?" ) if len(missing_fields) == 1: return f"Certo, per completare la prenotazione mi serve ancora questo dato: {missing_fields[0]}." joined = ", ".join(missing_fields[:-1]) + f" e {missing_fields[-1]}" return f"Certo, per completare la prenotazione mi servono ancora questi dati: {joined}." def is_other_customer_booking_info_request(incoming_text: str) -> bool: return any(pattern.search(incoming_text) for pattern in OTHER_CUSTOMER_BOOKING_INFO_PATTERNS) def build_other_customer_booking_info_reply(draft: dict) -> str: missing_fields = compute_missing_fields(draft) base_reply = ( "Mi dispiace, ma per privacy non posso condividere informazioni sulle prenotazioni di altri clienti." ) if not missing_fields: return f"{base_reply} Se vuoi, posso comunque controllare subito la disponibilita per la tua richiesta." if len(missing_fields) == 1: return ( f"{base_reply} Se vuoi, posso controllare la disponibilita per la tua richiesta: " f"mi serve ancora questo dato: {missing_fields[0]}." ) joined = ", ".join(missing_fields[:-1]) + f" e {missing_fields[-1]}" return ( f"{base_reply} Se vuoi, posso controllare la disponibilita per la tua richiesta: " f"mi servono ancora questi dati: {joined}." ) def is_greeting_only_message(incoming_text: str) -> bool: normalized = re.sub(r"[^\wÀ-ÿ]+", " ", incoming_text.casefold()).strip() return normalized in GREETING_ONLY_PATTERNS def build_greeting_booking_context_reply(missing_fields: list[str]) -> str: if not missing_fields: return "Ciao! Ti aiuto volentieri. Dimmi pure se vuoi confermare, modificare o cancellare la richiesta." if len(missing_fields) == 1: return f"Ciao! Ti aiuto volentieri. Per completare la prenotazione mi serve ancora questo dato: {missing_fields[0]}." joined = ", ".join(missing_fields[:-1]) + f" e {missing_fields[-1]}" return f"Ciao! Ti aiuto volentieri. Per completare la prenotazione mi servono ancora questi dati: {joined}." def is_self_reservation_lookup_request(incoming_text: str) -> bool: return any(pattern.search(incoming_text) for pattern in SELF_RESERVATION_LOOKUP_PATTERNS) def is_booking_for_other_person(draft: dict) -> bool: sender_name = draft.get("sender_name") customer_name = draft.get("customer_name") if not is_meaningful_name(str(sender_name or "")) or not is_meaningful_name(str(customer_name or "")): return False return not reservation_name_matches(str(sender_name), str(customer_name)) def build_reference_phone_confirmation_reply(draft: dict) -> str: customer_name = str(draft.get("customer_name") or "questa persona").strip() sender_phone = mask_phone_for_reply(draft.get("sender_phone") or draft.get("customer_phone")) return ( f"Perfetto, la prenotazione sara a nome {customer_name}. Possiamo lasciare come numero di riferimento il tuo " f"attuale ({sender_phone}), oppure se preferisci puoi inviarmi il numero della persona prenotata, cosi la " "contattiamo direttamente in caso di necessita." ) def build_invalid_reference_phone_reply() -> str: return ( "Per usarlo come numero di riferimento mi serve un recapito valido. Puoi inviarmelo in un formato corretto, " "per esempio +39 333 123 4567?" ) def message_looks_like_phone_attempt(incoming_text: str) -> bool: lowered = incoming_text.casefold() if any(keyword in lowered for keyword in PHONE_REFERENCE_KEYWORDS): return True return sum(character.isdigit() for character in incoming_text) >= 4 def maybe_resolve_reference_phone_for_other_person(draft: dict, incoming_text: str) -> str | None: if not is_booking_for_other_person(draft): draft.pop("reference_phone_confirmed", None) return None sender_phone = normalize_lookup_phone(draft.get("sender_phone")) current_customer_phone = normalize_lookup_phone(draft.get("customer_phone")) if current_customer_phone and sender_phone and current_customer_phone != sender_phone: draft["reference_phone_confirmed"] = True return None if draft.get("reference_phone_confirmed"): return None explicit_phone = extract_explicit_phone(incoming_text) if explicit_phone: draft["customer_phone"] = explicit_phone draft["reference_phone_confirmed"] = True return None lowered = incoming_text.casefold() if any(pattern in lowered for pattern in KEEP_SENDER_REFERENCE_PATTERNS): draft["customer_phone"] = draft.get("sender_phone") or draft.get("customer_phone") draft["reference_phone_confirmed"] = True return None if any(pattern in lowered for pattern in OTHER_CONTACT_REFERENCE_PATTERNS): customer_name = str(draft.get("customer_name") or "la persona prenotata").strip() return f"Perfetto, allora inviami pure il numero di {customer_name} da usare come contatto di riferimento." if message_looks_like_phone_attempt(incoming_text): return build_invalid_reference_phone_reply() return build_reference_phone_confirmation_reply(draft) def compute_cancellation_missing_fields(draft: dict) -> list[str]: missing: list[str] = [] if not draft.get("customer_name") or not is_meaningful_name(str(draft.get("customer_name"))): missing.append("nome della prenotazione") if not draft.get("reservation_date"): missing.append("giorno della prenotazione") return missing def build_cancellation_missing_fields_reply(missing_fields: list[str]) -> str: if len(missing_fields) == 1: return f"Certo, per cancellare la prenotazione mi serve ancora questo dato: {missing_fields[0]}." joined = ", ".join(missing_fields[:-1]) + f" e il {missing_fields[-1]}" return f"Certo, per cancellare la prenotazione mi servono ancora questi dati: {joined}." def build_cancellation_not_found_reply(draft: dict) -> str: booking_date = parse_iso_date(draft["reservation_date"]) customer_name = draft.get("customer_name") or "indicato" slot_text = format_date_italian(booking_date) if draft.get("start_time"): slot_text = f"{slot_text} alle {draft['start_time']}" return ( f"Non riesco a trovare una prenotazione attiva a nome {customer_name} per {slot_text}. " "Se vuoi, prova a indicarmi anche l'orario o il numero di telefono usato per prenotare, cosi controllo meglio." ) def mask_phone_for_reply(value: str | None) -> str: digits = normalize_lookup_phone(value) if not digits: return "telefono non disponibile" if len(digits) <= 4: return digits return f"...{digits[-4:]}" def build_cancellation_disambiguation_reply(candidates: list[Reservation], draft: dict) -> str: booking_date = parse_iso_date(draft["reservation_date"]) customer_name = draft.get("customer_name") or "indicato" options = [ f"{reservation.start_time.strftime('%H:%M')} ({mask_phone_for_reply(reservation.customer.phone)})" for reservation in candidates[:4] ] options_text = "; ".join(options) return ( f"Ho trovato piu prenotazioni a nome {customer_name} per {format_date_italian(booking_date)}. " f"Per capire qual e quella giusta da cancellare, puoi indicarmi l'orario esatto oppure il numero di telefono usato per prenotare? " f"Opzioni trovate: {options_text}." ) def build_cancellation_success_reply(snapshot: dict[str, object]) -> str: booking_date = parse_iso_date(str(snapshot["reservation_date"])) start_time = str(snapshot["start_time"]) return ( f"Perfetto, ho cancellato la prenotazione a nome {snapshot['customer_name']} " f"per {format_booking_slot(booking_date, start_time)}." ) def find_reservations_for_cancellation(*, venue_id: int, draft: dict, db: Session) -> list[Reservation]: reservation_date = parse_iso_date(draft["reservation_date"]) stmt = ( select(Reservation) .options(joinedload(Reservation.customer)) .join(Customer, Reservation.customer_id == Customer.id) .where( Reservation.venue_id == venue_id, Reservation.reservation_date == reservation_date, Reservation.status.in_(ACTIVE_RESERVATION_STATUSES), ) .order_by(Reservation.start_time, Reservation.id) ) candidates = list(db.scalars(stmt).unique()) lookup_name = str(draft.get("customer_name") or "").strip() if lookup_name: candidates = [ reservation for reservation in candidates if reservation.customer is not None and reservation_name_matches(reservation.customer.name, lookup_name) ] lookup_phone = normalize_lookup_phone(draft.get("customer_phone")) if lookup_phone: candidates = [ reservation for reservation in candidates if reservation.customer is not None and normalize_lookup_phone(reservation.customer.phone) is not None and normalize_lookup_phone(reservation.customer.phone).endswith(lookup_phone) ] lookup_time = draft.get("start_time") if lookup_time and is_valid_hhmm_time(str(lookup_time)): candidates = [ reservation for reservation in candidates if reservation.start_time.strftime("%H:%M") == str(lookup_time) ] return candidates def find_reservations_for_modification( *, venue_id: int, draft: dict, sender_phone: str, db: Session, ) -> list[Reservation]: today_local = datetime.now(ROME_TZ).date() stmt = ( select(Reservation) .options(joinedload(Reservation.customer)) .join(Customer, Reservation.customer_id == Customer.id) .where( Reservation.venue_id == venue_id, Reservation.status.in_(ACTIVE_RESERVATION_STATUSES), Reservation.reservation_date >= today_local, ) .order_by(Reservation.reservation_date, Reservation.start_time, Reservation.id) ) candidates = list(db.scalars(stmt).unique()) lookup_phone = normalize_lookup_phone(draft.get("customer_phone")) sender_lookup_phone = normalize_lookup_phone(sender_phone) if lookup_phone: candidates = [ reservation for reservation in candidates if reservation.customer is not None and normalize_lookup_phone(reservation.customer.phone) is not None and normalize_lookup_phone(reservation.customer.phone).endswith(lookup_phone) ] elif sender_lookup_phone: sender_matches = [ reservation for reservation in candidates if reservation.customer is not None and normalize_lookup_phone(reservation.customer.phone) is not None and normalize_lookup_phone(reservation.customer.phone).endswith(sender_lookup_phone) ] if sender_matches: candidates = sender_matches lookup_name = str(draft.get("customer_name") or "").strip() if lookup_name: candidates = [ reservation for reservation in candidates if reservation.customer is not None and reservation_name_matches(reservation.customer.name, lookup_name) ] lookup_date = draft.get("lookup_reservation_date") if lookup_date and is_valid_iso_date(str(lookup_date)): parsed_lookup_date = parse_iso_date(str(lookup_date)) candidates = [reservation for reservation in candidates if reservation.reservation_date == parsed_lookup_date] lookup_time = draft.get("lookup_start_time") if lookup_time and is_valid_hhmm_time(str(lookup_time)): candidates = [reservation for reservation in candidates if reservation.start_time.strftime("%H:%M") == str(lookup_time)] return candidates def find_reservations_for_self_lookup(*, venue_id: int, sender_phone: str, db: Session) -> list[Reservation]: today_local = datetime.now(ROME_TZ).date() sender_lookup_phone = normalize_lookup_phone(sender_phone) if not sender_lookup_phone: return [] stmt = ( select(Reservation) .options(joinedload(Reservation.customer)) .join(Customer, Reservation.customer_id == Customer.id) .where( Reservation.venue_id == venue_id, Reservation.status.in_(ACTIVE_RESERVATION_STATUSES), Reservation.reservation_date >= today_local, ) .order_by(Reservation.reservation_date, Reservation.start_time, Reservation.id) ) candidates = list(db.scalars(stmt).unique()) return [ reservation for reservation in candidates if reservation.customer is not None and normalize_lookup_phone(reservation.customer.phone) is not None and normalize_lookup_phone(reservation.customer.phone).endswith(sender_lookup_phone) ] def build_self_reservation_lookup_reply(reservations: list[Reservation]) -> str: if not reservations: return ( "Al momento non vedo prenotazioni attive associate a questo numero. " "Se vuoi, posso aiutarti subito a farne una nuova." ) if len(reservations) == 1: reservation = reservations[0] customer_name = reservation.customer.name if reservation.customer is not None else "te" return ( f"Sì, risulta una prenotazione a nome {customer_name} per " f"{format_booking_slot(reservation.reservation_date, reservation.start_time.strftime('%H:%M'))}, " f"{reservation.guests} persone. Se vuoi, posso anche aiutarti a modificarla o cancellarla." ) options = [ ( f"{format_booking_slot(reservation.reservation_date, reservation.start_time.strftime('%H:%M'))}, " f"{reservation.guests} persone" ) for reservation in reservations[:4] ] options_text = "; ".join(options) return ( f"Sì, risultano {len(reservations)} prenotazioni attive associate a questo numero: {options_text}. " "Se vuoi, posso aiutarti a modificarne o cancellarne una." ) def build_modification_prompt_reply(reservation: Reservation) -> str: customer_name = reservation.customer.name if reservation.customer is not None else "cliente" return ( f"Ho trovato la prenotazione a nome {customer_name} per " f"{format_booking_slot(reservation.reservation_date, reservation.start_time.strftime('%H:%M'))}, " f"{reservation.guests} persone. Dimmi pure cosa desideri modificare: giorno, orario o numero di persone." ) def build_modification_not_found_reply(draft: dict) -> str: lookup_date = draft.get("lookup_reservation_date") customer_name = draft.get("customer_name") or "indicato" if lookup_date and is_valid_iso_date(str(lookup_date)): slot_text = format_date_italian(parse_iso_date(str(lookup_date))) lookup_time = draft.get("lookup_start_time") if lookup_time and is_valid_hhmm_time(str(lookup_time)): slot_text = f"{slot_text} alle {lookup_time}" return ( f"Non riesco a trovare una prenotazione attiva da modificare per {customer_name} il {slot_text}. " "Se vuoi, prova a indicarmi anche l'orario o il numero di telefono usato per prenotare, cosi controllo meglio." ) return ( "Non riesco a trovare prenotazioni attive associate a questo numero. Se vuoi fare una nuova prenotazione, " "scrivimi pure giorno, orario e numero di persone." ) def build_modification_disambiguation_reply(candidates: list[Reservation]) -> str: options = [ ( f"{format_date_italian(reservation.reservation_date)} alle {reservation.start_time.strftime('%H:%M')} " f"({mask_phone_for_reply(reservation.customer.phone if reservation.customer else None)})" ) for reservation in candidates[:4] ] options_text = "; ".join(options) return ( "Ho trovato piu prenotazioni attive da modificare. " "Per capire qual e quella giusta, indicami il giorno esatto oppure l'orario o il numero di telefono usato per prenotare. " f"Opzioni trovate: {options_text}." ) def build_modification_success_reply(reservation: Reservation) -> str: customer_name = reservation.customer.name if reservation.customer is not None else "cliente" return ( f"Perfetto, ho aggiornato la prenotazione a nome {customer_name}: " f"{format_booking_slot(reservation.reservation_date, reservation.start_time.strftime('%H:%M'))}, " f"{reservation.guests} persone." ) def build_modification_unavailable_reply( *, reservation_date: date, start_time: time, guests: int, duration_minutes: int, alternatives: list[time], requested_area: str | None = None, ) -> str: slot_text = format_booking_slot(reservation_date, start_time.strftime("%H:%M")) area_text = f" in {requested_area}" if requested_area else "" if not alternatives: return ( f"Mi dispiace, al momento non ho disponibilita per la modifica richiesta su {slot_text}{area_text}, per {guests} persone con turno da " f"{duration_minutes} minuti. Se vuoi, indicami un altro orario o un'altra data." ) alternatives_text = ", ".join(slot.strftime("%H:%M") for slot in alternatives) return ( f"Mi dispiace, al momento non ho disponibilita per la modifica richiesta su {slot_text}{area_text}, per {guests} persone con turno da " f"{duration_minutes} minuti. Posso proporti questi orari: {alternatives_text}. " "Scrivimi pure quello che preferisci." ) def resolve_modification_target( *, venue_id: int, sender_phone: str, sender_name: str | None, session: WhatsAppBookingSession, draft: dict, db: Session, ) -> Reservation | str: target_reservation_id = draft.get("target_reservation_id") if target_reservation_id is not None: try: reservation = get_reservation_or_404(int(target_reservation_id), db) except HTTPException: draft.pop("target_reservation_id", None) else: if reservation.venue_id == venue_id and reservation.status in ACTIVE_RESERVATION_STATUSES: return reservation draft.pop("target_reservation_id", None) sender_candidates = find_reservations_for_modification( venue_id=venue_id, draft={}, sender_phone=sender_phone, db=db, ) has_lookup_details = any( draft.get(field) for field in ("lookup_reservation_date", "lookup_start_time", "customer_name", "customer_phone") ) if not has_lookup_details and len(sender_candidates) == 1: return sender_candidates[0] if not has_lookup_details and len(sender_candidates) > 1: return ( "Ho trovato piu prenotazioni attive associate a questo numero. " "Per capire quale vuoi modificare, dimmi il giorno della prenotazione e, se serve, anche l'orario." ) candidates = find_reservations_for_modification( venue_id=venue_id, draft=draft, sender_phone=sender_phone, db=db, ) if not candidates: return build_modification_not_found_reply(draft) if len(candidates) > 1: return build_modification_disambiguation_reply(candidates) return candidates[0] def maybe_apply_modification( *, draft: dict, reservation: Reservation, db: Session, ) -> tuple[str, bool] | None: new_guest_count = resolve_modified_guest_count(draft, reservation.guests) if new_guest_count is not None and new_guest_count < 1: draft.pop("new_guests", None) draft.pop("guest_delta", None) return ("Va bene, ma il numero di persone deve restare almeno 1. Indicami il totale aggiornato.", False) if new_guest_count is not None: draft["new_guests"] = new_guest_count draft.pop("pending_new_guests", None) payload_data: dict[str, object] = {} if draft.get("new_reservation_date") and is_valid_iso_date(str(draft["new_reservation_date"])): payload_data["reservation_date"] = parse_iso_date(str(draft["new_reservation_date"])) if draft.get("new_start_time") and is_valid_hhmm_time(str(draft["new_start_time"])): payload_data["start_time"] = parse_hhmm_time(str(draft["new_start_time"])) if new_guest_count is not None: payload_data["guests"] = new_guest_count pending_new_guests = draft.get("pending_new_guests") if "guests" not in payload_data and pending_new_guests not in (None, "", 0): payload_data["guests"] = int(pending_new_guests) if not payload_data: return None candidate_date = payload_data.get("reservation_date", reservation.reservation_date) candidate_time = payload_data.get("start_time", reservation.start_time) candidate_guests = int(payload_data.get("guests", reservation.guests)) actual_changes = ( candidate_date != reservation.reservation_date or candidate_time != reservation.start_time or candidate_guests != reservation.guests ) if not actual_changes: if draft.get("new_reservation_date") == reservation.reservation_date.isoformat(): draft.pop("new_reservation_date", None) if draft.get("new_start_time") == reservation.start_time.strftime("%H:%M"): draft.pop("new_start_time", None) if draft.get("new_guests") == reservation.guests: draft.pop("new_guests", None) return None if actual_changes: available = check_availability( venue_id=reservation.venue_id, reservation_date=candidate_date, start_time=candidate_time, guests=candidate_guests, duration_minutes=reservation.duration_minutes, area_preference=reservation.area_preference, exclude_reservation_id=reservation.id, db=db, ) if not available: alternatives = suggest_alternative_slots( venue_id=reservation.venue_id, reservation_date=candidate_date, requested_time=candidate_time, guests=candidate_guests, duration_minutes=reservation.duration_minutes, area_preference=reservation.area_preference, exclude_reservation_id=reservation.id, db=db, ) return ( build_modification_unavailable_reply( reservation_date=candidate_date, start_time=candidate_time, guests=candidate_guests, duration_minutes=reservation.duration_minutes, alternatives=alternatives, requested_area=reservation.area_preference, ), False, ) updated = update_reservation( reservation.id, ReservationUpdate(**payload_data), db, ) draft.clear() return (build_modification_success_reply(updated), True) def resolve_modified_guest_count(draft: dict, current_guests: int) -> int | None: if draft.get("new_guests") not in (None, "", 0): return int(draft["new_guests"]) guest_delta = draft.get("guest_delta") if guest_delta not in (None, ""): draft.pop("guest_delta", None) return current_guests + int(guest_delta) return None def check_availability( *, venue_id: int, reservation_date: date, start_time: time, guests: int, duration_minutes: int, area_preference: str | None = None, exclude_reservation_id: int | None = None, db: Session, ) -> bool: availability = build_availability_snapshot( venue_id=venue_id, reservation_date=reservation_date, start_time=start_time, guests=guests, duration_minutes=duration_minutes, area_preference=area_preference, exclude_reservation_id=exclude_reservation_id, db=db, ) if availability.preferred_room_ids: return availability.has_requested_availability return bool(availability.fallback_options) def suggest_alternative_slots( *, venue_id: int, reservation_date: date, requested_time: time, guests: int, duration_minutes: int, area_preference: str | None = None, exclude_reservation_id: int | None = None, db: Session, ) -> list[time]: seen: set[str] = set() suggestions: list[time] = [] requested_at = combine_date_time(reservation_date, requested_time) for step in range(ALTERNATIVE_STEP_MINUTES, ALTERNATIVE_WINDOW_MINUTES + ALTERNATIVE_STEP_MINUTES, ALTERNATIVE_STEP_MINUTES): for direction in (-1, 1): candidate_at = requested_at + timedelta(minutes=step * direction) if candidate_at.date() != reservation_date: continue slot = candidate_at.time().replace(second=0, microsecond=0) slot_key = slot.strftime("%H:%M") if slot_key in seen: continue seen.add(slot_key) if check_availability( venue_id=venue_id, reservation_date=reservation_date, start_time=slot, guests=guests, duration_minutes=duration_minutes, area_preference=area_preference, exclude_reservation_id=exclude_reservation_id, db=db, ): suggestions.append(slot) if len(suggestions) >= ALTERNATIVE_LIMIT: return suggestions return suggestions def format_room_names(room_names: list[str]) -> str: unique_names = list(dict.fromkeys(name.strip() for name in room_names if name and name.strip())) if not unique_names: return "" if len(unique_names) == 1: return unique_names[0] if len(unique_names) == 2: return f"{unique_names[0]} e {unique_names[1]}" return ", ".join(unique_names[:-1]) + f" e {unique_names[-1]}" def get_assigned_room_name(reservation: Reservation) -> str | None: if reservation.assigned_table is not None and reservation.assigned_table.room is not None: return reservation.assigned_table.room.name if reservation.assigned_combination is not None and reservation.assigned_combination.room is not None: return reservation.assigned_combination.room.name return None def build_area_preference_prompt(draft: dict, room_names: list[str]) -> str: booking_date = parse_iso_date(draft["reservation_date"]) rooms_text = format_room_names(room_names) return ( f"Ho disponibilita per {format_booking_slot(booking_date, draft['start_time'])}, " f"per {draft['guests']} persone, nelle aree {rooms_text}. " "Dimmi pure dove preferisci stare e poi te la confermo subito." ) def build_preferred_area_unavailable_reply( draft: dict, *, requested_area: str, available_room_names: list[str], ) -> str: booking_date = parse_iso_date(draft["reservation_date"]) available_rooms_text = format_room_names(available_room_names) if len(available_room_names) == 1: return ( f"Per {format_booking_slot(booking_date, draft['start_time'])} non ho disponibilita in {requested_area}, " f"ma posso proporti {available_rooms_text} per {draft['guests']} persone. " "Se per te va bene, rispondi pure CONFERMO e la blocco subito." ) return ( f"Per {format_booking_slot(booking_date, draft['start_time'])} non ho disponibilita in {requested_area}, " f"ma posso ancora proporti queste aree: {available_rooms_text}. " "Dimmi pure quale preferisci." ) def build_confirmation_reply(draft: dict, room_name: str | None = None) -> str: booking_date = parse_iso_date(draft["reservation_date"]) room_text = f" in {room_name}" if room_name else "" return ( f"Ho trovato disponibilita{room_text} per {format_booking_slot(booking_date, draft['start_time'])}, " f"per {draft['guests']} persone. La durata prevista del turno e di {draft['duration_minutes']} minuti. " "Se per te va bene, rispondi pure CONFERMO e la blocco subito." ) def build_unavailable_reply( draft: dict, alternatives: list[time], is_recheck: bool = False, *, requested_area: str | None = None, ) -> str: booking_date = parse_iso_date(draft["reservation_date"]) prefix = "Mi dispiace, quella disponibilita non e piu libera." if is_recheck else "Mi dispiace, in quell'orario non ho disponibilita." area_text = f" in {requested_area}" if requested_area else "" if not alternatives: return ( f"{prefix} Per {format_booking_slot(booking_date, draft['start_time'])}{area_text}, " f"con turno da {draft['duration_minutes']} minuti e {draft['guests']} persone, " "puoi indicarmi un altro orario o un'altra data." ) alternatives_text = ", ".join(slot.strftime("%H:%M") for slot in alternatives) return ( f"{prefix} Per {format_booking_slot(booking_date, draft['start_time'])}{area_text}, " f"con turno da {draft['duration_minutes']} minuti e {draft['guests']} persone, " f"posso proporti queste alternative: {alternatives_text}. Dimmi pure l'orario che preferisci." ) def build_capacity_configuration_missing_reply(draft: dict) -> str: booking_date = parse_iso_date(draft["reservation_date"]) return ( f"Mi dispiace, in questo momento non riesco ancora a verificare la disponibilita in modo affidabile per " f"{format_booking_slot(booking_date, draft['start_time'])}, con {draft['guests']} persone. " "Sto ancora aggiornando la configurazione della sala. Se vuoi, puoi riprovare piu tardi oppure indicarmi " "un altro orario e lo tengo come riferimento." ) def build_confirmed_reply(draft: dict, room_name: str | None = None) -> str: booking_date = parse_iso_date(draft["reservation_date"]) customer_name = draft.get("customer_name") or "cliente" room_text = f" in {room_name}" if room_name else "" return ( f"Perfetto, la prenotazione e confermata a nome {customer_name} per " f"{format_booking_slot(booking_date, draft['start_time'])}{room_text}, " f"{draft['guests']} persone. Se in seguito vuoi modificarla, puoi rispondere direttamente a questo messaggio." ) def format_booking_slot(booking_date: date, hhmm_value: str) -> str: return f"{format_date_italian(booking_date)} alle {hhmm_value}" def format_date_italian(value: date) -> str: return f"{WEEKDAY_NAMES[value.weekday()]} {value.day} {MONTH_NAMES[value.month - 1]} {value.year}" def is_meaningful_name(value: str | None) -> bool: if not value: return False if value.strip().casefold() in {"sconosciuto", "unknown"}: return False return any(character.isalpha() for character in value) def parse_iso_date(value: str) -> date: return date.fromisoformat(value) def parse_hhmm_time(value: str) -> time: return time.fromisoformat(value) def is_valid_iso_date(value: str) -> bool: try: date.fromisoformat(value) except ValueError: return False return True def is_valid_hhmm_time(value: str) -> bool: try: time.fromisoformat(value) except ValueError: return False return True def create_confirmed_whatsapp_reservation(draft: dict, venue_id: int, db: Session) -> Reservation | None: payload = ReservationCreate( venue_id=venue_id, customer_name=str(draft["customer_name"]).strip(), customer_phone=str(draft["customer_phone"]).strip(), reservation_date=parse_iso_date(draft["reservation_date"]), start_time=parse_hhmm_time(draft["start_time"]), duration_minutes=int(draft["duration_minutes"]), guests=int(draft["guests"]), status=ReservationStatus.confirmed, source=ReservationSource.whatsapp, notes=draft.get("notes"), area_preference=draft.get("area_preference"), ) return create_reservation_if_available(payload, db) def create_reservation_if_available(payload: ReservationCreate, db: Session) -> Reservation | None: from app.services.reservation_service import append_status_history, find_or_create_customer, get_reservation_or_404 reservation_id: int | None = None try: with db.begin_nested(): customer = find_or_create_customer(payload, db) reservation = Reservation( venue_id=payload.venue_id, customer_id=customer.id, reservation_date=payload.reservation_date, start_time=payload.start_time, duration_minutes=payload.duration_minutes, guests=payload.guests, status=payload.status, source=payload.source, notes=payload.notes, area_preference=payload.area_preference, ) db.add(reservation) db.flush() append_status_history(reservation, reservation.status, db) db.flush() scheduled = reservations_for_assignment(payload.reservation_date, payload.venue_id, db) preferred_room_ids = resolve_preferred_room_ids(payload.venue_id, payload.area_preference, db) candidate = select_best_candidate_across_rooms( reservation, payload.venue_id, scheduled, db, preferred_room_ids=preferred_room_ids, ) if candidate is None: raise ReservationNoLongerAvailable from app.services.assignment import assign_room_candidate assign_room_candidate(reservation, candidate, db) db.flush() reservation_id = reservation.id except ReservationNoLongerAvailable: return None if reservation_id is None: return None return get_reservation_or_404(reservation_id, db) def extract_booking_message( *, incoming_text: str, session: WhatsAppBookingSession | None, sender_name: str | None, ) -> tuple[BookingExtraction, str | None]: settings = get_settings() temporal_context = current_time_context() session_payload = { "status": session.status if session is not None else None, "draft": session.draft if session is not None else {}, "sender_name": sender_name, **temporal_context, "today": temporal_context["current_date"], } prompt = build_booking_extraction_prompt(incoming_text=incoming_text, session_payload=session_payload) try: with httpx.Client(timeout=settings.assistant_timeout_seconds) as client: response = client.post( f"{settings.assistant_api_base_url.rstrip('/')}/llm/openai/chat/completions", headers=build_internal_llm_headers(), json={ "temperature": 0, "messages": [ {"role": "system", "content": BOOKING_EXTRACTION_SYSTEM_PROMPT}, {"role": "user", "content": prompt}, ], }, ) response.raise_for_status() payload = response.json() except Exception: return apply_fallback_extraction(incoming_text, session), None model = payload.get("model") if isinstance(payload, dict) else None content = None if isinstance(payload, dict): choices = payload.get("choices") if isinstance(choices, list) and choices: first_choice = choices[0] if isinstance(first_choice, dict): message = first_choice.get("message") if isinstance(message, dict): content = message.get("content") if not isinstance(content, str) or not content.strip(): return apply_fallback_extraction(incoming_text, session), str(model or "") try: normalized = parse_json_object(content) extraction = BookingExtraction.model_validate(normalized) except (json.JSONDecodeError, ValidationError, ValueError): extraction = apply_fallback_extraction(incoming_text, session) enriched = apply_deterministic_overrides(extraction, incoming_text, session) return enriched, str(model or "") BOOKING_EXTRACTION_SYSTEM_PROMPT = """ Sei un parser rigoroso per prenotazioni di un locale. Restituisci SOLO JSON valido, senza testo aggiuntivo. Non inventare mai dati mancanti. Se un dato non e espresso in modo chiaro, restituisci null. Interpreta una conferma solo se il cliente conferma esplicitamente una proposta gia presente nel contesto. Schema richiesto: { "booking_related": boolean, "intent": "booking_request" | "confirm" | "decline" | "cancel" | "modify" | "info" | "other", "reservation_date": "YYYY-MM-DD" | null, "start_time": "HH:MM" | null, "guests": integer | null, "customer_name": string | null, "customer_phone": string | null, "area_preference": string | null, "notes": string | null, "confidence": "high" | "medium" | "low" } Regole: - booking_related=true solo se il messaggio riguarda una prenotazione, conferma, modifica o disponibilita. - intent=modify se il cliente vuole cambiare una prenotazione gia esistente, per esempio giorno, orario o numero di persone. - Se il cliente prenota a nome di un'altra persona, customer_name deve essere il nome della prenotazione, anche se diverso dal mittente. - Interpreta anche errori ortografici comuni del cliente se il significato resta chiaro. - Usa sempre il contesto temporale fornito dal sistema per interpretare riferimenti come oggi, domani, dopodomani, stasera e prossima settimana. - Se l'orario non e abbastanza preciso, usa null. - Se il nome non e esplicito, usa null. """.strip() def build_booking_extraction_prompt(*, incoming_text: str, session_payload: dict) -> str: return ( "Contesto sessione JSON:\n" f"{json.dumps(session_payload, ensure_ascii=True)}\n\n" "Messaggio cliente:\n" f"{incoming_text.strip()}" ) def parse_json_object(raw_content: str) -> dict: content = raw_content.strip() if content.startswith("```"): content = re.sub(r"^```(?:json)?\s*", "", content) content = re.sub(r"\s*```$", "", content) start = content.find("{") end = content.rfind("}") if start == -1 or end == -1 or end <= start: raise ValueError("JSON object non trovato") return json.loads(content[start : end + 1]) def apply_fallback_extraction(incoming_text: str, session: WhatsAppBookingSession | None) -> BookingExtraction: extracted_date = extract_explicit_date(incoming_text) extracted_time = extract_explicit_time(incoming_text) extracted_guests = extract_explicit_guest_count(incoming_text) booking_related = ( message_has_booking_keywords(incoming_text) or should_use_session_for_extraction(session) or extracted_date is not None or extracted_time is not None or extracted_guests is not None ) intent = "other" if text_contains_any_stem_variant(incoming_text, CANCELLATION_PATTERNS): intent = "cancel" elif text_contains_any_stem_variant(incoming_text, MODIFICATION_PATTERNS): intent = "modify" elif text_contains_any_phrase_variant(incoming_text, DECLINE_PATTERNS): intent = "decline" elif text_contains_any_phrase_variant(incoming_text, CONFIRM_PATTERNS): intent = "confirm" if session is not None else "booking_request" elif booking_related: intent = "booking_request" return BookingExtraction( booking_related=booking_related, intent=intent, reservation_date=extracted_date, start_time=extracted_time, guests=extracted_guests, customer_name=extract_explicit_name(incoming_text), customer_phone=extract_explicit_phone(incoming_text), confidence="low", ) def apply_deterministic_overrides( extraction: BookingExtraction, incoming_text: str, session: WhatsAppBookingSession | None, ) -> BookingExtraction: fallback = apply_fallback_extraction(incoming_text, session) payload = extraction.model_dump() if not payload["reservation_date"] and fallback.reservation_date: payload["reservation_date"] = fallback.reservation_date if not payload["start_time"] and fallback.start_time: payload["start_time"] = fallback.start_time if payload["guests"] is None and fallback.guests is not None: payload["guests"] = fallback.guests if not payload["customer_name"] and fallback.customer_name: payload["customer_name"] = fallback.customer_name if not payload["customer_phone"] and fallback.customer_phone: payload["customer_phone"] = fallback.customer_phone if payload["intent"] == "other" and fallback.intent != "other": payload["intent"] = fallback.intent if not payload["booking_related"] and fallback.booking_related: payload["booking_related"] = True return BookingExtraction.model_validate(payload) def normalize_temporal_text(value: str) -> str: normalized = unicodedata.normalize("NFKD", value or "").encode("ascii", "ignore").decode("ascii") return normalized.casefold() def extract_relative_datetime(incoming_text: str) -> datetime | None: normalized = normalize_temporal_text(incoming_text) now_local = datetime.now(ROME_TZ).replace(second=0, microsecond=0) patterns: list[tuple[str, str]] = [ (r"\b(?:tra|fra)\s+(un|una|uno|due|tre|quattro|cinque|sei|sette|otto|nove|dieci|\d{1,3})(?:\s+|')or[ae]\b", "hours"), (r"\b(?:tra|fra)\s+(un|una|uno|due|tre|quattro|cinque|sei|sette|otto|nove|dieci|\d{1,3})\s+minut[oi]\b", "minutes"), (r"\b(?:tra|fra)\s+mezz(?:'|a\s+)ora\b", "half_hour"), ] for pattern, unit in patterns: match = re.search(pattern, normalized, re.IGNORECASE) if not match: continue if unit == "half_hour": target = now_local + timedelta(minutes=30) else: amount = parse_small_number_token(match.group(1)) if amount is None: continue delta = timedelta(hours=amount) if unit == "hours" else timedelta(minutes=amount) target = now_local + delta minute_remainder = target.minute % 5 if minute_remainder: target += timedelta(minutes=5 - minute_remainder) return target.replace(second=0, microsecond=0) return None def extract_weekday_date(incoming_text: str, today_local: date) -> str | None: normalized = normalize_temporal_text(incoming_text) weekday_aliases = { "lunedi": 0, "martedi": 1, "mercoledi": 2, "giovedi": 3, "venerdi": 4, "sabato": 5, "domenica": 6, } matches = list( re.finditer( r"\b(?:quest[oa]\s+|prossim[oa]\s+)?(lunedi|martedi|mercoledi|giovedi|venerdi|sabato|domenica)\b", normalized, ) ) if not matches: tokens = tokenize_for_matching(normalized) for index, token in enumerate(tokens): matched_weekday = next( (weekday for weekday in WEEKDAY_NAMES if fuzzy_token_match(token, weekday, min_ratio=0.84)), None, ) if matched_weekday is None: continue target_weekday = weekday_aliases.get(matched_weekday) if target_weekday is None: return None days_ahead = (target_weekday - today_local.weekday()) % 7 prefix_tokens = tokens[max(0, index - 2) : index] if any(fuzzy_token_match(prefix, "prossimo", min_ratio=0.84) for prefix in prefix_tokens) and days_ahead == 0: days_ahead = 7 return (today_local + timedelta(days=days_ahead)).isoformat() return None match = matches[0] weekday_token = match.group(1) target_weekday = weekday_aliases.get(weekday_token) if target_weekday is None: return None days_ahead = (target_weekday - today_local.weekday()) % 7 prefix = normalized[max(0, match.start() - 12) : match.start()] if "prossim" in prefix and days_ahead == 0: days_ahead = 7 return (today_local + timedelta(days=days_ahead)).isoformat() def extract_explicit_date(incoming_text: str) -> str | None: lowered = normalize_temporal_text(incoming_text) today_local = datetime.now(ROME_TZ).date() relative_datetime = extract_relative_datetime(incoming_text) if relative_datetime is not None: return relative_datetime.date().isoformat() if text_contains_phrase_variant(lowered, "dopodomani", min_ratio=0.84): return (today_local + timedelta(days=2)).isoformat() if text_contains_phrase_variant(lowered, "domani", min_ratio=0.84): return (today_local + timedelta(days=1)).isoformat() if text_contains_any_phrase_variant(lowered, ("stasera", "questa sera"), min_ratio=0.84): return today_local.isoformat() if text_contains_phrase_variant(lowered, "oggi", min_ratio=0.84): return today_local.isoformat() weekday_date = extract_weekday_date(incoming_text, today_local) if weekday_date: return weekday_date iso_match = re.search(r"\b(20\d{2})-(\d{2})-(\d{2})\b", incoming_text) if iso_match: return f"{iso_match.group(1)}-{iso_match.group(2)}-{iso_match.group(3)}" slash_match = re.search(r"\b(\d{1,2})/(\d{1,2})(?:/(\d{2,4}))?\b", incoming_text) if slash_match: day_value = int(slash_match.group(1)) month_value = int(slash_match.group(2)) year_value = slash_match.group(3) if year_value is None: year = today_local.year elif len(year_value) == 2: year = 2000 + int(year_value) else: year = int(year_value) try: resolved = date(year, month_value, day_value) except ValueError: return None if year_value is None and resolved < today_local: try: resolved = date(today_local.year + 1, month_value, day_value) except ValueError: return None return resolved.isoformat() return None def extract_explicit_time(incoming_text: str) -> str | None: relative_datetime = extract_relative_datetime(incoming_text) if relative_datetime is not None: return relative_datetime.strftime("%H:%M") match = re.search(r"\b([01]?\d|2[0-3])[:.]([0-5]\d)\b", incoming_text) if match: hour_value = int(match.group(1)) minute_value = int(match.group(2)) return f"{hour_value:02d}:{minute_value:02d}" normalized = normalize_match_text(incoming_text) simple_match = re.search(r"\balle\s+([01]?\d|2[0-3])\b", normalized, re.IGNORECASE) hour_value: int | None = None if simple_match: hour_value = int(simple_match.group(1)) else: tokens = tokenize_for_matching(normalized) for index, token in enumerate(tokens[:-1]): if fuzzy_token_match(token, "alle", min_ratio=0.84) or token == "ore": next_token = tokens[index + 1] if re.fullmatch(r"([01]?\d|2[0-3])", next_token): hour_value = int(next_token) break if hour_value is None: return None lowered = normalized if hour_value <= 12 and text_contains_any_phrase_variant(lowered, ("sera", "stasera", "pomeriggio"), min_ratio=0.84): hour_value = 12 if hour_value == 12 else hour_value + 12 if hour_value <= 12: return None return f"{hour_value:02d}:00" def extract_explicit_guest_count(incoming_text: str) -> int | None: patterns = [ r"\bsiamo in (un|una|uno|due|tre|quattro|cinque|sei|sette|otto|nove|dieci|\d{1,3})\b", r"\bper (un|una|uno|due|tre|quattro|cinque|sei|sette|otto|nove|dieci|\d{1,3}) (?:persone|persona|coperti|coperto)\b", r"\b(un|una|uno|due|tre|quattro|cinque|sei|sette|otto|nove|dieci|\d{1,3}) (?:persone|persona|coperti|coperto)\b", r"\bin (un|una|uno|due|tre|quattro|cinque|sei|sette|otto|nove|dieci|\d{1,3})(?=\s*[?.!,]|$)\b", ] for pattern in patterns: match = re.search(pattern, incoming_text, re.IGNORECASE) if match: amount = parse_small_number_token(match.group(1)) if amount is not None: return amount return None def parse_small_number_token(value: str) -> int | None: cleaned = value.strip().casefold() if cleaned.isdigit(): return int(cleaned) return ITALIAN_SMALL_NUMBERS.get(cleaned) def extract_guest_delta(incoming_text: str) -> int | None: patterns: list[tuple[str, int]] = [ (r"\baggiung(?:i|ere)?\s+(un|una|uno|due|tre|quattro|cinque|sei|sette|otto|nove|dieci|\d{1,2})\s+(?:persona|persone|coperto|coperti)\b", 1), (r"\baument(?:a|are)?\s+(?:a\s+)?(un|una|uno|due|tre|quattro|cinque|sei|sette|otto|nove|dieci|\d{1,2})\s+(?:persona|persone|coperto|coperti)\b", 1), (r"\btogli(?:ere)?\s+(un|una|uno|due|tre|quattro|cinque|sei|sette|otto|nove|dieci|\d{1,2})\s+(?:persona|persone|coperto|coperti)\b", -1), (r"\briduc(?:i|e|ere)?\s+(?:di\s+)?(un|una|uno|due|tre|quattro|cinque|sei|sette|otto|nove|dieci|\d{1,2})\s+(?:persona|persone|coperto|coperti)\b", -1), ] for pattern, direction in patterns: match = re.search(pattern, incoming_text, re.IGNORECASE) if not match: continue amount = parse_small_number_token(match.group(1)) if amount is not None: return amount * direction return None def extract_explicit_name(incoming_text: str) -> str | None: patterns = [ r"\ba\s+nome\s+di\s+([A-Za-zÀ-ÿ][A-Za-zÀ-ÿ' .-]{1,80}?)(?=\s+(?:per|il|domani|oggi|dopodomani|alle)\b|[,.]|$)", r"\ba\s+nome\s+([A-Za-zÀ-ÿ][A-Za-zÀ-ÿ' .-]{1,80}?)(?=\s+(?:per|il|domani|oggi|dopodomani|alle)\b|[,.]|$)", r"\bper\s+conto\s+di\s+([A-Za-zÀ-ÿ][A-Za-zÀ-ÿ' .-]{1,80}?)(?=\s+(?:per|il|domani|oggi|dopodomani|alle)\b|[,.]|$)", r"\bnome\s+([A-Za-zÀ-ÿ][A-Za-zÀ-ÿ' .-]{1,80}?)(?=\s+(?:per|il|domani|oggi|dopodomani|alle)\b|[,.]|$)", ] for pattern in patterns: match = re.search(pattern, incoming_text, re.IGNORECASE) if not match: continue candidate = re.sub(r"\s+", " ", match.group(1)).strip(" .,-") if is_meaningful_name(candidate): return candidate return None def extract_explicit_phone(incoming_text: str) -> str | None: match = re.search(r"(\+?\d[\d\s().-]{5,}\d)", incoming_text) if not match: return None try: return normalize_customer_phone(match.group(1)) except ValueError: return None