buff3.0/backend-admin/app/models/session.py

"""
Session模型 - 用户会话管理
"""
import secrets
import string
from datetime import timezone, timedelta
from sqlalchemy import Column, Integer, String, DateTime, SmallInteger, Text, ForeignKey
from sqlalchemy.orm import relationship
from app.database import Base
from app.utils.timezone import beijing_now, beijing_now_naive

BEIJING_TZ = timezone(timedelta(hours=8))

class Session(Base):
    """用户会话表"""
    __tablename__ = "sessions"

    id = Column(Integer, primary_key=True, index=True, comment="主键ID")
    session_id = Column(String(64), unique=True, index=True, nullable=False, comment="会话ID")
    user_id = Column(Integer, ForeignKey("users.id"), index=True, nullable=False, comment="用户ID")
    machine_code = Column(String(100), nullable=True, comment="机器码")
    ip_address = Column(String(45), nullable=True, comment="IP地址")
    user_agent = Column(Text, nullable=True, comment="用户代理")
    is_active = Column(SmallInteger, default=1, index=True, comment="是否激活：0-否，1-是")
    expires_at = Column(DateTime, index=True, nullable=False, comment="过期时间")
    created_at = Column(DateTime, default=beijing_now_naive, comment="创建时间")
    updated_at = Column(DateTime, default=beijing_now_naive, onupdate=beijing_now_naive, comment="更新时间")
    last_activity = Column(DateTime, default=beijing_now_naive, onupdate=beijing_now_naive, comment="最后活动时间")

    # 关系
    user = relationship("User", back_populates="sessions")

    @staticmethod
    def generate_session_id() -> str:
        """生成随机会话ID"""
        characters = string.ascii_letters + string.digits
        return ''.join(secrets.choice(characters) for _ in range(32))

    def is_expired(self) -> bool:
        """检查会话是否过期（数据库存 naive 北京时间）"""
        current_time = beijing_now()
        if self.expires_at.tzinfo is None:
            expires_at = self.expires_at.replace(tzinfo=BEIJING_TZ)
        else:
            expires_at = self.expires_at
        return current_time > expires_at

    def update_activity(self):
        """更新最后活动时间"""
        self.last_activity = beijing_now_naive()
        self.updated_at = beijing_now_naive()

    def deactivate(self):
        """停用会话"""
        self.is_active = 0
        self.updated_at = beijing_now_naive()