buff/backend/app/models/session.py

"""
Session模型 - 用户会话管理
"""
import secrets
import string
from datetime import timezone
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

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, comment="创建时间")
    updated_at = Column(DateTime, default=beijing_now, onupdate=beijing_now, comment="更新时间")
    last_activity = Column(DateTime, default=beijing_now, onupdate=beijing_now, 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:
        """检查会话是否过期"""
        current_time = beijing_now()
        if self.expires_at.tzinfo is None:
            # 如果数据库中的时间是naive的，转换为aware进行比较
            from app.utils.timezone import datetime_to_beijing
            expires_at = datetime_to_beijing(self.expires_at.replace(tzinfo=timezone.utc))
        else:
            expires_at = self.expires_at
        return current_time > expires_at

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

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