from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, Text from sqlalchemy.orm import relationship from sqlalchemy.sql import func from server.models import Base class User(Base): """用户模型""" __tablename__ = "users" id = Column(Integer, primary_key=True, autoincrement=True) username = Column(String, nullable=False, unique=True, index=True) # 显示名称,2-20字符 user_id = Column(String, nullable=False, unique=True, index=True) # 登录ID,根据用户名生成 phone_number = Column(String, nullable=True, unique=True, index=True) # 手机号,可选登录方式 avatar = Column(String, nullable=True) # 头像URL password_hash = Column(String, nullable=False) role = Column(String, nullable=False, default="user") # 角色: superadmin, admin, user created_at = Column(DateTime, default=func.now()) last_login = Column(DateTime, nullable=True) # 登录失败限制相关字段 login_failed_count = Column(Integer, nullable=False, default=0) # 登录失败次数 last_failed_login = Column(DateTime, nullable=True) # 最后一次登录失败时间 login_locked_until = Column(DateTime, nullable=True) # 锁定到什么时候 # 关联操作日志 operation_logs = relationship("OperationLog", back_populates="user") def to_dict(self, include_password=False): result = { "id": self.id, "username": self.username, "user_id": self.user_id, "phone_number": self.phone_number, "avatar": self.avatar, "role": self.role, "created_at": self.created_at.isoformat() if self.created_at else None, "last_login": self.last_login.isoformat() if self.last_login else None, "login_failed_count": self.login_failed_count, "last_failed_login": self.last_failed_login.isoformat() if self.last_failed_login else None, "login_locked_until": self.login_locked_until.isoformat() if self.login_locked_until else None, } if include_password: result["password_hash"] = self.password_hash return result def is_login_locked(self): """检查用户是否处于登录锁定状态""" if self.login_locked_until is None: return False from datetime import datetime return datetime.now() < self.login_locked_until def get_remaining_lock_time(self): """获取剩余锁定时间(秒)""" if not self.is_login_locked(): return 0 from datetime import datetime return int((self.login_locked_until - datetime.now()).total_seconds()) def calculate_lock_duration(self): """根据失败次数计算锁定时长(秒)""" if self.login_failed_count < 10: return 0 # 从第10次失败开始,等待时间从1秒开始,每次翻倍 wait_seconds = 2 ** (self.login_failed_count - 10) # 最大锁定时间:365天 max_seconds = 365 * 24 * 60 * 60 return min(wait_seconds, max_seconds) def increment_failed_login(self): """增加登录失败次数并设置锁定时间""" from datetime import datetime, timedelta self.login_failed_count += 1 self.last_failed_login = datetime.now() lock_duration = self.calculate_lock_duration() if lock_duration > 0: self.login_locked_until = datetime.now() + timedelta(seconds=lock_duration) def reset_failed_login(self): """重置登录失败相关字段""" self.login_failed_count = 0 self.last_failed_login = None self.login_locked_until = None class OperationLog(Base): """操作日志模型""" __tablename__ = "operation_logs" id = Column(Integer, primary_key=True, autoincrement=True) user_id = Column(Integer, ForeignKey("users.id"), nullable=False) operation = Column(String, nullable=False) details = Column(Text, nullable=True) ip_address = Column(String, nullable=True) timestamp = Column(DateTime, default=func.now()) # 关联用户 user = relationship("User", back_populates="operation_logs") def to_dict(self): return { "id": self.id, "user_id": self.user_id, "operation": self.operation, "details": self.details, "ip_address": self.ip_address, "timestamp": self.timestamp.isoformat() if self.timestamp else None, }