747c5de88f
- 数据库用户模型新增登录失败限制相关字段(login_failed_count, last_failed_login, login_locked_until) - 增加用户方法判断是否处于登录锁定状态及计算锁定持续时间 - 登录接口新增登录锁定检测，密码错误时增加失败计数及设置锁定时间 - 登录成功时重置失败计数和锁定状态 - 新增数据库迁移模块，自动检测并执行数据库结构更新，确保新增字段同步 - 前端登录页添加账户锁定状态显示及倒计时功能 - 前端请求中捕获423锁定状态，展示对应锁定提示，禁用登录按钮 - 优化本地状态管理，支持登录锁定错误抛出与异常处理机制
111 lines
4.1 KiB
Python
111 lines
4.1 KiB
Python
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)
|
||
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,
|
||
"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,
|
||
}
|