buff2.0/backend-admin/app/services/session_service.py

"""
Session服务 - 用户会话管理
"""
from datetime import datetime, timedelta, timezone

BEIJING_TZ = timezone(timedelta(hours=8))
from typing import Optional, Tuple
from sqlalchemy.orm import Session as DBSession
from sqlalchemy import and_, or_
from app.database import get_db
from app.models.session import Session
from app.models.user import User
from app.core.logger import logger
from app.utils.timezone import beijing_now, beijing_now_naive

# 会话配置（硬编码）
SESSION_TIMEOUT_HOURS = 24           # 会话超时时间（小时）
PREVENT_DUPLICATE_LOGIN = True       # 是否防止重复登录
MAX_SESSIONS_PER_USER = 1            # 每个用户最大会话数
SESSION_EXTEND_ENABLED = True        # 是否启用心跳延长会话
SESSION_EXTEND_THRESHOLD_HOURS = 4   # 会话延长阈值（小时）
SESSION_EXTEND_HOURS = 2             # 每次延长的小时数

class SessionService:
    """Session服务类"""

    @staticmethod
    def get_session_config(db: DBSession = None) -> dict:
        """获取Session相关配置"""
        return {
            "session_timeout_hours": SESSION_TIMEOUT_HOURS,
            "prevent_duplicate_login": PREVENT_DUPLICATE_LOGIN,
            "max_sessions_per_user": MAX_SESSIONS_PER_USER,
            "session_extend_enabled": SESSION_EXTEND_ENABLED,
            "session_extend_threshold_hours": SESSION_EXTEND_THRESHOLD_HOURS,
            "session_extend_hours": SESSION_EXTEND_HOURS,
        }

    @staticmethod
    def create_session(
        db: DBSession,
        user_id: int,
        machine_code: Optional[str] = None,
        ip_address: Optional[str] = None,
        user_agent: Optional[str] = None
    ) -> Tuple[bool, str, Optional[str]]:
        """
        创建新的用户会话

        Returns:
            tuple: (是否成功, 消息, session_id)
        """
        try:
            config = SessionService.get_session_config(db)

            # 检查用户是否存在且状态正常
            user = db.query(User).filter(
                and_(
                    User.id == user_id,
                    User.status == 1,
                    User.is_deleted == 0
                )
            ).first()

            if not user:
                return False, "用户不存在或已被禁用", None

            # 如果启用单点登录，先清理该用户的所有其他会话
            if config["prevent_duplicate_login"]:
                SessionService.deactivate_user_sessions(db, user_id)

            # 生成新的会话ID
            session_id = Session.generate_session_id()

            # 计算过期时间
            expires_at = beijing_now_naive() + timedelta(hours=config["session_timeout_hours"])

            # 创建新会话
            new_session = Session(
                session_id=session_id,
                user_id=user_id,
                machine_code=machine_code,
                ip_address=ip_address,
                user_agent=user_agent,
                is_active=1,
                expires_at=expires_at,
                created_at=beijing_now_naive(),
                updated_at=beijing_now_naive(),
                last_activity=beijing_now_naive()
            )

            db.add(new_session)
            db.commit()
            db.refresh(new_session)

            logger.info(f"用户 {user_id} 创建新会话: {session_id}")
            return True, "会话创建成功", session_id

        except Exception as e:
            db.rollback()
            error_msg = f"创建会话失败: {str(e)}"
            logger.exception("创建会话失败")
            return False, error_msg, None

    @staticmethod
    def validate_session(db: DBSession, session_id: str, update_activity: bool = True, allow_expired_extend: bool = False) -> Tuple[bool, Optional[Session], str]:
        """
        验证会话是否有效

        Args:
            allow_expired_extend: 是否允许延长已过期的活跃Session（用于心跳场景）

        Returns:
            tuple: (是否有效, Session对象, 错误消息)
        """
        try:
            session = db.query(Session).filter(
                and_(
                    Session.session_id == session_id,
                    Session.is_active == 1
                )
            ).first()

            if not session:
                return False, None, "会话不存在"

            # 检查用户是否还存在且状态正常
            user = db.query(User).filter(
                and_(
                    User.id == session.user_id,
                    User.status == 1,
                    User.is_deleted == 0
                )
            ).first()

            if not user:
                # 用户不存在或被禁用，停用会话
                session.deactivate()
                db.commit()
                return False, None, "用户不存在或已被禁用"

            # 检查用户是否已过期
            if user.expire_time:
                # 确保比较的是同一时区的datetime
                current_time = beijing_now()
                if user.expire_time.tzinfo is None:
                    # 数据库中为 naive 北京时间，补时区后与 beijing_now() 比较
                    expire_time = user.expire_time.replace(tzinfo=BEIJING_TZ)
                else:
                    expire_time = user.expire_time

                if expire_time < current_time:
                    session.deactivate()
                    db.commit()
                    return False, None, "用户卡密已过期"

            # 检查单点登录：如果启用了单点登录，验证当前Session是否是唯一的活跃Session
            config = SessionService.get_session_config(db)
            if config.get("prevent_duplicate_login", True):
                # 获取用户的所有活跃Session
                active_sessions = SessionService.get_user_active_sessions(db, user.id)
                active_session_ids = [s.session_id for s in active_sessions]

                # 如果当前Session不在活跃列表中，说明被其他地方登录踢掉了
                if session.session_id not in active_session_ids:
                    logger.warning(f"用户 {user.username} 的Session {session_id[:8]}... 已被其他地方登录踢掉")
                    return False, None, "账号在其他地方登录"

            # 检查是否过期
            is_expired = session.is_expired()

            if is_expired:
                # 如果允许延长已过期Session（心跳场景），尝试延长
                if allow_expired_extend and SessionService.should_extend_session(db, session):
                    logger.info(f"Session {session_id[:8]}... 已过期但允许延长，执行延长操作")
                    SessionService.extend_session_on_heartbeat(db, session)
                    # 重新检查是否仍然过期
                    is_expired = session.is_expired()

                # 如果仍然过期，则停用会话
                if is_expired:
                    session.deactivate()
                    db.commit()
                    return False, None, "会话已过期"

            # 更新活动时间
            if update_activity:
                session.update_activity()

                # 检查是否需要基于心跳延长session（正常情况下的延长）
                if not allow_expired_extend:  # 避免重复延长
                    SessionService.extend_session_on_heartbeat(db, session)

                db.commit()

            return True, session, ""

        except Exception as e:
            logger.exception("验证会话失败")
            return False, None, f"验证会话异常: {str(e)}"

    @staticmethod
    def deactivate_session(db: DBSession, session_id: str) -> bool:
        """停用指定会话"""
        try:
            session = db.query(Session).filter(Session.session_id == session_id).first()
            if session:
                session.deactivate()
                db.commit()
                logger.info(f"会话 {session_id} 已停用")
                return True
            return False
        except Exception as e:
            db.rollback()
            logger.exception("停用会话失败")
            return False

    @staticmethod
    def deactivate_user_sessions(db: DBSession, user_id: int, exclude_session_id: Optional[str] = None) -> int:
        """
        停用指定用户的所有会话

        Args:
            exclude_session_id: 排除的会话ID（用于单点登录时保留当前会话）

        Returns:
            int: 停用的会话数量
        """
        try:
            query = db.query(Session).filter(
                and_(
                    Session.user_id == user_id,
                    Session.is_active == 1
                )
            )

            if exclude_session_id:
                query = query.filter(Session.session_id != exclude_session_id)

            sessions = query.all()
            count = len(sessions)

            for session in sessions:
                session.deactivate()

            if count > 0:
                db.commit()
                logger.info(f"用户 {user_id} 的 {count} 个会话已停用")

            return count

        except Exception as e:
            db.rollback()
            logger.exception("停用用户会话失败")
            return 0

    @staticmethod
    def cleanup_expired_sessions(db: DBSession) -> int:
        """
        清理过期的会话

        注意：为了保持用户一致在线体验，我们不清理活跃的过期会话，
        只清理长期非活跃的旧会话记录（保留30天用于审计）

        Returns:
            int: 清理的会话数量
        """
        try:
            from app.utils.history import create_session_history

            # 先写历史表，再改/删：查出即将被删除的会话，逐条写入 sessions_his
            old_sessions = beijing_now_naive() - timedelta(days=30)
            sessions_to_clean = db.query(Session).filter(
                and_(
                    Session.is_active == 0,
                    Session.updated_at < old_sessions
                )
            ).all()
            for s in sessions_to_clean:
                create_session_history(db, s, "DELETE")

            # 清理长期非活跃会话（保留最近30天的数据用于审计）
            old_count = db.query(Session).filter(
                and_(
                    Session.is_active == 0,
                    Session.updated_at < old_sessions
                )
            ).delete()

            # 注意：我们不清理活跃的过期会话，让用户保持在线
            # 只有当用户主动登出或心跳失败时，才会标记为非活跃

            if old_count > 0:
                db.commit()
                logger.info(f"清理了 {old_count} 个30天前的非活跃会话记录")

            return old_count

        except Exception as e:
            db.rollback()
            logger.error(f"清理过期会话失败: {e}")
            return 0

    @staticmethod
    def get_user_active_sessions(db: DBSession, user_id: int) -> list:
        """获取用户的所有活跃会话"""
        try:
            sessions = db.query(Session).filter(
                and_(
                    Session.user_id == user_id,
                    Session.is_active == 1,
                    Session.expires_at > beijing_now_naive()
                )
            ).order_by(Session.created_at.desc()).all()

            return sessions

        except Exception as e:
            logger.exception("获取用户活跃会话失败")
            return []

    @staticmethod
    def should_extend_session(db: DBSession, session: Session) -> bool:
        """
        检查是否应该延长session（基于心跳）

        Returns:
            bool: 是否需要延长
        """
        try:
            config = SessionService.get_session_config(db)

            # 检查是否启用心跳延长功能
            if not config.get("session_extend_enabled", True):
                return False

            # 计算距离过期还有多少小时
            current_time = beijing_now()
            if session.expires_at.tzinfo is None:
                # 数据库中为 naive 北京时间，补时区后与 beijing_now() 比较
                expires_at = session.expires_at.replace(tzinfo=BEIJING_TZ)
            else:
                expires_at = session.expires_at

            time_until_expiry = expires_at - current_time
            hours_until_expiry = time_until_expiry.total_seconds() / 3600

            # 如果距离过期时间小于阈值，则需要延长
            threshold_hours = config.get("session_extend_threshold_hours", 4)
            return hours_until_expiry <= threshold_hours

        except Exception as e:
            logger.exception("检查是否需要延长session失败")
            return False

    @staticmethod
    def extend_session_on_heartbeat(db: DBSession, session: Session) -> bool:
        """
        基于心跳延长session（如果需要）

        Args:
            session: Session对象

        Returns:
            bool: 是否执行了延长操作
        """
        try:
            # 检查是否需要延长
            if not SessionService.should_extend_session(db, session):
                return False

            config = SessionService.get_session_config(db)
            extend_hours = config.get("session_extend_hours", 2)

            # 执行延长
            session.expires_at = beijing_now_naive() + timedelta(hours=extend_hours)
            session.update_activity()
            db.commit()

            logger.info(f"心跳延长session {session.session_id[:8]}... {extend_hours}小时，新过期时间: {session.expires_at}")
            return True

        except Exception as e:
            db.rollback()
            logger.exception("心跳延长session失败")
            return False

    @staticmethod
    def extend_session(db: DBSession, session_id: str, hours: int = 24) -> bool:
        """
        延长会话过期时间

        Args:
            hours: 延长的小时数

        Returns:
            bool: 是否成功
        """
        try:
            session = db.query(Session).filter(
                and_(
                    Session.session_id == session_id,
                    Session.is_active == 1
                )
            ).first()

            if not session:
                return False

            # 延长过期时间
            session.expires_at = beijing_now_naive() + timedelta(hours=hours)
            session.update_activity()
            db.commit()

            logger.info(f"会话 {session_id} 过期时间已延长 {hours} 小时")
            return True

        except Exception as e:
            db.rollback()
            logger.exception("延长会话失败")
            return False