session_manager.py•4.33 kB
"""
Módulo de gestión de sesiones de WebDriver.
"""
import os
import shutil
import uuid
import time
from datetime import datetime, timedelta
from typing import Dict, Optional
from selenium.webdriver.remote.webdriver import WebDriver
from config import ServerConfig
class SessionInfo:
"""Información de una sesión de navegador."""
def __init__(self, session_id: str, driver: WebDriver, browser_type: str, user_data_dir: Optional[str] = None):
self.session_id = session_id
self.driver = driver
self.browser_type = browser_type
self.user_data_dir = user_data_dir # Para limpieza de perfiles de Chrome
self.created_at = time.time()
self.last_activity = time.time()
def update_activity(self):
"""Actualiza el timestamp de la última actividad."""
self.last_activity = time.time()
def is_expired(self, timeout: int) -> bool:
"""Verifica si la sesión ha expirado."""
return (time.time() - self.last_activity) > timeout
class SessionManager:
"""Gestor de sesiones de navegador."""
def __init__(self, config: ServerConfig):
self.config = config
self.sessions: Dict[str, SessionInfo] = {}
def create_session(self, driver: WebDriver, browser_type: str, user_data_dir: Optional[str] = None) -> str:
"""Crea una nueva sesión y devuelve su ID."""
if len(self.sessions) >= self.config.max_sessions:
# Limpiar sesiones expiradas primero
self.cleanup_expired_sessions()
# Si aún hay demasiadas sesiones, cerrar la más antigua
if len(self.sessions) >= self.config.max_sessions:
oldest_session_id = min(
self.sessions.keys(),
key=lambda sid: self.sessions[sid].last_activity
)
self.close_session(oldest_session_id)
session_id = str(uuid.uuid4())
session_info = SessionInfo(session_id, driver, browser_type, user_data_dir)
self.sessions[session_id] = session_info
return session_id
def get_session(self, session_id: str) -> Optional[SessionInfo]:
"""Obtiene una sesión por su ID."""
session = self.sessions.get(session_id)
if session:
session.update_activity()
return session
def close_session(self, session_id: str) -> bool:
"""Cierra una sesión específica."""
session = self.sessions.get(session_id)
if session:
try:
session.driver.quit()
except Exception:
pass # Ignorar errores al cerrar el driver
# Eliminar el directorio de datos de usuario si existe
if session.user_data_dir and os.path.exists(session.user_data_dir):
try:
shutil.rmtree(session.user_data_dir)
except Exception as e:
print(f"Error al eliminar user_data_dir {session.user_data_dir}: {e}")
del self.sessions[session_id]
return True
return False
def close_all_sessions(self):
"""Cierra todas las sesiones activas."""
for session_id in list(self.sessions.keys()):
self.close_session(session_id)
def cleanup_expired_sessions(self):
"""Limpia las sesiones expiradas."""
expired_sessions = [
session_id for session_id, session in self.sessions.items()
if session.is_expired(self.config.session_timeout)
]
for session_id in expired_sessions:
self.close_session(session_id)
def get_session_count(self) -> int:
"""Devuelve el número de sesiones activas."""
return len(self.sessions)
def list_sessions(self) -> Dict[str, Dict]:
"""Lista todas las sesiones activas con información básica."""
return {
session_id: {
"browser_type": session.browser_type,
"created_at": session.created_at,
"last_activity": session.last_activity,
"age_seconds": time.time() - session.created_at
}
for session_id, session in self.sessions.items()
}