TermiNexus
ec43c4dfe6
- core.User 新增 DefaultRoleUUID（UUIDField 可空）+ 手写迁移 0004 - _SetCurrentRoleCookie 登录/注册优先激活默认角色（须持有且启用），否则回退首个角色 - GetUserRoles 返回 IsDefault 标记；新增 SetDefaultRole 接口（空 RoleUUID=清除默认）
331 lines
12 KiB
Python
331 lines
12 KiB
Python
import uuid
|
||
import datetime
|
||
from django.db import models
|
||
from django.contrib.auth.models import BaseUserManager
|
||
from gvsdsdk.model_base import QModel
|
||
from gvsdsdk.model_utils import UUIDObj
|
||
|
||
|
||
"""模块：Core 数据模型
|
||
定义系统核心表：用户、头像、双因素认证、登录日志、账户激活。
|
||
所有 ORM 字段与方法遵循帕斯卡命名法（首字母大写、缩写大写、单单词小写）。"""
|
||
|
||
|
||
def _elt_end_time_default():
|
||
"""已废弃：ELT 令牌过期时间默认值，仅为迁移历史兼容保留"""
|
||
from gvsds.app_config import getint as _app_getint
|
||
days = _app_getint('elt', 'TOKEN_LIFETIME_DAYS', fallback=30)
|
||
return datetime.datetime.utcnow() + datetime.timedelta(days=days)
|
||
|
||
|
||
class UUIDField(models.BinaryField):
|
||
"""UUID 字段：数据库存储为 BINARY(16)，Python 侧自动转换 str/bytes/uuid.UUID"""
|
||
|
||
description = "UUID stored as BINARY(16)"
|
||
|
||
def __init__(self, *args, **kwargs):
|
||
kwargs.setdefault('max_length', 16)
|
||
kwargs.setdefault('editable', False)
|
||
super().__init__(*args, **kwargs)
|
||
|
||
def db_type(self, connection):
|
||
vendor = connection.vendor
|
||
if vendor == 'mysql':
|
||
return 'BINARY(16)'
|
||
elif vendor == 'sqlite':
|
||
return 'BLOB'
|
||
elif vendor == 'postgresql':
|
||
return 'BYTEA'
|
||
return 'BINARY(16)'
|
||
|
||
def from_db_value(self, value, expression, connection):
|
||
"""数据库读取：BINARY(16) → UUIDObj"""
|
||
if value is None:
|
||
return None
|
||
if isinstance(value, UUIDObj):
|
||
return value
|
||
return UUIDObj(value)
|
||
|
||
def get_prep_value(self, value):
|
||
"""写入数据库/查询：UUIDObj/str/bytes/uuid.UUID → BINARY(16)"""
|
||
if value is None:
|
||
return None
|
||
if isinstance(value, UUIDObj):
|
||
return value.bytes
|
||
if isinstance(value, str):
|
||
try:
|
||
return uuid.UUID(value).bytes
|
||
except ValueError:
|
||
return None
|
||
if isinstance(value, uuid.UUID):
|
||
return value.bytes
|
||
if isinstance(value, bytes):
|
||
return value
|
||
return None
|
||
|
||
def get_prep_lookup(self, lookup_type, value):
|
||
"""确保查询时也正确转换 UUID 字符串为 BINARY(16)"""
|
||
if lookup_type in ('exact', 'iexact', 'in'):
|
||
if value is None:
|
||
return None
|
||
if lookup_type == 'in' and isinstance(value, (list, tuple)):
|
||
return [self.get_prep_value(v) for v in value]
|
||
return self.get_prep_value(value)
|
||
return super().get_prep_lookup(lookup_type, value)
|
||
|
||
def value_to_string(self, obj):
|
||
"""序列化：返回 UUID 字符串"""
|
||
value = self.value_from_object(obj)
|
||
if isinstance(value, UUIDObj):
|
||
return value.str or ''
|
||
if isinstance(value, str):
|
||
return value
|
||
if isinstance(value, bytes) and len(value) == 16:
|
||
return str(uuid.UUID(bytes=value))
|
||
if isinstance(value, uuid.UUID):
|
||
return str(value)
|
||
return ''
|
||
|
||
def deconstruct(self):
|
||
name, path, args, kwargs = super().deconstruct()
|
||
del kwargs['max_length']
|
||
return name, path, args, kwargs
|
||
|
||
|
||
class UserManager(BaseUserManager):
|
||
"""用户管理器——提供用户创建与超级用户创建。
|
||
方法名保留蛇形命名以兼容 Django 内部约定。"""
|
||
|
||
def create_user(self, username, password=None, email='', **extra_fields):
|
||
"""创建普通用户"""
|
||
if not username:
|
||
raise ValueError('The Username field must be set')
|
||
user = self.model(
|
||
UserUUID=UUIDObj(uuid.uuid4()),
|
||
UserName=username,
|
||
UserEmail=email or '',
|
||
**extra_fields
|
||
)
|
||
if password:
|
||
user.SetPassword(password)
|
||
user.save(using=self._db)
|
||
return user
|
||
|
||
def create_superuser(self, username, password=None, **extra_fields):
|
||
"""创建超级用户"""
|
||
extra_fields.setdefault('UserAccountLicense', 1)
|
||
extra_fields.setdefault('UserPositionStatus', 0)
|
||
extra_fields.setdefault('IsStaff', True)
|
||
extra_fields.setdefault('IsSuperuser', True)
|
||
return self.create_user(username, password, **extra_fields)
|
||
|
||
|
||
class User(QModel):
|
||
"""用户表——系统核心身份模型"""
|
||
|
||
UserUUID = UUIDField(primary_key=True, verbose_name='用户UUID')
|
||
UserName = models.CharField(max_length=256, unique=True, verbose_name='用户名')
|
||
UserPassword = models.CharField(max_length=60, default='', verbose_name='密码') # 无默认密码（防默认口令后门），须经注册/重置流程设置
|
||
UserPositionStatus = models.SmallIntegerField(default=0, verbose_name='职位状态')
|
||
UserPersonalSignature = models.CharField(max_length=1024, default='尚未及提笔自叙，浅墨先勾鹤影', verbose_name='个人签名')
|
||
UserEmail = models.CharField(max_length=256, default='', verbose_name='邮箱')
|
||
UserEmailVerify = models.SmallIntegerField(default=0, verbose_name='邮箱验证')
|
||
UserAccountLicense = models.SmallIntegerField(default=1, verbose_name='账户许可')
|
||
UserGender = models.SmallIntegerField(default=-1, verbose_name='性别')
|
||
UserTele = models.CharField(max_length=16, null=True, blank=True, verbose_name='电话')
|
||
UserCreateTime = models.DateTimeField(default=datetime.datetime.utcnow, verbose_name='创建时间')
|
||
UserLastLoginDate = models.DateTimeField(default=datetime.datetime.utcnow, verbose_name='最后登录')
|
||
IsStaff = models.BooleanField(default=False, verbose_name='是否员工')
|
||
IsSuperuser = models.BooleanField(default=False, verbose_name='是否超级用户')
|
||
# 默认登录角色：登录/注册后自动激活的角色（空 = 取用户首个角色）
|
||
DefaultRoleUUID = UUIDField(null=True, blank=True, verbose_name='默认角色UUID')
|
||
|
||
objects = UserManager()
|
||
|
||
class Meta:
|
||
db_table = 'User'
|
||
app_label = 'core'
|
||
verbose_name = '用户'
|
||
verbose_name_plural = '用户'
|
||
|
||
def __str__(self):
|
||
return self.UserName
|
||
|
||
@property
|
||
def UUIDStr(self):
|
||
"""返回 UUID 字符串形式"""
|
||
return UUIDObj(self.UserUUID).str
|
||
|
||
@property
|
||
def IsActive(self):
|
||
return self.UserAccountLicense == 1
|
||
|
||
@property
|
||
def IsAnonymous(self):
|
||
return False
|
||
|
||
@property
|
||
def IsAuthenticated(self):
|
||
return True
|
||
|
||
@property
|
||
def PK(self):
|
||
return UUIDObj(self.UserUUID)
|
||
|
||
@property
|
||
def ID(self):
|
||
return UUIDObj(self.UserUUID)
|
||
|
||
def SetPassword(self, raw_password):
|
||
"""设置密码——bcrypt 哈希"""
|
||
import bcrypt
|
||
salt = bcrypt.gensalt(rounds=12)
|
||
hashed = bcrypt.hashpw(raw_password.encode('utf-8'), salt)
|
||
self.UserPassword = hashed.decode('utf-8')
|
||
|
||
def CheckPassword(self, raw_password):
|
||
"""验证密码"""
|
||
import bcrypt
|
||
try:
|
||
return bcrypt.checkpw(
|
||
raw_password.encode('utf-8'),
|
||
self.UserPassword.encode('utf-8')
|
||
)
|
||
except Exception:
|
||
return False
|
||
|
||
def GetTele(self):
|
||
"""获取电话，空则返回默认文本"""
|
||
return self.UserTele or '未填写'
|
||
|
||
def GetCompanies(self):
|
||
"""获取用户所属公司列表（通过 UserRole.CompanyUUID 直接关联）"""
|
||
from gvsdsdk.auth.managers import CompanyManager
|
||
return CompanyManager.GetCompaniesForUser(self.UserUUID)
|
||
|
||
def GetRoles(self):
|
||
"""获取用户角色列表（角色不绑定公司/部门，租户内独立存在）"""
|
||
from gvsdsdk.auth.managers import RoleManager
|
||
return RoleManager.GetRolesForUser(self.UserUUID)
|
||
|
||
def GetDepartments(self):
|
||
"""获取用户部门列表（通过 UserRoleDepartment 关联表）"""
|
||
from gvsdsdk.auth.managers import DepartmentManager
|
||
return DepartmentManager.GetDepartmentsForUser(self.UserUUID)
|
||
|
||
def HasPermission(self, perm_code):
|
||
"""检查用户是否拥有指定权限
|
||
|
||
通过 UserRole → RolePermission → Permission 链路查询，
|
||
汇总所有角色的操作权限后判断（委托 gvsdsdk.RoleManager）。
|
||
"""
|
||
from gvsdsdk.auth.managers import RoleManager
|
||
return RoleManager.HasUserPermission(self.UserUUID, perm_code)
|
||
|
||
# Django auth 框架兼容（内部使用蛇形命名）
|
||
@property
|
||
def is_active(self):
|
||
return self.IsActive
|
||
|
||
@property
|
||
def is_anonymous(self):
|
||
return self.IsAnonymous
|
||
|
||
@property
|
||
def is_authenticated(self):
|
||
return self.IsAuthenticated
|
||
|
||
@property
|
||
def pk(self):
|
||
return self.PK
|
||
|
||
@property
|
||
def id(self):
|
||
return self.ID
|
||
|
||
def set_password(self, raw_password):
|
||
self.SetPassword(raw_password)
|
||
|
||
def check_password(self, raw_password):
|
||
return self.CheckPassword(raw_password)
|
||
|
||
|
||
class UserAvatar(QModel):
|
||
"""用户头像表"""
|
||
|
||
UserUUID = UUIDField(primary_key=True, verbose_name='用户UUID')
|
||
UserAvatar = models.BinaryField(editable=True, verbose_name='头像数据')
|
||
AvatarUploadTime = models.DateTimeField(default=datetime.datetime.utcnow, verbose_name='上传时间')
|
||
AvatarFileType = models.CharField(max_length=16, default='png', verbose_name='文件类型')
|
||
|
||
class Meta:
|
||
db_table = 'UserAvatar'
|
||
app_label = 'core'
|
||
verbose_name = '用户头像'
|
||
verbose_name_plural = '用户头像'
|
||
|
||
|
||
class User2FA(QModel):
|
||
"""用户双因素认证表"""
|
||
|
||
User2FAUUID = UUIDField(primary_key=True, verbose_name='2FA UUID')
|
||
UserUUID = UUIDField(db_index=True, verbose_name='用户UUID')
|
||
SecretKey = models.CharField(max_length=64, verbose_name='密钥')
|
||
IsEnabled = models.SmallIntegerField(default=0, verbose_name='是否启用')
|
||
Verified = models.SmallIntegerField(default=0, verbose_name='已验证')
|
||
LastUsedAt = models.DateTimeField(null=True, blank=True, verbose_name='最后使用')
|
||
CreateTime = models.DateTimeField(default=datetime.datetime.utcnow, verbose_name='创建时间')
|
||
|
||
class Meta:
|
||
db_table = 'User2FA'
|
||
app_label = 'core'
|
||
verbose_name = '双因素认证'
|
||
verbose_name_plural = '双因素认证'
|
||
|
||
|
||
class UserPending2FA(QModel):
|
||
"""双因素认证待验证表"""
|
||
|
||
PendingUUID = UUIDField(primary_key=True, verbose_name='待确认UUID')
|
||
UserUUID = UUIDField(unique=True, verbose_name='用户UUID')
|
||
SecretKey = models.CharField(max_length=64, verbose_name='密钥')
|
||
ExpireAt = models.DateTimeField(verbose_name='过期时间')
|
||
|
||
class Meta:
|
||
db_table = 'UserPending2FA'
|
||
app_label = 'core'
|
||
verbose_name = '2FA待确认'
|
||
verbose_name_plural = '2FA待确认'
|
||
|
||
|
||
class UserLoginLog(QModel):
|
||
"""用户登录日志表"""
|
||
|
||
LogUUID = UUIDField(primary_key=True, verbose_name='日志UUID')
|
||
UserUUID = UUIDField(db_index=True, verbose_name='用户UUID')
|
||
LoginIP = models.CharField(max_length=39, verbose_name='登录IP')
|
||
LoginPlatform = models.CharField(max_length=32, verbose_name='登录平台')
|
||
LoginStatus = models.SmallIntegerField(verbose_name='登录状态')
|
||
LoginTime = models.DateTimeField(default=datetime.datetime.utcnow, verbose_name='登录时间')
|
||
UserAgent = models.CharField(max_length=512, default='', verbose_name='用户代理')
|
||
|
||
class Meta:
|
||
db_table = 'UserLoginLog'
|
||
app_label = 'core'
|
||
verbose_name = '登录日志'
|
||
verbose_name_plural = '登录日志'
|
||
|
||
|
||
class AccountActivation(QModel):
|
||
"""账户激活码表"""
|
||
|
||
AccountActivationCode = models.BinaryField(primary_key=True, verbose_name='激活码')
|
||
UserUUID = UUIDField(verbose_name='用户UUID')
|
||
CreateTime = models.DateTimeField(default=datetime.datetime.utcnow, verbose_name='创建时间')
|
||
|
||
class Meta:
|
||
db_table = 'AccountActivation'
|
||
app_label = 'core'
|
||
verbose_name = '账户激活'
|
||
verbose_name_plural = '账户激活'
|