""" Multi-user session management. This module provides session management for multiple users, where each API key can have its own GeoGuessr session. """ import asyncio import logging from typing import Optional from ..config import settings from .session import SessionManager, UserSession from .user_context import UserContext logger = logging.getLogger(__name__) class MultiUserSessionManager: """ Manages GeoGuessr sessions for multiple users. Each API key can have its own GeoGuessr session, allowing multiple users to access their own accounts through the same MCP server instance. """ def __init__(self): """Initialize the multi-user session manager.""" # Map API keys to their session managers self._user_managers: dict[str, SessionManager] = {} self._lock = asyncio.Lock() # Create default session manager if default cookie is configured if settings.DEFAULT_NCFA_COOKIE: logger.info("Default GeoGuessr cookie configured - will be used as fallback") async def get_user_context(self, api_key: str) -> UserContext: """ Get or create a user context for an API key. Args: api_key: The API key identifying the user Returns: UserContext: The context for this user """ async with self._lock: # Get or create session manager for this user if api_key not in self._user_managers: # Create new session manager with default cookie as fallback self._user_managers[api_key] = SessionManager( default_cookie=settings.DEFAULT_NCFA_COOKIE ) logger.info(f"Created new session manager for API key {api_key[:8]}...") manager = self._user_managers[api_key] # Get the session (may return default session if no user login) session = await manager.get_session() # Create user context context = UserContext(api_key=api_key, session=session) return context async def login_user( self, api_key: str, email: str, password: str ) -> tuple[str, UserSession]: """ Login a specific user (API key) to their GeoGuessr account. Args: api_key: The API key identifying the user email: GeoGuessr email password: GeoGuessr password Returns: tuple[str, UserSession]: (session_token, UserSession) Raises: ValueError: If login fails """ async with self._lock: if api_key not in self._user_managers: self._user_managers[api_key] = SessionManager() manager = self._user_managers[api_key] # Perform login session_token, session = await manager.login(email, password) logger.info( f"User {session.username} logged in successfully for API key {api_key[:8]}..." ) return session_token, session async def logout_user(self, api_key: str, session_token: str) -> bool: """ Logout a specific user's session. Args: api_key: The API key identifying the user session_token: The session token to logout Returns: bool: True if logout successful, False otherwise """ async with self._lock: if api_key not in self._user_managers: return False manager = self._user_managers[api_key] success = await manager.logout(session_token) if success: logger.info(f"User logged out for API key {api_key[:8]}...") return success async def set_user_cookie(self, api_key: str, cookie: str) -> bool: """ Set a GeoGuessr cookie for a specific user. Args: api_key: The API key identifying the user cookie: The NCFA cookie value Returns: bool: True if cookie is valid, False otherwise """ # Validate cookie first profile = await SessionManager.validate_cookie(cookie) if not profile: return False async with self._lock: if api_key not in self._user_managers: self._user_managers[api_key] = SessionManager() manager = self._user_managers[api_key] await manager.set_default_cookie(cookie) logger.info( f"Cookie set for user {profile.get('nick', 'unknown')} (API key {api_key[:8]}...)" ) return True async def get_session_for_api_key(self, api_key: str) -> Optional[UserSession]: """ Get the active session for a specific API key. Args: api_key: The API key identifying the user Returns: UserSession if available, None otherwise """ async with self._lock: if api_key not in self._user_managers: return None manager = self._user_managers[api_key] return await manager.get_session() async def get_auth_status(self, api_key: str) -> dict: """ Get authentication status for a specific user. Args: api_key: The API key identifying the user Returns: dict: Authentication status information """ context = await self.get_user_context(api_key) return { "authenticated": context.is_authenticated, "user_id": context.user_id if context.is_authenticated else None, "username": context.username if context.is_authenticated else None, "api_key": f"{api_key[:8]}..." if len(api_key) > 8 else "***", } # Global multi-user session manager instance multi_user_session_manager = MultiUserSessionManager()