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.schemas.booking import VenueBookingSettingsRead, VenueBookingSettingsUpdate from shared.assistant_profiles import compose_assistant_prompt, get_assistant_profile from shared.tenant_features import uses_legacy_whatsapp_fallback_for_tenant_slug DEFAULT_TURN_DURATION_MINUTES = 120 DEFAULT_WHATSAPP_ASSISTANT_BASE_PROMPT = get_assistant_profile("whatsapp").base_prompt def uses_legacy_whatsapp_fallback_for_venue(venue: Venue | None) -> bool: if venue is None: return False return uses_legacy_whatsapp_fallback_for_tenant_slug(venue.portal_tenant_slug) def get_default_venue(db: Session) -> Venue: venue = db.scalar(select(Venue).order_by(Venue.id)) if venue is None: raise ValueError("Nessun locale configurato") return venue def ensure_booking_settings(venue_id: int, db: Session) -> VenueBookingSettings: settings = db.scalar(select(VenueBookingSettings).where(VenueBookingSettings.venue_id == venue_id)) if settings is not None: if settings.whatsapp_assistant_prompt is None: settings.whatsapp_assistant_prompt = "" if settings.whatsapp_business_account_id is None: settings.whatsapp_business_account_id = "" if settings.whatsapp_business_id is None: settings.whatsapp_business_id = "" if settings.whatsapp_phone_number_id is None: settings.whatsapp_phone_number_id = "" if settings.whatsapp_access_token is None: settings.whatsapp_access_token = "" if settings.whatsapp_display_phone_number is None: settings.whatsapp_display_phone_number = "" if settings.whatsapp_verified_name is None: settings.whatsapp_verified_name = "" db.flush() return settings settings = VenueBookingSettings( venue_id=venue_id, turn_duration_minutes=DEFAULT_TURN_DURATION_MINUTES, whatsapp_assistant_prompt="", whatsapp_business_account_id="", whatsapp_business_id="", whatsapp_phone_number_id="", whatsapp_access_token="", whatsapp_display_phone_number="", whatsapp_verified_name="", ) db.add(settings) db.flush() return settings def get_saved_whatsapp_access_token(settings: VenueBookingSettings) -> str | None: cleaned = settings.whatsapp_access_token.strip() return cleaned or None def get_saved_whatsapp_phone_number_id(settings: VenueBookingSettings) -> str | None: cleaned = settings.whatsapp_phone_number_id.strip() return cleaned or None def uses_global_whatsapp_fallback(settings: VenueBookingSettings, venue: Venue) -> bool: if not uses_legacy_whatsapp_fallback_for_venue(venue): return False settings_config = get_settings() global_access_token = (settings_config.whatsapp_access_token or "").strip() global_phone_number_id = (settings_config.whatsapp_phone_number_id or "").strip() return bool( (global_access_token and not get_saved_whatsapp_access_token(settings)) or (global_phone_number_id and not get_saved_whatsapp_phone_number_id(settings)) ) def _select_auto_adoptable_legacy_venue(db: Session, *, preferred_venue: Venue | None = None) -> Venue | None: if preferred_venue is not None: if preferred_venue.portal_tenant_id: tenant_venue_ids = list( db.scalars(select(Venue.id).where(Venue.portal_tenant_id.is_not(None)).order_by(Venue.id).limit(2)) ) if tenant_venue_ids != [preferred_venue.id]: return None else: venue_ids = list(db.scalars(select(Venue.id).order_by(Venue.id).limit(2))) if venue_ids != [preferred_venue.id]: return None return preferred_venue tenant_venues = list( db.scalars(select(Venue).where(Venue.portal_tenant_id.is_not(None)).order_by(Venue.id).limit(2)) ) if len(tenant_venues) == 1: return tenant_venues[0] venues = list(db.scalars(select(Venue).order_by(Venue.id).limit(2))) if len(venues) == 1: return venues[0] return None def adopt_legacy_whatsapp_configuration( db: Session, *, preferred_venue: Venue | None = None, expected_phone_number_id: str | None = None, ) -> tuple[Venue, VenueBookingSettings] | None: settings_config = get_settings() global_phone_number_id = (settings_config.whatsapp_phone_number_id or "").strip() global_access_token = (settings_config.whatsapp_access_token or "").strip() if not global_phone_number_id: return None normalized_expected_phone_number_id = (expected_phone_number_id or "").strip() if normalized_expected_phone_number_id and normalized_expected_phone_number_id != global_phone_number_id: return None candidate_venue = _select_auto_adoptable_legacy_venue(db, preferred_venue=preferred_venue) if candidate_venue is None: return None existing_owner_id = db.scalar( select(VenueBookingSettings.venue_id) .where(VenueBookingSettings.whatsapp_phone_number_id == global_phone_number_id) .limit(1) ) if existing_owner_id is not None and existing_owner_id != candidate_venue.id: return None settings = ensure_booking_settings(candidate_venue.id, db) saved_phone_number_id = get_saved_whatsapp_phone_number_id(settings) if saved_phone_number_id and saved_phone_number_id != global_phone_number_id: return None updates_applied = False if not saved_phone_number_id: settings.whatsapp_phone_number_id = global_phone_number_id updates_applied = True if global_access_token and not get_saved_whatsapp_access_token(settings): settings.whatsapp_access_token = global_access_token updates_applied = True if updates_applied: db.flush() return candidate_venue, settings def get_booking_settings(db: Session, venue_id: int | None = None) -> tuple[Venue, VenueBookingSettings]: venue = get_default_venue(db) if venue_id is None else db.get(Venue, venue_id) if venue is None: raise ValueError("Locale non trovato") settings = ensure_booking_settings(venue.id, db) if uses_legacy_whatsapp_fallback_for_venue(venue): adopted_configuration = adopt_legacy_whatsapp_configuration(db, preferred_venue=venue) if adopted_configuration is not None: venue, settings = adopted_configuration return venue, settings def update_booking_settings(venue_id: int, payload: VenueBookingSettingsUpdate, db: Session) -> tuple[Venue, VenueBookingSettings]: venue = db.get(Venue, venue_id) if venue is None: raise ValueError("Locale non trovato") settings = ensure_booking_settings(venue.id, db) settings.turn_duration_minutes = payload.turn_duration_minutes settings.whatsapp_assistant_prompt = payload.whatsapp_assistant_prompt.strip() settings.whatsapp_business_account_id = payload.whatsapp_business_account_id.strip() settings.whatsapp_business_id = payload.whatsapp_business_id.strip() cleaned_phone_number_id = payload.whatsapp_phone_number_id.strip() if cleaned_phone_number_id: conflicting_venue_id = db.scalar( select(VenueBookingSettings.venue_id) .where( VenueBookingSettings.venue_id != venue.id, VenueBookingSettings.whatsapp_phone_number_id == cleaned_phone_number_id, ) .limit(1) ) if conflicting_venue_id is not None: raise ValueError("Questo Phone Number ID è già associato a un altro locale") settings.whatsapp_phone_number_id = cleaned_phone_number_id settings.whatsapp_display_phone_number = payload.whatsapp_display_phone_number.strip() settings.whatsapp_verified_name = payload.whatsapp_verified_name.strip() if payload.reset_whatsapp_access_token: settings.whatsapp_access_token = "" elif payload.whatsapp_access_token is not None: settings.whatsapp_access_token = payload.whatsapp_access_token.strip() db.commit() db.refresh(settings) return venue, settings def get_effective_whatsapp_access_token(settings: VenueBookingSettings, venue: Venue) -> str | None: venue_token = settings.whatsapp_access_token.strip() if venue_token: return venue_token if uses_legacy_whatsapp_fallback_for_venue(venue): global_token = (get_settings().whatsapp_access_token or "").strip() return global_token or None return None def get_effective_whatsapp_phone_number_id(settings: VenueBookingSettings, venue: Venue) -> str | None: venue_phone_number_id = settings.whatsapp_phone_number_id.strip() if venue_phone_number_id: return venue_phone_number_id if uses_legacy_whatsapp_fallback_for_venue(venue): global_phone_number_id = (get_settings().whatsapp_phone_number_id or "").strip() return global_phone_number_id or None return None def build_whatsapp_assistant_prompt(venue_name: str, custom_prompt: str = "") -> str: return compose_assistant_prompt( "whatsapp", venue_name=venue_name, locale_instructions=custom_prompt, ) def serialize_booking_settings(venue: Venue, settings: VenueBookingSettings) -> VenueBookingSettingsRead: saved_access_token = get_saved_whatsapp_access_token(settings) saved_phone_number_id = get_saved_whatsapp_phone_number_id(settings) return VenueBookingSettingsRead( venue_id=venue.id, venue_name=venue.name, turn_duration_minutes=settings.turn_duration_minutes, whatsapp_enabled=True, whatsapp_assistant_prompt=settings.whatsapp_assistant_prompt or "", whatsapp_assistant_base_prompt=DEFAULT_WHATSAPP_ASSISTANT_BASE_PROMPT, whatsapp_assistant_effective_prompt=build_whatsapp_assistant_prompt( venue.name, settings.whatsapp_assistant_prompt or "", ), whatsapp_business_account_id=settings.whatsapp_business_account_id or "", whatsapp_business_id=settings.whatsapp_business_id or "", whatsapp_phone_number_id=get_effective_whatsapp_phone_number_id(settings, venue) or "", whatsapp_display_phone_number=settings.whatsapp_display_phone_number or "", whatsapp_verified_name=settings.whatsapp_verified_name or "", whatsapp_access_token_configured=bool(get_effective_whatsapp_access_token(settings, venue)), whatsapp_phone_number_id_saved=bool(saved_phone_number_id), whatsapp_access_token_saved=bool(saved_access_token), whatsapp_uses_global_fallback=uses_global_whatsapp_fallback(settings, venue), )