mini-erp/mini-backend/core/session.py

import secrets
import logging
from datetime import datetime, timedelta

# ============================================================================
SESSION_TIMEOUT_MINUTES = 30

employee_sessions: dict[str, dict] = {}

active_tokens: dict[str, str] = {}

logger = logging.getLogger(__name__)

# ============================================================================
def create_employee_session(webUser: str, emp_id, first_name: str, last_name: str) -> str:
    for token, wu in list(active_tokens.items()):
        if wu == webUser:
            del active_tokens[token]
            logger.info(f"Sesión anterior removida para el usuario {webUser} (nuevo login)")
            break

    token = secrets.token_urlsafe(32)
    active_tokens[token] = webUser
    employee_sessions[webUser] = {
        "last_activity": datetime.now(),
        "emp_id": emp_id,
        "first_name": first_name,
        "last_name": last_name
    }
    logger.info(f"Sesión creada para el usuario {webUser} - {first_name} {last_name}")
    return token

# ============================================================================
def validate_and_refresh_session(token: str) -> str | None:
    webUser = active_tokens.get(token)
    if not webUser:
        return None

    session = employee_sessions.get(webUser)
    if not session:
        active_tokens.pop(token, None)
        return None

    now = datetime.now()
    if now - session["last_activity"] > timedelta(minutes=SESSION_TIMEOUT_MINUTES):
        logger.info(f"Sesión expirada para el usuario {webUser} (inactivo > {SESSION_TIMEOUT_MINUTES} min)")
        active_tokens.pop(token, None)
        employee_sessions.pop(webUser, None)
        return None

    session["last_activity"] = now
    return webUser

# ============================================================================
def remove_employee_session_by_token(token: str):
    webUser = active_tokens.pop(token, None)
    if webUser:
        employee_sessions.pop(webUser, None)
        logger.info(f"Sesión cerrada para el usuario {webUser}")

# ============================================================================
def get_employee_name(webUser: str) -> str:
    session = employee_sessions.get(webUser)
    if not session:
        return "-"
    return f"{session.get('first_name', '')} {session.get('last_name', '')}".strip()

# ============================================================================
def get_emp_id(webUser: str):
    session = employee_sessions.get(webUser)
    if not session:
        return "-"
    return session.get("emp_id", "-")
# ============================================================================