"""Hotel service for managing hotel configuration.""" import secrets from datetime import UTC, datetime from typing import Any import bcrypt from sqlalchemy import and_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import joinedload from .db import Hotel, WebhookEndpoint from .logging_config import get_logger _LOGGER = get_logger(__name__) def hash_password(password: str) -> str: """Hash password using bcrypt. Args: password: Plain text password Returns: Bcrypt hashed password """ salt = bcrypt.gensalt(rounds=12) return bcrypt.hashpw(password.encode('utf-8'), salt).decode('utf-8') def verify_password(password: str, password_hash: str) -> bool: """Verify password against bcrypt hash. Args: password: Plain text password password_hash: Bcrypt hash to verify against Returns: True if password matches, False otherwise """ return bcrypt.checkpw( password.encode('utf-8'), password_hash.encode('utf-8') ) def generate_webhook_secret() -> str: """Generate cryptographically secure webhook secret. Returns: 64-character URL-safe random string """ return secrets.token_urlsafe(48) # 48 bytes = 64 URL-safe chars async def sync_config_to_database( db_session: AsyncSession, config: dict[str, Any] ) -> dict[str, int]: """Sync alpine_bits_auth from config.yaml to database. Creates/updates hotels and generates webhook_endpoints if missing. Idempotent - safe to run on every startup. Args: db_session: Database session config: Application configuration dict Returns: Statistics dict with counts of created/updated records """ stats = {"hotels_created": 0, "hotels_updated": 0, "endpoints_created": 0} alpine_bits_auth = config.get("alpine_bits_auth", []) if not alpine_bits_auth: _LOGGER.info("No hotels found in alpine_bits_auth config") return stats for hotel_config in alpine_bits_auth: hotel_id = hotel_config.get("hotel_id") if not hotel_id: _LOGGER.warning("Skipping hotel config without hotel_id: %s", hotel_config) continue # Check if hotel exists result = await db_session.execute( select(Hotel).where(Hotel.hotel_id == hotel_id) ) hotel = result.scalar_one_or_none() if not hotel: # Create new hotel password_hash = hash_password(hotel_config["password"]) hotel = Hotel( hotel_id=hotel_id, hotel_name=hotel_config.get("hotel_name", hotel_id), username=hotel_config["username"], password_hash=password_hash, meta_account_id=hotel_config.get("meta_account"), google_account_id=hotel_config.get("google_account"), push_endpoint_url=hotel_config.get("push_endpoint", {}).get("url"), push_endpoint_token=hotel_config.get("push_endpoint", {}).get("token"), push_endpoint_username=hotel_config.get("push_endpoint", {}).get("username"), created_at=datetime.now(UTC), updated_at=datetime.now(UTC), is_active=True, ) db_session.add(hotel) await db_session.flush() stats["hotels_created"] += 1 _LOGGER.info("Created hotel: %s", hotel_id) else: # Update existing hotel (config may have changed) # Note: We do NOT update password_hash for security reasons hotel.hotel_name = hotel_config.get("hotel_name", hotel_id) hotel.meta_account_id = hotel_config.get("meta_account") hotel.google_account_id = hotel_config.get("google_account") push_endpoint = hotel_config.get("push_endpoint", {}) hotel.push_endpoint_url = push_endpoint.get("url") hotel.push_endpoint_token = push_endpoint.get("token") hotel.push_endpoint_username = push_endpoint.get("username") hotel.updated_at = datetime.now(UTC) stats["hotels_updated"] += 1 _LOGGER.debug("Updated hotel: %s", hotel_id) # Ensure hotel has at least default webhook endpoints result = await db_session.execute( select(WebhookEndpoint).where(WebhookEndpoint.hotel_id == hotel_id) ) existing_endpoints = result.scalars().all() if not existing_endpoints: # Create default webhook endpoints for backward compatibility for webhook_type in ["wix_form", "generic"]: webhook_secret = generate_webhook_secret() endpoint = WebhookEndpoint( hotel_id=hotel_id, webhook_secret=webhook_secret, webhook_type=webhook_type, description=f"Auto-generated {webhook_type} endpoint", is_enabled=True, created_at=datetime.now(UTC), ) db_session.add(endpoint) stats["endpoints_created"] += 1 _LOGGER.info( "Created webhook endpoint for hotel %s, type=%s, secret=%s", hotel_id, webhook_type, webhook_secret ) await db_session.commit() _LOGGER.info( "Config sync complete: %d hotels created, %d updated, %d endpoints created", stats["hotels_created"], stats["hotels_updated"], stats["endpoints_created"] ) return stats class HotelService: """Service for hotel configuration access. Always reads from database (synced from config at startup). """ def __init__(self, db_session: AsyncSession): """Initialize HotelService. Args: db_session: Database session """ self.db_session = db_session async def get_hotel_by_id(self, hotel_id: str) -> Hotel | None: """Get hotel by hotel_id. Args: hotel_id: Hotel identifier Returns: Hotel instance or None if not found """ result = await self.db_session.execute( select(Hotel) .where( and_( Hotel.hotel_id == hotel_id, Hotel.is_active == True ) ) ) return result.scalar_one_or_none() async def get_hotel_by_webhook_secret( self, webhook_secret: str ) -> tuple[Hotel, WebhookEndpoint] | tuple[None, None]: """Get hotel and webhook_endpoint by webhook_secret. Args: webhook_secret: Webhook secret string Returns: Tuple of (Hotel, WebhookEndpoint) or (None, None) if not found """ result = await self.db_session.execute( select(WebhookEndpoint) .where( and_( WebhookEndpoint.webhook_secret == webhook_secret, WebhookEndpoint.is_enabled == True ) ) .options(joinedload(WebhookEndpoint.hotel)) ) endpoint = result.scalar_one_or_none() if endpoint and endpoint.hotel.is_active: return endpoint.hotel, endpoint return None, None async def get_hotel_by_username(self, username: str) -> Hotel | None: """Get hotel by AlpineBits username. Args: username: AlpineBits username Returns: Hotel instance or None if not found """ result = await self.db_session.execute( select(Hotel) .where( and_( Hotel.username == username, Hotel.is_active == True ) ) ) return result.scalar_one_or_none() async def authenticate_hotel(self, username: str, password: str) -> Hotel | None: """Authenticate a hotel using username and password. Args: username: AlpineBits username password: Plain text password submitted via HTTP basic auth Returns: Hotel instance if the credentials are valid and the hotel is active, otherwise None. """ hotel = await self.get_hotel_by_username(username) if not hotel: return None if not password: return None if verify_password(password, hotel.password_hash): return hotel return None