""" Session API endpoints for the AI Voice Bot server. This module contains session management endpoints. """ from typing import TYPE_CHECKING from fastapi import APIRouter, Request, Response # Import shared models import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))) from shared.models import SessionResponse, HealthResponse from logger import logger if TYPE_CHECKING: from ..core.session_manager import SessionManager class SessionAPI: """Session API endpoint handlers""" def __init__(self, session_manager: "SessionManager", public_url: str = "/"): self.session_manager = session_manager self.router = APIRouter(prefix=f"{public_url}api") self._register_routes() def _is_valid_session_id(self, session_id: str) -> bool: """Check if session ID has the correct format (32-character hex string)""" if not session_id or len(session_id) != 32: return False # Check if it's a valid hexadecimal string try: int(session_id, 16) return True except ValueError: return False def _register_routes(self): """Register all session routes""" @self.router.get("/health", response_model=HealthResponse) def health(): return HealthResponse(status="ok") @self.router.get("/session", response_model=SessionResponse) def get_session(request: Request, response: Response): # Check for existing session cookie session_id = request.cookies.get("session_id") if session_id and self._is_valid_session_id(session_id): # Try to get existing session existing_session = self.session_manager.get_session(session_id) if existing_session: logger.info(f"Found existing session from cookie: {session_id[:8]}") return SessionResponse( id=existing_session.id, name=existing_session.name or "", lobbies=[], # Could be populated based on existing session protected=False, has_media=existing_session.has_media, bot_run_id=existing_session.bot_run_id, bot_provider_id=existing_session.bot_provider_id, bot_instance_id=existing_session.bot_instance_id, ) else: # Cookie exists but session doesn't - create new session with this ID logger.info( f"Creating new session with cookie ID: {session_id[:8]}" ) session = self.session_manager.create_session(session_id=session_id) else: # No valid cookie - create completely new session session = self.session_manager.create_session() logger.info(f"Created new session: {session.getName()}") # Set the session cookie (expires in 30 days) response.set_cookie( key="session_id", value=session.id, max_age=30 * 24 * 60 * 60, # 30 days in seconds httponly=True, secure=False, # Set to True in production with HTTPS samesite="lax", ) return SessionResponse( id=session.id, name=session.name or "", lobbies=[], # New sessions start with no lobbies protected=False, has_media=session.has_media, bot_run_id=session.bot_run_id, bot_provider_id=session.bot_provider_id, bot_instance_id=session.bot_instance_id, )