NO docker
This commit is contained in:
beabigegg
297
app/models/sys_user.py
Normal file
297
app/models/sys_user.py
Normal file
@@ -0,0 +1,297 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
系統使用者模型
|
||||
專門用於記錄帳號密碼和登入相關資訊
|
||||
|
||||
Author: PANJIT IT Team
|
||||
Created: 2025-10-01
|
||||
"""
|
||||
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Optional, Dict, Any
|
||||
from sqlalchemy import Column, Integer, String, Text, Boolean, DateTime, JSON, Enum as SQLEnum, BigInteger
|
||||
from werkzeug.security import generate_password_hash, check_password_hash
|
||||
from app import db
|
||||
from app.utils.logger import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class SysUser(db.Model):
|
||||
"""系統使用者模型 - 專門處理帳號密碼和登入記錄"""
|
||||
__tablename__ = 'sys_user'
|
||||
|
||||
id = Column(BigInteger, primary_key=True)
|
||||
|
||||
# 帳號資訊
|
||||
username = Column(String(255), nullable=False, unique=True, comment='登入帳號')
|
||||
password_hash = Column(String(512), comment='密碼雜湊 (如果需要本地儲存)')
|
||||
email = Column(String(255), nullable=False, unique=True, comment='電子郵件')
|
||||
display_name = Column(String(255), comment='顯示名稱')
|
||||
|
||||
# API 認證資訊
|
||||
api_user_id = Column(String(255), comment='API 回傳的使用者 ID')
|
||||
api_access_token = Column(Text, comment='API 回傳的 access_token')
|
||||
api_token_expires_at = Column(DateTime, comment='API Token 過期時間')
|
||||
|
||||
# 登入相關
|
||||
auth_method = Column(SQLEnum('API', 'LDAP', name='sys_user_auth_method'),
|
||||
default='API', comment='認證方式')
|
||||
last_login_at = Column(DateTime, comment='最後登入時間')
|
||||
last_login_ip = Column(String(45), comment='最後登入 IP')
|
||||
login_count = Column(Integer, default=0, comment='登入次數')
|
||||
login_success_count = Column(Integer, default=0, comment='成功登入次數')
|
||||
login_fail_count = Column(Integer, default=0, comment='失敗登入次數')
|
||||
|
||||
# 帳號狀態
|
||||
is_active = Column(Boolean, default=True, comment='是否啟用')
|
||||
is_locked = Column(Boolean, default=False, comment='是否鎖定')
|
||||
locked_until = Column(DateTime, comment='鎖定至何時')
|
||||
|
||||
# 審計欄位
|
||||
created_at = Column(DateTime, default=datetime.utcnow, comment='建立時間')
|
||||
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, comment='更新時間')
|
||||
|
||||
def __repr__(self):
|
||||
return f'<SysUser {self.username}>'
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
"""轉換為字典格式"""
|
||||
return {
|
||||
'id': self.id,
|
||||
'username': self.username,
|
||||
'email': self.email,
|
||||
'display_name': self.display_name,
|
||||
'api_user_id': self.api_user_id,
|
||||
'auth_method': self.auth_method,
|
||||
'last_login_at': self.last_login_at.isoformat() if self.last_login_at else None,
|
||||
'login_count': self.login_count,
|
||||
'login_success_count': self.login_success_count,
|
||||
'login_fail_count': self.login_fail_count,
|
||||
'is_active': self.is_active,
|
||||
'is_locked': self.is_locked,
|
||||
'api_token_expires_at': self.api_token_expires_at.isoformat() if self.api_token_expires_at else None,
|
||||
'created_at': self.created_at.isoformat() if self.created_at else None
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def get_or_create(cls, email: str, **kwargs) -> 'SysUser':
|
||||
"""
|
||||
取得或建立系統使用者 (方案A: 使用 email 作為主要識別鍵)
|
||||
|
||||
Args:
|
||||
email: 電子郵件 (主要識別鍵)
|
||||
**kwargs: 其他欄位
|
||||
|
||||
Returns:
|
||||
SysUser: 系統使用者實例
|
||||
"""
|
||||
try:
|
||||
# 使用 email 作為主要識別 (專門用於登入記錄)
|
||||
sys_user = cls.query.filter_by(email=email).first()
|
||||
|
||||
if sys_user:
|
||||
# 更新現有記錄
|
||||
sys_user.username = kwargs.get('username', sys_user.username) # API name (姓名+email)
|
||||
sys_user.display_name = kwargs.get('display_name', sys_user.display_name) # API name (姓名+email)
|
||||
sys_user.api_user_id = kwargs.get('api_user_id', sys_user.api_user_id) # Azure Object ID
|
||||
sys_user.api_access_token = kwargs.get('api_access_token', sys_user.api_access_token)
|
||||
sys_user.api_token_expires_at = kwargs.get('api_token_expires_at', sys_user.api_token_expires_at)
|
||||
sys_user.auth_method = kwargs.get('auth_method', sys_user.auth_method)
|
||||
sys_user.updated_at = datetime.utcnow()
|
||||
|
||||
logger.info(f"更新現有系統使用者: {email}")
|
||||
else:
|
||||
# 建立新記錄
|
||||
sys_user = cls(
|
||||
username=kwargs.get('username', ''), # API name (姓名+email 格式)
|
||||
email=email, # 純 email，主要識別鍵
|
||||
display_name=kwargs.get('display_name', ''), # API name (姓名+email 格式)
|
||||
api_user_id=kwargs.get('api_user_id'), # Azure Object ID
|
||||
api_access_token=kwargs.get('api_access_token'),
|
||||
api_token_expires_at=kwargs.get('api_token_expires_at'),
|
||||
auth_method=kwargs.get('auth_method', 'API'),
|
||||
login_count=0,
|
||||
login_success_count=0,
|
||||
login_fail_count=0
|
||||
)
|
||||
db.session.add(sys_user)
|
||||
logger.info(f"建立新系統使用者: {email}")
|
||||
|
||||
db.session.commit()
|
||||
return sys_user
|
||||
|
||||
except Exception as e:
|
||||
db.session.rollback()
|
||||
logger.error(f"取得或建立系統使用者失敗: {str(e)}")
|
||||
raise
|
||||
|
||||
@classmethod
|
||||
def get_by_email(cls, email: str) -> Optional['SysUser']:
|
||||
"""根據 email 查找系統使用者"""
|
||||
return cls.query.filter_by(email=email).first()
|
||||
|
||||
def record_login_attempt(self, success: bool, ip_address: str = None, auth_method: str = None):
|
||||
"""
|
||||
記錄登入嘗試
|
||||
|
||||
Args:
|
||||
success: 是否成功
|
||||
ip_address: IP 地址
|
||||
auth_method: 認證方式
|
||||
"""
|
||||
try:
|
||||
self.login_count = (self.login_count or 0) + 1
|
||||
|
||||
if success:
|
||||
self.login_success_count = (self.login_success_count or 0) + 1
|
||||
self.last_login_at = datetime.utcnow()
|
||||
self.last_login_ip = ip_address
|
||||
if auth_method:
|
||||
self.auth_method = auth_method
|
||||
|
||||
# 成功登入時解除鎖定
|
||||
if self.is_locked:
|
||||
self.is_locked = False
|
||||
self.locked_until = None
|
||||
|
||||
else:
|
||||
self.login_fail_count = (self.login_fail_count or 0) + 1
|
||||
|
||||
# 檢查是否需要鎖定帳號 (連續失敗5次)
|
||||
if self.login_fail_count >= 5:
|
||||
self.is_locked = True
|
||||
self.locked_until = datetime.utcnow() + timedelta(minutes=30) # 鎖定30分鐘
|
||||
|
||||
self.updated_at = datetime.utcnow()
|
||||
db.session.commit()
|
||||
|
||||
except Exception as e:
|
||||
db.session.rollback()
|
||||
logger.error(f"記錄登入嘗試失敗: {str(e)}")
|
||||
|
||||
def is_account_locked(self) -> bool:
|
||||
"""檢查帳號是否被鎖定"""
|
||||
if not self.is_locked:
|
||||
return False
|
||||
|
||||
# 檢查鎖定時間是否已過
|
||||
if self.locked_until and datetime.utcnow() > self.locked_until:
|
||||
self.is_locked = False
|
||||
self.locked_until = None
|
||||
db.session.commit()
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def set_password(self, password: str):
|
||||
"""設置密碼雜湊 (如果需要本地儲存密碼)"""
|
||||
self.password_hash = generate_password_hash(password)
|
||||
|
||||
def check_password(self, password: str) -> bool:
|
||||
"""檢查密碼 (如果有本地儲存密碼)"""
|
||||
if not self.password_hash:
|
||||
return False
|
||||
return check_password_hash(self.password_hash, password)
|
||||
|
||||
def update_api_token(self, access_token: str, expires_at: datetime = None):
|
||||
"""更新 API Token"""
|
||||
self.api_access_token = access_token
|
||||
self.api_token_expires_at = expires_at
|
||||
self.updated_at = datetime.utcnow()
|
||||
db.session.commit()
|
||||
|
||||
def is_api_token_valid(self) -> bool:
|
||||
"""檢查 API Token 是否有效"""
|
||||
if not self.api_access_token or not self.api_token_expires_at:
|
||||
return False
|
||||
return datetime.utcnow() < self.api_token_expires_at
|
||||
|
||||
|
||||
class LoginLog(db.Model):
|
||||
"""登入記錄模型"""
|
||||
__tablename__ = 'login_logs'
|
||||
|
||||
id = Column(BigInteger, primary_key=True)
|
||||
|
||||
# 基本資訊
|
||||
username = Column(String(255), nullable=False, comment='登入帳號')
|
||||
auth_method = Column(SQLEnum('API', 'LDAP', name='login_log_auth_method'),
|
||||
nullable=False, comment='認證方式')
|
||||
|
||||
# 登入結果
|
||||
login_success = Column(Boolean, nullable=False, comment='是否成功')
|
||||
error_message = Column(Text, comment='錯誤訊息（失敗時）')
|
||||
|
||||
# 環境資訊
|
||||
ip_address = Column(String(45), comment='IP 地址')
|
||||
user_agent = Column(Text, comment='瀏覽器資訊')
|
||||
|
||||
# API 回應 (可選，用於除錯)
|
||||
api_response_summary = Column(JSON, comment='API 回應摘要')
|
||||
|
||||
# 時間
|
||||
login_at = Column(DateTime, default=datetime.utcnow, comment='登入時間')
|
||||
|
||||
def __repr__(self):
|
||||
return f'<LoginLog {self.username}:{self.auth_method}:{self.login_success}>'
|
||||
|
||||
@classmethod
|
||||
def create_log(cls, username: str, auth_method: str, login_success: bool,
|
||||
error_message: str = None, ip_address: str = None,
|
||||
user_agent: str = None, api_response_summary: Dict = None) -> 'LoginLog':
|
||||
"""
|
||||
建立登入記錄
|
||||
|
||||
Args:
|
||||
username: 使用者帳號
|
||||
auth_method: 認證方式
|
||||
login_success: 是否成功
|
||||
error_message: 錯誤訊息
|
||||
ip_address: IP 地址
|
||||
user_agent: 瀏覽器資訊
|
||||
api_response_summary: API 回應摘要
|
||||
|
||||
Returns:
|
||||
LoginLog: 登入記錄
|
||||
"""
|
||||
try:
|
||||
log = cls(
|
||||
username=username,
|
||||
auth_method=auth_method,
|
||||
login_success=login_success,
|
||||
error_message=error_message,
|
||||
ip_address=ip_address,
|
||||
user_agent=user_agent,
|
||||
api_response_summary=api_response_summary
|
||||
)
|
||||
|
||||
db.session.add(log)
|
||||
db.session.commit()
|
||||
return log
|
||||
|
||||
except Exception as e:
|
||||
db.session.rollback()
|
||||
logger.error(f"建立登入記錄失敗: {str(e)}")
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def get_recent_failed_attempts(cls, username: str, minutes: int = 15) -> int:
|
||||
"""
|
||||
取得最近失敗的登入嘗試次數
|
||||
|
||||
Args:
|
||||
username: 使用者帳號
|
||||
minutes: 時間範圍（分鐘）
|
||||
|
||||
Returns:
|
||||
int: 失敗次數
|
||||
"""
|
||||
since = datetime.utcnow() - timedelta(minutes=minutes)
|
||||
return cls.query.filter(
|
||||
cls.username == username,
|
||||
cls.login_success == False,
|
||||
cls.login_at >= since
|
||||
).count()
|
||||
Reference in New Issue
Block a user