""" 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, Cookie import json # 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, LobbyModel from shared.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) async def get_session( request: Request, response: Response, session_id: str | None = Cookie(default=None), ): if session_id is None: # Create new session session = self.session_manager.create_session() session_id = session.id logger.info(f"Created new session: {session.getName()}") response.set_cookie(key="session_id", value=session_id) else: # Validate that session_id is a hex string of length 32 if not self._is_valid_session_id(session_id): from fastapi import Response as FastAPIResponse return FastAPIResponse( content=json.dumps({"error": "Invalid session_id"}), status_code=400, media_type="application/json", ) logger.info(f"[{session_id[:8]}]: Browser hand-shake achieved.") session = self.session_manager.get_session(session_id) if not session: # Create new session with the provided ID session = self.session_manager.create_session(session_id=session_id) logger.info(f"{session.getName()}: New session created.") else: session.update_last_used() # Update activity on session resumption logger.info(f"{session.getName()}: Existing session resumed.") # Note: Original implementation parts all lobbies for this session that have no active websocket # This would require implementing the part() method and websocket management # For now, we'll just log the session resumption # Return session response with lobby information return SessionResponse( id=session_id, name=session.name if session.name else "", lobbies=[ LobbyModel(id=lobby.id, name=lobby.name, private=lobby.private) for lobby in session.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, )