265 lines
7.9 KiB
Python
265 lines
7.9 KiB
Python
"""
|
||
用户相关模型
|
||
"""
|
||
from datetime import datetime
|
||
from typing import List, Optional, TYPE_CHECKING
|
||
|
||
from sqlalchemy import (
|
||
Boolean,
|
||
Column,
|
||
DateTime,
|
||
Integer,
|
||
String,
|
||
Table,
|
||
Text,
|
||
ForeignKey,
|
||
)
|
||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||
|
||
from app.models.base import BaseModel
|
||
|
||
# 避免循环导入
|
||
if TYPE_CHECKING:
|
||
from .role import Role
|
||
|
||
|
||
# 用户角色关联表（多对多）
|
||
user_roles_table = Table(
|
||
"sys_user_role",
|
||
BaseModel.metadata,
|
||
Column("user_id", Integer, ForeignKey("sys_user.id"), primary_key=True),
|
||
Column("role_id", Integer, ForeignKey("sys_role.id"), primary_key=True),
|
||
)
|
||
|
||
|
||
class User(BaseModel):
|
||
"""
|
||
用户表
|
||
"""
|
||
|
||
__tablename__ = "sys_user"
|
||
|
||
username: Mapped[str] = mapped_column(
|
||
String(50), unique=True, index=True, nullable=False, comment="用户名"
|
||
)
|
||
password: Mapped[str] = mapped_column(String(255), nullable=False, comment="密码哈希")
|
||
nickname: Mapped[Optional[str]] = mapped_column(
|
||
String(50), nullable=True, comment="昵称"
|
||
)
|
||
email: Mapped[Optional[str]] = mapped_column(
|
||
String(100), unique=True, index=True, nullable=True, comment="邮箱"
|
||
)
|
||
phone: Mapped[Optional[str]] = mapped_column(
|
||
String(20), nullable=True, comment="手机号"
|
||
)
|
||
avatar: Mapped[Optional[str]] = mapped_column(
|
||
String(255), nullable=True, comment="头像URL"
|
||
)
|
||
status: Mapped[bool] = mapped_column(
|
||
Boolean, default=True, comment="状态：0-禁用，1-正常"
|
||
)
|
||
is_superuser: Mapped[bool] = mapped_column(
|
||
Boolean, default=False, comment="是否超级用户"
|
||
)
|
||
last_login_time: Mapped[Optional[datetime]] = mapped_column(
|
||
DateTime(timezone=True), nullable=True, comment="最后登录时间"
|
||
)
|
||
last_login_ip: Mapped[Optional[str]] = mapped_column(
|
||
String(50), nullable=True, comment="最后登录IP"
|
||
)
|
||
# 关联的企业实体ID（MCN机构ID或主播ID）
|
||
entity_id: Mapped[Optional[str]] = mapped_column(
|
||
String(50), nullable=True, comment="关联的企业实体ID（MCN机构或主播）"
|
||
)
|
||
# 实体类型：mcn-机构，streamer-主播
|
||
entity_type: Mapped[Optional[str]] = mapped_column(
|
||
String(20), nullable=True, comment="关联实体类型：mcn-机构，streamer-主播"
|
||
)
|
||
|
||
# 关系
|
||
roles: Mapped[List["Role"]] = relationship(
|
||
"Role", secondary=user_roles_table, back_populates="users"
|
||
)
|
||
|
||
async def check_password(self, plain_password: str) -> bool:
|
||
"""
|
||
检查密码
|
||
"""
|
||
from app.utils.helpers import verify_password
|
||
|
||
return verify_password(plain_password, self.password)
|
||
|
||
async def set_password(self, plain_password: str) -> None:
|
||
"""
|
||
设置密码
|
||
"""
|
||
from app.utils.helpers import get_password_hash
|
||
|
||
self.password = get_password_hash(plain_password)
|
||
|
||
@classmethod
|
||
async def get_by_username(
|
||
cls, session, username: str
|
||
) -> Optional["User"]:
|
||
"""
|
||
根据用户名获取用户
|
||
"""
|
||
from sqlalchemy import select
|
||
|
||
result = await session.execute(select(cls).where(cls.username == username))
|
||
return result.scalar_one_or_none()
|
||
|
||
@classmethod
|
||
async def get_by_email(cls, session, email: str) -> Optional["User"]:
|
||
"""
|
||
根据邮箱获取用户
|
||
"""
|
||
from sqlalchemy import select
|
||
|
||
result = await session.execute(select(cls).where(cls.email == email))
|
||
return result.scalar_one_or_none()
|
||
|
||
async def has_role(self, role_name: str) -> bool:
|
||
"""
|
||
检查是否拥有特定角色
|
||
"""
|
||
return any(role.role_code == role_name for role in self.roles)
|
||
|
||
async def has_permission(self, permission: str) -> bool:
|
||
"""
|
||
检查是否拥有特定权限
|
||
"""
|
||
for role in self.roles:
|
||
for perm in role.permissions:
|
||
if perm.permission_code == permission:
|
||
return True
|
||
return False
|
||
|
||
|
||
class Role(BaseModel):
|
||
"""
|
||
角色表
|
||
"""
|
||
|
||
__tablename__ = "sys_role"
|
||
|
||
role_name: Mapped[str] = mapped_column(
|
||
String(50), nullable=False, comment="角色名称"
|
||
)
|
||
role_code: Mapped[str] = mapped_column(
|
||
String(50), unique=True, nullable=False, comment="角色编码"
|
||
)
|
||
description: Mapped[Optional[str]] = mapped_column(
|
||
Text, nullable=True, comment="角色描述"
|
||
)
|
||
status: Mapped[bool] = mapped_column(
|
||
Boolean, default=True, comment="状态：0-禁用，1-正常"
|
||
)
|
||
|
||
# 关系
|
||
users: Mapped[List[User]] = relationship(
|
||
"User", secondary=user_roles_table, back_populates="roles"
|
||
)
|
||
|
||
|
||
class Permission(BaseModel):
|
||
"""
|
||
权限表
|
||
"""
|
||
|
||
__tablename__ = "sys_permission"
|
||
|
||
permission_name: Mapped[str] = mapped_column(
|
||
String(50), nullable=False, comment="权限名称"
|
||
)
|
||
permission_code: Mapped[str] = mapped_column(
|
||
String(100), unique=True, nullable=False, comment="权限编码"
|
||
)
|
||
resource_type: Mapped[str] = mapped_column(
|
||
String(20), nullable=False, comment="资源类型：menu-菜单，button-按钮"
|
||
)
|
||
resource_path: Mapped[Optional[str]] = mapped_column(
|
||
String(100), nullable=True, comment="资源路径"
|
||
)
|
||
description: Mapped[Optional[str]] = mapped_column(
|
||
Text, nullable=True, comment="权限描述"
|
||
)
|
||
status: Mapped[bool] = mapped_column(
|
||
Boolean, default=True, comment="状态：0-禁用，1-正常"
|
||
)
|
||
|
||
|
||
class UserLoginLog(BaseModel):
|
||
"""
|
||
用户登录日志表
|
||
"""
|
||
|
||
__tablename__ = "sys_user_login_log"
|
||
|
||
user_id: Mapped[int] = mapped_column(
|
||
Integer, nullable=False, comment="用户ID"
|
||
)
|
||
username: Mapped[str] = mapped_column(
|
||
String(50), nullable=False, comment="用户名"
|
||
)
|
||
login_time: Mapped[datetime] = mapped_column(
|
||
DateTime(timezone=True), default=datetime.utcnow, comment="登录时间"
|
||
)
|
||
login_ip: Mapped[Optional[str]] = mapped_column(
|
||
String(50), nullable=True, comment="登录IP"
|
||
)
|
||
user_agent: Mapped[Optional[str]] = mapped_column(
|
||
Text, nullable=True, comment="用户代理"
|
||
)
|
||
status: Mapped[bool] = mapped_column(
|
||
Boolean, default=True, comment="登录状态：0-失败，1-成功"
|
||
)
|
||
failure_reason: Mapped[Optional[str]] = mapped_column(
|
||
String(255), nullable=True, comment="失败原因"
|
||
)
|
||
|
||
|
||
class UserOperationLog(BaseModel):
|
||
"""
|
||
用户操作日志表
|
||
"""
|
||
|
||
__tablename__ = "sys_user_operation_log"
|
||
|
||
user_id: Mapped[int] = mapped_column(
|
||
Integer, nullable=False, comment="用户ID"
|
||
)
|
||
username: Mapped[str] = mapped_column(
|
||
String(50), nullable=False, comment="用户名"
|
||
)
|
||
module: Mapped[str] = mapped_column(
|
||
String(50), nullable=False, comment="模块名称"
|
||
)
|
||
operation: Mapped[str] = mapped_column(
|
||
String(100), nullable=False, comment="操作名称"
|
||
)
|
||
method: Mapped[Optional[str]] = mapped_column(
|
||
String(10), nullable=True, comment="请求方法"
|
||
)
|
||
path: Mapped[Optional[str]] = mapped_column(
|
||
String(200), nullable=True, comment="请求路径"
|
||
)
|
||
params: Mapped[Optional[Text]] = mapped_column(
|
||
Text, nullable=True, comment="请求参数"
|
||
)
|
||
result: Mapped[Optional[Text]] = mapped_column(
|
||
Text, nullable=True, comment="执行结果"
|
||
)
|
||
operation_time: Mapped[datetime] = mapped_column(
|
||
DateTime(timezone=True), default=datetime.utcnow, comment="操作时间"
|
||
)
|
||
ip_address: Mapped[Optional[str]] = mapped_column(
|
||
String(50), nullable=True, comment="IP地址"
|
||
)
|
||
user_agent: Mapped[Optional[str]] = mapped_column(
|
||
Text, nullable=True, comment="用户代理"
|
||
)
|
||
status: Mapped[bool] = mapped_column(
|
||
Boolean, default=True, comment="操作状态：0-失败，1-成功"
|
||
)
|