import hashlib import hmac import json import re from dataclasses import dataclass from typing import Any import httpx from fastapi import HTTPException, status from sqlalchemy import select from sqlalchemy.orm import Session from app.core.config import get_settings from app.models.booking import VenueBookingSettings from app.models.venue import Venue from app.services.booking_assistant_service import is_greeting_only_message, maybe_handle_booking_message from app.services.booking_settings_service import ( adopt_legacy_whatsapp_configuration, build_whatsapp_assistant_prompt, get_booking_settings, get_effective_whatsapp_access_token, get_effective_whatsapp_phone_number_id, get_saved_whatsapp_access_token, get_saved_whatsapp_phone_number_id, uses_legacy_whatsapp_fallback_for_venue, uses_global_whatsapp_fallback, ) from app.services.whatsapp_contact_service import get_known_whatsapp_contact_name, remember_whatsapp_contact from app.services.time_context_service import build_time_context_system_message from app.models.whatsapp import WhatsAppAssistantTurn, WhatsAppEventLog from app.schemas.whatsapp import WhatsAppConfigValidationResponse, WhatsAppSendTestResponse, WhatsAppStatusResponse WEBHOOK_PATH = "/webhooks/whatsapp" INBOUND_MESSAGE_EVENT = "inbound_message" OUTBOUND_AGENT_EVENT = "outbound_agent" OUTBOUND_ERROR_EVENT = "outbound_error" ASSISTANT_ERROR_EVENT = "assistant_error" DETERMINISTIC_WHATSAPP_MODEL = "whatsapp-deterministic" CONTROL_TOKEN_PATTERN = re.compile(r"<\|[^>]+?\|>") STALE_INBOUND_GRACE_SECONDS = 90 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 @dataclass class WhatsAppGraphConfiguration: venue: Venue settings: VenueBookingSettings access_token: str | None phone_number_id: str | None def mask_value(value: str | None, visible_digits: int = 4) -> str | None: if not value: return None if len(value) <= visible_digits: return "*" * len(value) return f"{'*' * max(len(value) - visible_digits, 4)}{value[-visible_digits:]}" def normalize_phone_number(value: str) -> str: trimmed = value.strip() if not trimmed: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Numero di telefono non valido") has_plus = trimmed.startswith("+") digits = "".join(character for character in trimmed if character.isdigit()) if len(digits) < 6: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Numero di telefono non valido") return f"+{digits}" if has_plus else digits def get_whatsapp_configuration( *, db: Session, venue_id: int | None = None, phone_number_id: str | None = None, strict_phone_number_match: bool = False, ) -> WhatsAppGraphConfiguration: if phone_number_id: normalized_phone_number_id = phone_number_id.strip() if normalized_phone_number_id: stmt = ( select(VenueBookingSettings, Venue) .join(Venue, Venue.id == VenueBookingSettings.venue_id) .where(VenueBookingSettings.whatsapp_phone_number_id == normalized_phone_number_id) .limit(1) ) row = db.execute(stmt).first() if row is not None: settings, venue = row return WhatsAppGraphConfiguration( venue=venue, settings=settings, access_token=get_effective_whatsapp_access_token(settings, venue), phone_number_id=get_effective_whatsapp_phone_number_id(settings, venue), ) adopted_configuration = adopt_legacy_whatsapp_configuration( db, expected_phone_number_id=normalized_phone_number_id, ) if adopted_configuration is not None: settings, venue = adopted_configuration[1], adopted_configuration[0] return WhatsAppGraphConfiguration( venue=venue, settings=settings, access_token=get_effective_whatsapp_access_token(settings, venue), phone_number_id=get_effective_whatsapp_phone_number_id(settings, venue), ) if strict_phone_number_match: raise ValueError("Nessun locale associato al phone_number_id ricevuto") venue, booking_settings = get_booking_settings(db, venue_id=venue_id) return WhatsAppGraphConfiguration( venue=venue, settings=booking_settings, access_token=get_effective_whatsapp_access_token(booking_settings, venue), phone_number_id=get_effective_whatsapp_phone_number_id(booking_settings, venue), ) def get_whatsapp_status(*, db: Session, venue_id: int | None = None) -> WhatsAppStatusResponse: settings = get_settings() graph_configuration = get_whatsapp_configuration(db=db, venue_id=venue_id) configured = bool(graph_configuration.access_token and graph_configuration.phone_number_id and settings.whatsapp_verify_token) webhook_url = f"{settings.public_base_url.rstrip('/')}{WEBHOOK_PATH}" using_global_fallback = uses_global_whatsapp_fallback(graph_configuration.settings, graph_configuration.venue) message = ( "Configurazione WhatsApp pronta per invio test e webhook." if configured and not using_global_fallback else ( "Linea operativa, ma sta ancora usando un fallback legacy globale. Salva i dati direttamente nel locale per completare la migrazione." if configured and uses_legacy_whatsapp_fallback_for_venue(graph_configuration.venue) else "Completa token, phone number id o verify token per attivare la linea del locale." ) ) return WhatsAppStatusResponse( venue_id=graph_configuration.venue.id, venue_name=graph_configuration.venue.name, configured=configured, phone_number_id_masked=mask_value(graph_configuration.phone_number_id), phone_number_id=graph_configuration.phone_number_id or "", business_account_id=graph_configuration.settings.whatsapp_business_account_id or "", business_id=graph_configuration.settings.whatsapp_business_id or "", display_phone_number=graph_configuration.settings.whatsapp_display_phone_number or None, verified_name=graph_configuration.settings.whatsapp_verified_name or None, access_token_configured=bool(graph_configuration.access_token), phone_number_id_saved=bool(get_saved_whatsapp_phone_number_id(graph_configuration.settings)), access_token_saved=bool(get_saved_whatsapp_access_token(graph_configuration.settings)), uses_global_fallback=using_global_fallback, verify_token_configured=bool(settings.whatsapp_verify_token), app_secret_configured=bool(settings.whatsapp_app_secret), graph_api_version=settings.whatsapp_graph_api_version, webhook_path=WEBHOOK_PATH, webhook_url=webhook_url, message=message, ) def require_graph_api_configuration(*, db: Session, venue_id: int | None = None) -> WhatsAppGraphConfiguration: missing: list[str] = [] configuration = get_whatsapp_configuration(db=db, venue_id=venue_id) if not configuration.access_token: missing.append("whatsapp_access_token") if not configuration.phone_number_id: missing.append("whatsapp_phone_number_id") if missing: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Configurazione WhatsApp incompleta per il locale: {', '.join(missing)}", ) return configuration def require_webhook_configuration() -> Any: settings = get_settings() if not settings.whatsapp_verify_token: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Configurazione WhatsApp incompleta: WHATSAPP_VERIFY_TOKEN", ) return settings def build_graph_url(path: str) -> str: settings = get_settings() return f"{settings.whatsapp_api_base_url.rstrip('/')}/{settings.whatsapp_graph_api_version}/{path.lstrip('/')}" def extract_graph_error(response: httpx.Response) -> str: try: payload = response.json() except ValueError: return response.text or f"Errore Graph API {response.status_code}" error = payload.get("error") if isinstance(payload, dict) else None if isinstance(error, dict): message = error.get("message") or "Errore Graph API" code = error.get("code") error_type = error.get("type") details = " - ".join(str(part) for part in [error_type, code, message] if part not in (None, "")) return details or f"Errore Graph API {response.status_code}" return response.text or f"Errore Graph API {response.status_code}" def extract_http_error(response: httpx.Response, fallback_prefix: str) -> str: try: payload = response.json() except ValueError: detail = response.text.strip() else: detail = "" if isinstance(payload, dict): error_payload = payload.get("error") if isinstance(error_payload, dict): detail = str(error_payload.get("message") or error_payload.get("detail") or "").strip() if not detail: detail = str(payload.get("detail") or payload.get("message") or "").strip() if not detail: detail = json.dumps(payload, ensure_ascii=True) detail = " ".join(detail.split())[:400] if detail: return f"{fallback_prefix}: {detail}" return f"{fallback_prefix} ({response.status_code})" def create_log_entry( *, db: Session, event_type: str, summary: str, payload: dict, venue_id: int | None = None, contact_phone: str | None = None, wa_message_id: str | None = None, ) -> WhatsAppEventLog: log_entry = WhatsAppEventLog( venue_id=venue_id, event_type=event_type, contact_phone=contact_phone, wa_message_id=wa_message_id, summary=summary, payload=payload, ) db.add(log_entry) db.flush() return log_entry def create_assistant_turn( *, db: Session, venue_id: int, contact_phone: str, role: str, content: str, assistant_route: str | None = None, assistant_model: str | None = None, source_wa_message_id: str | None = None, trace: dict[str, object] | None = None, ) -> WhatsAppAssistantTurn | None: normalized_content = normalize_assistant_reply_content(content) if role == "assistant" else " ".join(content.split()) normalized_content = normalized_content.strip() if not normalized_content: return None turn = WhatsAppAssistantTurn( venue_id=venue_id, contact_phone=contact_phone, role=role, content=normalized_content, assistant_route=assistant_route, assistant_model=assistant_model, source_wa_message_id=source_wa_message_id, trace=trace or {}, ) db.add(turn) db.flush() return turn def has_event_log(db: Session, *, event_type: str, wa_message_id: str | None) -> bool: if not wa_message_id: return False stmt = ( select(WhatsAppEventLog.id) .where(WhatsAppEventLog.event_type == event_type, WhatsAppEventLog.wa_message_id == wa_message_id) .limit(1) ) return db.scalar(stmt) is not None def parse_unix_timestamp(value: object) -> int | None: if isinstance(value, int): return value if isinstance(value, str) and value.strip(): try: return int(value.strip()) except ValueError: return None return None def latest_inbound_timestamp(*, db: Session, venue_id: int, contact_phone: str) -> int | None: recent_logs = db.scalars( select(WhatsAppEventLog) .where( WhatsAppEventLog.event_type == INBOUND_MESSAGE_EVENT, WhatsAppEventLog.venue_id == venue_id, WhatsAppEventLog.contact_phone == contact_phone, ) .order_by(WhatsAppEventLog.created_at.desc(), WhatsAppEventLog.id.desc()) .limit(10) ) latest_timestamp: int | None = None for log_entry in recent_logs: payload = log_entry.payload if isinstance(log_entry.payload, dict) else {} candidate = parse_unix_timestamp(payload.get("timestamp") if isinstance(payload, dict) else None) if candidate is None: continue if latest_timestamp is None or candidate > latest_timestamp: latest_timestamp = candidate return latest_timestamp def is_stale_inbound_message(*, db: Session, venue_id: int, contact_phone: str, message: dict) -> tuple[bool, int | None, int | None]: incoming_timestamp = parse_unix_timestamp(message.get("timestamp")) if incoming_timestamp is None: return False, None, None latest_timestamp = latest_inbound_timestamp(db=db, venue_id=venue_id, contact_phone=contact_phone) if latest_timestamp is None: return False, incoming_timestamp, None is_stale = incoming_timestamp < (latest_timestamp - STALE_INBOUND_GRACE_SECONDS) return is_stale, incoming_timestamp, latest_timestamp def extract_message_text(message: dict) -> str | None: if message.get("type") != "text": return None text_body = message.get("text", {}).get("body") if not isinstance(text_body, str): return None cleaned = text_body.strip() return cleaned or None def extract_logged_reply(payload: dict) -> str | None: reply = payload.get("assistant_reply") if not isinstance(reply, str): return None return normalize_assistant_reply_content(reply) def build_conversation_history(db: Session, *, contact_phone: str, venue_id: int) -> list[dict[str, str]]: settings = get_settings() turn_stmt = ( select(WhatsAppAssistantTurn) .where( WhatsAppAssistantTurn.venue_id == venue_id, WhatsAppAssistantTurn.contact_phone == contact_phone, ) .order_by(WhatsAppAssistantTurn.created_at.desc(), WhatsAppAssistantTurn.id.desc()) .limit(max(settings.assistant_history_limit * 2, 8)) ) turns = list(db.scalars(turn_stmt)) if turns: history = [ {"role": turn.role, "content": turn.content} for turn in reversed(turns) if turn.role in {"user", "assistant"} and turn.content ] return history[-settings.assistant_history_limit :] stmt = ( select(WhatsAppEventLog) .where( WhatsAppEventLog.venue_id == venue_id, WhatsAppEventLog.contact_phone == contact_phone, WhatsAppEventLog.event_type.in_([INBOUND_MESSAGE_EVENT, OUTBOUND_AGENT_EVENT]), ) .order_by(WhatsAppEventLog.created_at.desc(), WhatsAppEventLog.id.desc()) .limit(max(settings.assistant_history_limit * 2, 8)) ) logs = list(db.scalars(stmt)) history: list[dict[str, str]] = [] for log_entry in reversed(logs): if log_entry.event_type == INBOUND_MESSAGE_EVENT: content = extract_message_text(log_entry.payload) role = "user" else: content = extract_logged_reply(log_entry.payload) role = "assistant" if not content: continue history.append({"role": role, "content": content}) return history[-settings.assistant_history_limit :] def list_whatsapp_conversations(*, db: Session, venue_id: int, limit: int = 20) -> list[dict[str, object]]: turns = list( db.scalars( select(WhatsAppAssistantTurn) .where(WhatsAppAssistantTurn.venue_id == venue_id) .order_by(WhatsAppAssistantTurn.created_at.desc(), WhatsAppAssistantTurn.id.desc()) .limit(max(limit * 30, 120)) ) ) grouped: list[dict[str, object]] = [] by_contact: dict[str, dict[str, object]] = {} for turn in turns: entry = by_contact.get(turn.contact_phone) if entry is None: if len(grouped) >= limit: continue entry = { "contact_phone": turn.contact_phone, "turn_count": 0, "last_turn_at": turn.created_at, "last_role": turn.role, "last_message": turn.content, "last_user_message": None, "last_assistant_message": None, } by_contact[turn.contact_phone] = entry grouped.append(entry) entry["turn_count"] = int(entry["turn_count"]) + 1 if turn.role == "user" and not entry["last_user_message"]: entry["last_user_message"] = turn.content if turn.role == "assistant" and not entry["last_assistant_message"]: entry["last_assistant_message"] = turn.content if grouped: return grouped recent_logs = list( db.scalars( select(WhatsAppEventLog) .where( WhatsAppEventLog.venue_id == venue_id, WhatsAppEventLog.contact_phone.is_not(None), WhatsAppEventLog.event_type.in_([INBOUND_MESSAGE_EVENT, OUTBOUND_AGENT_EVENT]), ) .order_by(WhatsAppEventLog.created_at.desc(), WhatsAppEventLog.id.desc()) .limit(max(limit * 20, 80)) ) ) fallback_grouped: list[dict[str, object]] = [] fallback_seen: set[str] = set() for log_entry in recent_logs: contact_phone = str(log_entry.contact_phone or "").strip() if not contact_phone or contact_phone in fallback_seen: continue payload = log_entry.payload if isinstance(log_entry.payload, dict) else {} last_message = extract_message_text(payload) if log_entry.event_type == INBOUND_MESSAGE_EVENT else extract_logged_reply(payload) fallback_grouped.append( { "contact_phone": contact_phone, "turn_count": 0, "last_turn_at": log_entry.created_at, "last_role": "user" if log_entry.event_type == INBOUND_MESSAGE_EVENT else "assistant", "last_message": last_message, "last_user_message": last_message if log_entry.event_type == INBOUND_MESSAGE_EVENT else None, "last_assistant_message": last_message if log_entry.event_type == OUTBOUND_AGENT_EVENT else None, } ) fallback_seen.add(contact_phone) if len(fallback_grouped) >= limit: break return fallback_grouped def list_whatsapp_assistant_turns( *, db: Session, venue_id: int, contact_phone: str, limit: int = 60, ) -> list[WhatsAppAssistantTurn]: normalized_phone = normalize_phone_number(contact_phone) return list( db.scalars( select(WhatsAppAssistantTurn) .where( WhatsAppAssistantTurn.venue_id == venue_id, WhatsAppAssistantTurn.contact_phone == normalized_phone, ) .order_by(WhatsAppAssistantTurn.created_at.desc(), WhatsAppAssistantTurn.id.desc()) .limit(limit) ) ) def summarize_conversation_history(conversation: list[dict[str, str]]) -> dict[str, object]: recent_messages = [] for entry in conversation[-6:]: if not isinstance(entry, dict): continue role = str(entry.get("role") or "").strip() content = " ".join(str(entry.get("content") or "").split()) if not role or not content: continue recent_messages.append( { "role": role, "content": content if len(content) <= 180 else f"{content[:177]}...", } ) return { "count": len(conversation), "recent": recent_messages, } def trim_reply_for_whatsapp(reply: str) -> str: settings = get_settings() cleaned = reply.replace("\r\n", "\n").strip() max_chars = max(settings.assistant_reply_max_chars, 200) if len(cleaned) <= max_chars: return cleaned truncated = cleaned[: max_chars - 1].rstrip() if " " in truncated: truncated = truncated.rsplit(" ", 1)[0].rstrip() return f"{truncated}…" def extract_content_from_json_payload(raw_text: str) -> str | None: candidate = raw_text.strip() if not candidate: return None if candidate.startswith("```"): candidate = re.sub(r"^```(?:json)?\s*", "", candidate) candidate = re.sub(r"\s*```$", "", candidate) json_candidates: list[str] = [] if candidate.startswith("{") and candidate.endswith("}"): json_candidates.append(candidate) start = candidate.find("{") end = candidate.rfind("}") if start != -1 and end != -1 and end > start: embedded = candidate[start : end + 1] if embedded not in json_candidates: json_candidates.append(embedded) for json_candidate in json_candidates: try: parsed = json.loads(json_candidate) except json.JSONDecodeError: continue if isinstance(parsed, dict): content = parsed.get("content") if isinstance(content, str) and content.strip(): return content.strip() return None def normalize_assistant_reply_content(reply: str) -> str | None: cleaned = reply.replace("\r\n", "\n").strip() if not cleaned: return None json_content = extract_content_from_json_payload(cleaned) if json_content: cleaned = json_content if "<|message|>" in cleaned: cleaned = cleaned.split("<|message|>", 1)[1].strip() nested_json_content = extract_content_from_json_payload(cleaned) if nested_json_content: cleaned = nested_json_content cleaned = CONTROL_TOKEN_PATTERN.sub("", cleaned).strip() if not cleaned: return None nested_json_content = extract_content_from_json_payload(cleaned) if nested_json_content: cleaned = nested_json_content cleaned = CONTROL_TOKEN_PATTERN.sub("", cleaned).strip() if not cleaned or cleaned.startswith('{"role"'): return None return cleaned def build_whatsapp_greeting_reply(sender_name: str | None) -> str: display_name = (sender_name or "").strip() if display_name: first_name = display_name.split()[0] return ( f"Ciao {first_name}! Ti aiuto volentieri con le prenotazioni. " "Se vuoi riservare un tavolo, scrivimi giorno, orario e numero di persone." ) return ( "Ciao! Ti aiuto volentieri con le prenotazioni. " "Se vuoi riservare un tavolo, scrivimi giorno, orario e numero di persone." ) def build_whatsapp_unclear_reply(sender_name: str | None) -> str: display_name = (sender_name or "").strip() if display_name: first_name = display_name.split()[0] return ( f"Scusami {first_name}, non ho capito bene la richiesta. " "Puoi riscrivermela indicando giorno, orario e numero di persone?" ) return ( "Scusami, non ho capito bene la richiesta. " "Puoi riscrivermela indicando giorno, orario e numero di persone?" ) def deliver_whatsapp_text_message( recipient: str, message: str, *, configuration: WhatsAppGraphConfiguration, ) -> tuple[str, dict, dict, str | None]: settings = get_settings() normalized_recipient = normalize_phone_number(recipient) clean_message = trim_reply_for_whatsapp(message) if not configuration.phone_number_id or not configuration.access_token: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Configurazione WhatsApp incompleta per il locale") endpoint = build_graph_url(f"{configuration.phone_number_id}/messages") request_body = { "messaging_product": "whatsapp", "recipient_type": "individual", "to": normalized_recipient, "type": "text", "text": { "preview_url": False, "body": clean_message, }, } with httpx.Client(timeout=settings.whatsapp_timeout_seconds) as client: response = client.post( endpoint, headers={ "Authorization": f"Bearer {configuration.access_token}", "Content-Type": "application/json", }, json=request_body, ) if response.status_code >= 400: error_detail = extract_graph_error(response) raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"Invio WhatsApp fallito: {error_detail}") payload = response.json() message_id = None messages = payload.get("messages") if isinstance(payload, dict) else None if isinstance(messages, list) and messages: first_message = messages[0] if isinstance(first_message, dict): message_id = first_message.get("id") return normalized_recipient, request_body, payload, message_id def request_assistant_reply( message: str, conversation: list[dict[str, str]], db: Session, *, venue_id: int, ) -> tuple[str, str]: settings = get_settings() try: venue, booking_settings = get_booking_settings(db, venue_id=venue_id) except ValueError as exc: raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc endpoint = f"{settings.assistant_api_base_url.rstrip('/')}/llm/openai/chat/completions" system_prompt = build_whatsapp_assistant_prompt(venue.name, booking_settings.whatsapp_assistant_prompt or "") time_context_prompt = build_time_context_system_message() request_body = { "messages": [ {"role": "system", "content": system_prompt}, {"role": "system", "content": time_context_prompt}, *conversation, {"role": "user", "content": message}, ], } with httpx.Client(timeout=settings.assistant_timeout_seconds) as client: response = client.post( endpoint, headers=build_internal_llm_headers(), json=request_body, ) if response.status_code >= 400: detail = extract_http_error(response, "Assistente non disponibile") raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=detail) payload = response.json() reply = None model = payload.get("model") if isinstance(payload, dict) else 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_payload = first_choice.get("message") if isinstance(message_payload, dict): reply = message_payload.get("content") if not reply: reply = first_choice.get("text") if not isinstance(reply, str) or not reply.strip(): raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="Assistente privo di risposta utilizzabile") normalized_reply = normalize_assistant_reply_content(reply) if not normalized_reply: raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="Assistente privo di risposta pulita utilizzabile") return trim_reply_for_whatsapp(normalized_reply), str(model or "") def maybe_personalize_booking_reply( reply: str, db: Session, *, venue_id: int, ) -> tuple[str, str | None]: base_reply = trim_reply_for_whatsapp(reply) try: venue, booking_settings = get_booking_settings(db, venue_id=venue_id) except ValueError: return base_reply, None custom_prompt = (booking_settings.whatsapp_assistant_prompt or "").strip() if not custom_prompt: return base_reply, None settings = get_settings() endpoint = f"{settings.assistant_api_base_url.rstrip('/')}/llm/openai/chat/completions" system_prompt = build_whatsapp_assistant_prompt(venue.name, custom_prompt) time_context_prompt = build_time_context_system_message() rewrite_guardrails = ( "Riceverai un messaggio prenotazioni gia verificato dal sistema. " "Riscrivilo applicando le istruzioni del locale, ma senza alterare alcun fatto operativo. " "Mantieni invariati disponibilita, stato della prenotazione, data, ora, durata, numero di persone, area, " "nome cliente, alternative proposte e call to action. " "Puoi pero riformulare la durata con un'unita di misura equivalente, per esempio da 120 minuti a 2 ore, " "se le istruzioni del locale lo richiedono. " "Se un'istruzione locale puo essere applicata solo cambiando la formulazione, fallo. " "Se richiederebbe di inventare o cambiare i fatti, ignorala. " "Restituisci solo il testo finale del messaggio da inviare al cliente." ) request_body = { "temperature": 0, "messages": [ {"role": "system", "content": system_prompt}, {"role": "system", "content": time_context_prompt}, {"role": "system", "content": rewrite_guardrails}, {"role": "user", "content": f"Riscrivi questo messaggio senza cambiare i fatti:\n\n{base_reply}"}, ], } try: with httpx.Client(timeout=settings.assistant_timeout_seconds) as client: response = client.post( endpoint, headers=build_internal_llm_headers(), json=request_body, ) except httpx.HTTPError: return base_reply, None if response.status_code >= 400: return base_reply, None payload = response.json() rewritten_reply = None model = payload.get("model") if isinstance(payload, dict) else 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_payload = first_choice.get("message") if isinstance(message_payload, dict): rewritten_reply = message_payload.get("content") if not rewritten_reply: rewritten_reply = first_choice.get("text") if not isinstance(rewritten_reply, str) or not rewritten_reply.strip(): return base_reply, None normalized_reply = normalize_assistant_reply_content(rewritten_reply) if not normalized_reply: return base_reply, None return trim_reply_for_whatsapp(normalized_reply), str(model or "") def reply_to_inbound_text_message( *, db: Session, venue_id: int, sender: str, sender_name: str | None, message: dict, conversation: list[dict[str, str]], ) -> None: incoming_text = extract_message_text(message) if not incoming_text: return assistant_base_reply: str | None = None assistant_model = DETERMINISTIC_WHATSAPP_MODEL assistant_route = "whatsapp-deterministic" assistant_trace: dict[str, object] = { "surface": "whatsapp", "incoming_text": incoming_text, "conversation": summarize_conversation_history(conversation), "source_message_id": message.get("id"), } try: normalized_sender = normalize_phone_number(sender) graph_configuration = require_graph_api_configuration(db=db, venue_id=venue_id) try: booking_result = maybe_handle_booking_message( db=db, sender_phone=normalized_sender, sender_name=sender_name, incoming_text=incoming_text, venue_id=venue_id, turn_duration_minutes=graph_configuration.settings.turn_duration_minutes, ) if booking_result is not None: assistant_base_reply = booking_result.reply assistant_reply, personalization_model = maybe_personalize_booking_reply( booking_result.reply, db, venue_id=venue_id, ) assistant_model = booking_result.assistant_model assistant_route = booking_result.route assistant_trace.update(booking_result.trace or {}) assistant_trace["route"] = assistant_route assistant_trace["personalized"] = bool(personalization_model) if personalization_model: assistant_model = f"{assistant_model}+prompt:{personalization_model}" assistant_trace["personalization_model"] = personalization_model elif is_greeting_only_message(incoming_text): assistant_reply = build_whatsapp_greeting_reply(sender_name) assistant_model = DETERMINISTIC_WHATSAPP_MODEL assistant_route = "greeting-only" assistant_base_reply = None assistant_trace["route"] = assistant_route else: assistant_reply, assistant_model = request_assistant_reply(incoming_text, conversation, db, venue_id=venue_id) assistant_route = "llm-freeform" assistant_base_reply = None assistant_trace["route"] = assistant_route except HTTPException as exc: assistant_reply = build_whatsapp_unclear_reply(sender_name) assistant_model = DETERMINISTIC_WHATSAPP_MODEL assistant_route = "fallback-unclear" assistant_base_reply = assistant_reply assistant_trace["route"] = assistant_route assistant_trace["error_detail"] = exc.detail create_log_entry( db=db, event_type=ASSISTANT_ERROR_EVENT, summary=f"Assistente in fallback verso {sender}: {exc.detail}", payload={ "source_message_id": message.get("id"), "incoming_message": message, "conversation": conversation, "detail": exc.detail, "fallback_reply": assistant_reply, "assistant_trace": assistant_trace, }, venue_id=venue_id, contact_phone=sender, wa_message_id=message.get("id"), ) recipient, request_body, response_payload, reply_message_id = deliver_whatsapp_text_message( sender, assistant_reply, configuration=graph_configuration, ) except HTTPException as exc: create_log_entry( db=db, event_type=ASSISTANT_ERROR_EVENT, summary=f"Risposta agente non inviata a {sender}: {exc.detail}", payload={ "source_message_id": message.get("id"), "incoming_message": message, "conversation": conversation, "detail": exc.detail, "assistant_trace": assistant_trace, }, venue_id=venue_id, contact_phone=sender, wa_message_id=message.get("id"), ) return create_log_entry( db=db, event_type=OUTBOUND_AGENT_EVENT, summary=f"Risposta agente inviata a {recipient}", payload={ "source_message_id": message.get("id"), "assistant_reply": assistant_reply, "assistant_base_reply": assistant_base_reply if assistant_base_reply and assistant_base_reply != assistant_reply else None, "assistant_model": assistant_model, "assistant_route": assistant_route, "assistant_trace": assistant_trace, "request": request_body, "response": response_payload, }, venue_id=venue_id, contact_phone=recipient, wa_message_id=reply_message_id, ) create_assistant_turn( db=db, venue_id=venue_id, contact_phone=recipient, role="assistant", content=assistant_reply, assistant_route=assistant_route, assistant_model=assistant_model, source_wa_message_id=reply_message_id, trace=assistant_trace, ) def validate_whatsapp_config(db: Session, *, venue_id: int | None = None) -> WhatsAppConfigValidationResponse: configuration = require_graph_api_configuration(db=db, venue_id=venue_id) endpoint = build_graph_url(f"{configuration.phone_number_id}?fields=display_phone_number,verified_name") with httpx.Client(timeout=get_settings().whatsapp_timeout_seconds) as client: response = client.get( endpoint, headers={"Authorization": f"Bearer {configuration.access_token}"}, ) if response.status_code >= 400: detail = extract_graph_error(response) raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"Verifica token WhatsApp fallita: {detail}") payload = response.json() detail = "Connessione a Meta riuscita." configuration.settings.whatsapp_display_phone_number = str(payload.get("display_phone_number") or "").strip() configuration.settings.whatsapp_verified_name = str(payload.get("verified_name") or "").strip() create_log_entry( db=db, event_type="config_validation", summary=detail, payload=payload, venue_id=configuration.venue.id, contact_phone=payload.get("display_phone_number"), ) db.commit() return WhatsAppConfigValidationResponse( success=True, phone_number_id=configuration.phone_number_id or "", display_phone_number=payload.get("display_phone_number"), verified_name=payload.get("verified_name"), detail=detail, ) def send_test_whatsapp_message(to: str, message: str, db: Session, *, venue_id: int | None = None) -> WhatsAppSendTestResponse: clean_message = message.strip() configuration = require_graph_api_configuration(db=db, venue_id=venue_id) try: recipient, request_body, payload, message_id = deliver_whatsapp_text_message( to, clean_message, configuration=configuration, ) except HTTPException as exc: create_log_entry( db=db, event_type=OUTBOUND_ERROR_EVENT, summary=f"Invio test fallito verso {to}: {exc.detail}", payload={"to": to, "message": clean_message, "detail": exc.detail}, venue_id=configuration.venue.id, contact_phone=normalize_phone_number(to), ) db.commit() raise exc create_log_entry( db=db, event_type="outbound_test", summary=f"Messaggio test inviato a {recipient}", payload={"request": request_body, "response": payload}, venue_id=configuration.venue.id, contact_phone=recipient, wa_message_id=message_id, ) db.commit() return WhatsAppSendTestResponse( success=True, recipient=recipient, message_id=message_id, detail="Messaggio test inviato correttamente.", ) def verify_webhook_handshake(mode: str | None, verify_token: str | None, challenge: str | None) -> str: settings = require_webhook_configuration() if mode != "subscribe" or verify_token != settings.whatsapp_verify_token or challenge is None: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Verifica webhook WhatsApp fallita") return challenge def validate_webhook_signature(body: bytes, signature: str | None) -> None: settings = get_settings() if not settings.whatsapp_app_secret: if settings.is_production: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="WHATSAPP_APP_SECRET non configurato", ) return if not signature: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Firma webhook WhatsApp mancante", ) expected = "sha256=" + hmac.new(settings.whatsapp_app_secret.encode("utf-8"), body, hashlib.sha256).hexdigest() if not hmac.compare_digest(expected, signature): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Firma webhook WhatsApp non valida", ) def summarize_incoming_message(message: dict, contacts_by_wa_id: dict[str, str]) -> tuple[str, str | None]: sender = message.get("from") sender_name = contacts_by_wa_id.get(sender or "", sender or "sconosciuto") message_type = message.get("type") or "unknown" if message_type == "text": text_body = message.get("text", {}).get("body") or "" summary = f"Messaggio in arrivo da {sender_name}: {text_body[:160]}".strip() else: summary = f"Messaggio {message_type} in arrivo da {sender_name}" return summary, sender def summarize_status_update(status_item: dict) -> tuple[str, str | None]: recipient = status_item.get("recipient_id") delivery_status = status_item.get("status") or "unknown" summary = f"Aggiornamento messaggio verso {recipient or 'sconosciuto'}: {delivery_status}" return summary, recipient def process_webhook_payload(payload: dict, db: Session) -> int: processed_events = 0 entries = payload.get("entry") if not isinstance(entries, list): create_log_entry(db=db, event_type="webhook_unknown", summary="Webhook WhatsApp senza entry", payload=payload) db.commit() return 1 for entry in entries: if not isinstance(entry, dict): continue for change in entry.get("changes", []): if not isinstance(change, dict): continue value = change.get("value", {}) if not isinstance(value, dict): continue metadata = value.get("metadata", {}) phone_number_id = metadata.get("phone_number_id") if isinstance(metadata, dict) else None venue_configuration = None if isinstance(phone_number_id, str) and phone_number_id.strip(): try: venue_configuration = get_whatsapp_configuration( db=db, phone_number_id=phone_number_id, strict_phone_number_match=True, ) except ValueError: venue_configuration = None contacts = value.get("contacts", []) contacts_by_wa_id = { contact.get("wa_id"): contact.get("profile", {}).get("name", contact.get("wa_id")) for contact in contacts if isinstance(contact, dict) and contact.get("wa_id") } messages = value.get("messages", []) for message in messages: if not isinstance(message, dict): continue if has_event_log(db, event_type=INBOUND_MESSAGE_EVENT, wa_message_id=message.get("id")): continue message_text = extract_message_text(message) summary, sender = summarize_incoming_message(message, contacts_by_wa_id) sender_name = contacts_by_wa_id.get(sender or "", sender or "sconosciuto") stored_contact_name = get_known_whatsapp_contact_name(sender, db) if sender else None if sender: remembered_contact = remember_whatsapp_contact( phone=sender, display_name=sender_name if sender_name and sender_name != sender else stored_contact_name, db=db, ) if remembered_contact is not None: sender_name = remembered_contact.display_name elif stored_contact_name: sender_name = stored_contact_name if sender and message_text and venue_configuration is not None: stale_message, incoming_timestamp, latest_timestamp = is_stale_inbound_message( db=db, venue_id=venue_configuration.venue.id, contact_phone=sender, message=message, ) if stale_message: create_log_entry( db=db, event_type="inbound_ignored", summary=f"Messaggio WhatsApp ignorato perche piu vecchio del contesto attuale per {sender}", payload={ "message": message, "incoming_timestamp": incoming_timestamp, "latest_processed_timestamp": latest_timestamp, }, venue_id=venue_configuration.venue.id, contact_phone=sender, wa_message_id=message.get("id"), ) processed_events += 1 continue create_log_entry( db=db, event_type=INBOUND_MESSAGE_EVENT, summary=summary, payload=message, venue_id=venue_configuration.venue.id if venue_configuration is not None else None, contact_phone=sender, wa_message_id=message.get("id"), ) if sender and message_text and venue_configuration is not None: create_assistant_turn( db=db, venue_id=venue_configuration.venue.id, contact_phone=sender, role="user", content=message_text, source_wa_message_id=message.get("id"), trace={"surface": "whatsapp", "source": "inbound-webhook"}, ) processed_events += 1 if sender and message_text and venue_configuration is not None: conversation = build_conversation_history( db, contact_phone=sender, venue_id=venue_configuration.venue.id, ) reply_to_inbound_text_message( db=db, venue_id=venue_configuration.venue.id, sender=sender, sender_name=sender_name, message=message, conversation=conversation, ) processed_events += 1 elif sender and message_text and venue_configuration is None: create_log_entry( db=db, event_type=ASSISTANT_ERROR_EVENT, summary="Messaggio WhatsApp ricevuto su una linea non associata a nessun locale", payload={"message": message, "phone_number_id": phone_number_id}, contact_phone=sender, wa_message_id=message.get("id"), ) processed_events += 1 statuses = value.get("statuses", []) for status_item in statuses: if not isinstance(status_item, dict): continue summary, recipient = summarize_status_update(status_item) create_log_entry( db=db, event_type="delivery_status", summary=summary, payload=status_item, venue_id=venue_configuration.venue.id if venue_configuration is not None else None, contact_phone=recipient, wa_message_id=status_item.get("id"), ) processed_events += 1 if processed_events == 0: create_log_entry( db=db, event_type="webhook_unknown", summary="Webhook WhatsApp ricevuto senza messaggi o stati", payload=payload, ) processed_events = 1 db.commit() return processed_events def parse_webhook_body(body: bytes) -> dict: if not body: return {} try: payload = json.loads(body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Payload webhook WhatsApp non valido") from exc if not isinstance(payload, dict): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Payload webhook WhatsApp non valido") return payload def list_whatsapp_logs(db: Session, limit: int = 50, *, venue_id: int | None = None) -> list[WhatsAppEventLog]: stmt = select(WhatsAppEventLog).order_by(WhatsAppEventLog.created_at.desc(), WhatsAppEventLog.id.desc()) if venue_id is not None: stmt = stmt.where(WhatsAppEventLog.venue_id == venue_id) stmt = stmt.limit(limit) return list(db.scalars(stmt))