from typing import Optional from urllib.parse import urlparse from fastapi import Query from pydantic import ( BaseModel, ConfigDict, EmailStr, Field, field_validator, model_validator, ) from app.api.v1.module_system.auth.schema import SmsCodeSchema from app.api.v1.module_system.menu.schema import MenuOutSchema from app.api.v1.module_system.role.schema import RoleOutSchema from app.common.enums import QueueEnum from app.core.base_schema import BaseSchema, CommonSchema, TenantBySchema, UserBySchema from app.core.validator import DateTimeStr, email_validator, mobile_validator class CurrentUserUpdateSchema(BaseModel): """基础用户信息""" name: str | None = Field(default=None, description="名称") mobile: str | None = Field(default=None, description="手机号") email: EmailStr | None = Field(default=None, description="邮箱") gender: str | None = Field(default=None, description="性别") avatar: str | None = Field(default=None, description="头像") @field_validator("mobile") @classmethod def validate_mobile(cls, value: str | None): """ 校验手机号格式（委托到 `mobile_validator`）。 参数: - value (str | None): 手机号。 返回: - str | None: 校验后的手机号。 异常: - CustomException: 手机号格式非法时抛出。 """ return mobile_validator(value) @field_validator("email") @classmethod def validate_email(cls, value: str | None): """ 校验邮箱格式（为空则跳过；否则委托到 `email_validator`）。 参数: - value (str | None): 邮箱。 返回: - str | None: 校验后的邮箱。 异常: - CustomException: 邮箱格式非法时抛出。 """ if not value: return value return email_validator(value) @field_validator("avatar") @classmethod def validate_avatar(cls, value: str | None): """ 校验头像地址为合法的 HTTP/HTTPS URL。 参数: - value (str | None): 头像 URL。 返回: - str | None: 校验后的头像 URL。 异常: - ValueError: 头像 URL 非法时抛出。 """ if not value: return value parsed = urlparse(value) if parsed.scheme in ("http", "https") and parsed.netloc: return value raise ValueError("头像地址需为有效的HTTP/HTTPS URL") @model_validator(mode="after") def check_model(self): """ 校验基础用户信息的长度约束。 返回: - CurrentUserUpdateSchema: 校验后的同一实例。 异常: - ValueError: 字段长度超限时抛出。 """ if self.name and len(self.name) > 32: raise ValueError("名称长度不能超过32个字符") return self class UserRegisterSchema(BaseModel): """注册""" name: str | None = Field(default=None, description="名称") template_name: Optional[str] = Field(default="verify", description="模版名称") invite_code: str = Field(default=None, description="邀请码") mobile: str = Field(default=None, description="手机号") sms_code: Optional[str] = Field(default=None, description="验证码") username: str = Field(..., description="账号") password: str = Field(..., description="密码哈希值") role_ids: list[int] | None = Field(default=[1], description="角色ID") created_id: int | None = Field(default=1, description="创建人ID") description: str | None = Field(default=None, max_length=255, description="备注") @field_validator("mobile") @classmethod def validate_mobile(cls, value: str | None): """ 校验手机号格式（委托到 `mobile_validator`）。 参数: - value (str | None): 手机号。 返回: - str | None: 校验后的手机号。 异常: - CustomException: 手机号格式非法时抛出。 """ return mobile_validator(value) @field_validator("username") @classmethod def validate_username(cls, value: str): """ 校验并规范化账号：字母开头，长度 3-32，仅含字母/数字/_ . -。 参数: - value (str): 账号。 返回: - str: 规范化后的账号。 异常: - ValueError: 账号为空或不满足格式约束时抛出。 """ v = value.strip() if not v: raise ValueError("账号不能为空") # 字母开头，允许字母数字_.- # import re # # if not re.match(r"^[A-Za-z][A-Za-z0-9_.-]{2,31}$", v): # raise ValueError("账号需字母开头，3-32位，仅含字母/数字/_ . -") return v @model_validator(mode="after") def check_model(self): """ 校验注册信息的长度约束。 返回: - UserRegisterSchema: 校验后的同一实例。 异常: - ValueError: 任一字段长度超限时抛出。 """ if self.name and len(self.name) > 32: raise ValueError("名称长度不能超过32个字符") if self.username and len(self.username) > 32: raise ValueError("账号长度不能超过32个字符") if self.description and len(self.description) > 255: raise ValueError("备注长度不能超过255个字符") if self.password and len(self.password) > 128: raise ValueError("密码长度不能超过128个字符") return self class UserForgetPasswordSchema(BaseModel): """忘记密码""" username: str = Field(..., max_length=32, description="用户名") new_password: str = Field(..., max_length=128, description="新密码") mobile: str | None = Field(default=None, description="手机号") @field_validator("mobile") @classmethod def validate_mobile(cls, value: str | None): """ 校验手机号格式（委托到 `mobile_validator`）。 参数: - value (str | None): 手机号。 返回: - str | None: 校验后的手机号。 异常: - CustomException: 手机号格式非法时抛出。 """ return mobile_validator(value) class UserChangePasswordSchema(BaseModel): """修改密码""" old_password: str = Field(..., max_length=128, description="旧密码") new_password: str = Field(..., max_length=128, description="新密码") class ResetPasswordSchema(BaseModel): """重置密码""" id: int = Field(..., description="主键ID") password: str = Field(..., min_length=6, max_length=128, description="新密码") class UserCreateSchema(CurrentUserUpdateSchema): """新增""" model_config = ConfigDict(from_attributes=True) username: str | None = Field(default=None, max_length=32, description="用户名") password: str | None = Field(default=None, max_length=128, description="密码哈希值") status: str = Field(default="0", description="是否可用") description: str | None = Field(default=None, max_length=255, description="备注") is_superuser: bool | None = Field(default=False, description="是否超管") dept_id: int | None = Field(default=None, description="部门ID") tenant_id: int | None = Field(default=None, description="租户ID，仅平台管理员创建时可指定") role_ids: list[int] | None = Field(default=[], description="角色ID") position_ids: list[int] | None = Field(default=[], description="岗位ID") class UserUpdateSchema(UserCreateSchema): """更新""" model_config = ConfigDict(from_attributes=True) last_login: DateTimeStr | None = Field(default=None, description="最后登录时间") class UserOutSchema(UserUpdateSchema, BaseSchema, UserBySchema, TenantBySchema): """响应""" model_config = ConfigDict(arbitrary_types_allowed=True, from_attributes=True) tenant_id: int | None = Field( default=None, exclude=True, description="创建入参使用；列表/详情出参见 tenant", ) gitee_login: str | None = Field(default=None, max_length=32, description="Gitee登录") github_login: str | None = Field(default=None, max_length=32, description="Github登录") wx_login: str | None = Field(default=None, max_length=32, description="微信登录") qq_login: str | None = Field(default=None, max_length=32, description="QQ登录") dept_name: str | None = Field(default=None, description="部门名称") dept: CommonSchema | None = Field(default=None, description="部门") positions: list[CommonSchema] | None = Field(default=[], description="岗位") roles: list[RoleOutSchema] | None = Field(default=[], description="角色") menus: list[MenuOutSchema] | None = Field(default=[], description="菜单") class UserQueryParam: """用户管理查询参数""" def __init__( self, username: str | None = Query(None, description="用户名"), name: str | None = Query(None, description="名称"), mobile: str | None = Query(None, description="手机号", pattern=r"^1[3-9]\d{9}$"), email: str | None = Query( None, description="邮箱", pattern=r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$", ), dept_id: int | None = Query(None, description="部门ID"), tenant_id: int | None = Query(None, description="租户ID（仅平台管理员可筛选）"), status: str | None = Query(None, description="是否可用"), created_time: list[DateTimeStr] | None = Query( None, description="创建时间范围", examples=["2025-01-01 00:00:00", "2025-12-31 23:59:59"], ), updated_time: list[DateTimeStr] | None = Query( None, description="更新时间范围", examples=["2025-01-01 00:00:00", "2025-12-31 23:59:59"], ), created_id: int | None = Query(None, description="创建人"), updated_id: int | None = Query(None, description="更新人"), ) -> None: # 模糊查询字段 self.username = (QueueEnum.like.value, username) self.name = (QueueEnum.like.value, name) self.mobile = (QueueEnum.like.value, mobile) self.email = (QueueEnum.like.value, email) # 精确查询字段 self.dept_id = (QueueEnum.eq.value, dept_id) self.tenant_id = (QueueEnum.eq.value, tenant_id) self.created_id = (QueueEnum.eq.value, created_id) self.updated_id = (QueueEnum.eq.value, updated_id) self.status = (QueueEnum.eq.value, status) # 时间范围查询 if created_time and len(created_time) == 2: self.created_time = (QueueEnum.between.value, (created_time[0], created_time[1])) if updated_time and len(updated_time) == 2: self.updated_time = (QueueEnum.between.value, (updated_time[0], updated_time[1]))