feat: 优化表单提交逻辑和添加表单验证

refactor: 重构后端模型验证逻辑 fix: 修复资源文件路径处理和下载服务 style: 调整前端样式和布局 perf: 优化AI错误处理和提示信息 docs: 更新上传组件提示信息 test: 添加表单提交测试用例 chore: 更新依赖和配置文件
2026-09-20 20:39:55 +00:00 · 2025-10-28 23:32:10 +08:00
parent 96335b385f
commit a1f9475f3a
37 changed files with 723 additions and 64 deletions
@@ -1,9 +1,10 @@
 # -*- coding: utf-8 -*-

-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 from typing import Optional
+import re
 from app.core.base_schema import BaseSchema
-from app.core.validator import DateTimeStr
+from app.core.validator import DateTimeStr, datetime_validator


 class JobCreateSchema(BaseModel):
@@ -25,6 +26,52 @@ class JobCreateSchema(BaseModel):
    description: Optional[str] = Field(default=None, max_length=255, description='描述')
    status: Optional[bool] = Field(default=False, description='任务状态:启动,停止')

+    @model_validator(mode='before')
+    @classmethod
+    def _normalize(cls, data):
+        """前置归一化：字符串去空格、布尔/数字兼容转换。"""
+        if isinstance(data, dict):
+            for key in ('name', 'func', 'trigger', 'args', 'kwargs', 'jobstore', 'executor', 'trigger_args', 'start_date', 'end_date', 'description'):
+                val = data.get(key)
+                if isinstance(val, str):
+                    data[key] = val.strip()
+            for bkey in ('coalesce', 'status'):
+                val = data.get(bkey)
+                if isinstance(val, str):
+                    lowered = val.strip().lower()
+                    if lowered in {'true', '1', 'y', 'yes'}:
+                        data[bkey] = True
+                    elif lowered in {'false', '0', 'n', 'no'}:
+                        data[bkey] = False
+                elif isinstance(val, int):
+                    data[bkey] = bool(val)
+            val = data.get('max_instances')
+            if isinstance(val, str) and val.strip().isdigit():
+                data['max_instances'] = int(val.strip())
+        return data
+
+    @field_validator('trigger')
+    @classmethod
+    def _validate_trigger(cls, v: str) -> str:
+        allowed = {'cron', 'interval', 'date'}
+        v = v.strip()
+        if v not in allowed:
+            raise ValueError('触发器必须为 cron/interval/date')
+        return v
+
+    @model_validator(mode='after')
+    def _validate_dates(self):
+        """跨字段校验：结束时间不得早于开始时间。"""
+        if self.start_date and self.end_date:
+            try:
+                start = datetime_validator(self.start_date)
+                end = datetime_validator(self.end_date)
+            except Exception:
+                raise ValueError('时间格式必须为 YYYY-MM-DD HH:MM:SS')
+            if end < start:
+                raise ValueError('结束时间不能早于开始时间')
+        return self
+

 class JobUpdateSchema(JobCreateSchema):
    """定时任务更新模型"""
@@ -4,6 +4,7 @@ from typing import Optional
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

 from app.core.base_schema import BaseSchema
+from urllib.parse import urlparse


 class ApplicationCreateSchema(BaseModel):
@@ -17,7 +18,7 @@ class ApplicationCreateSchema(BaseModel):
    @model_validator(mode="before")
    @classmethod
    def _normalize(cls, data):
-        """模型级前置处理：去除首尾空格，空字符串转为 None（可选字段）。"""
+        """模型级前置处理：去除首尾空格，空字符串转为 None（可选字段），并规范布尔。"""
        if isinstance(data, dict):
            for key in ("name", "access_url", "icon_url", "description"):
                val = data.get(key)
@@ -27,6 +28,16 @@ class ApplicationCreateSchema(BaseModel):
                    if key in ("icon_url", "description") and val == "":
                        val = None
                    data[key] = val
+            # 规范布尔字符串/数字为布尔值
+            status_val = data.get("status")
+            if isinstance(status_val, str):
+                lowered = status_val.strip().lower()
+                if lowered in {"true", "1", "y", "yes"}:
+                    data["status"] = True
+                elif lowered in {"false", "0", "n", "no"}:
+                    data["status"] = False
+            elif isinstance(status_val, int):
+                data["status"] = bool(status_val)
        return data

    @field_validator('name')
@@ -36,6 +47,30 @@ class ApplicationCreateSchema(BaseModel):
            raise ValueError('应用名称长度不能超过64字符')
        return v

+    @field_validator('access_url')
+    @classmethod
+    def _validate_access_url(cls, v: str) -> str:
+        v = v.strip()
+        if not v:
+            raise ValueError('访问地址不能为空')
+        parsed = urlparse(v)
+        if parsed.scheme not in ('http', 'https'):
+            raise ValueError('访问地址必须为 http/https URL')
+        return v
+
+    @field_validator('icon_url')
+    @classmethod
+    def _validate_icon_url(cls, v: Optional[str]) -> Optional[str]:
+        if v is None:
+            return v
+        v = v.strip()
+        if v == "":
+            return None
+        parsed = urlparse(v)
+        if parsed.scheme not in ('http', 'https'):
+            raise ValueError('应用图标URL必须为 http/https URL')
+        return v
+

 class ApplicationUpdateSchema(ApplicationCreateSchema):
    """应用更新模型"""
@@ -1,4 +1,4 @@
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 from pydantic.alias_generators import to_camel
 from typing import Optional

@@ -11,6 +11,35 @@ class ImportFieldModel(BaseModel):
    is_required: Optional[str] = Field(description='是否必传')
    selected: Optional[bool] = Field(description='是否勾选')

+    @model_validator(mode='before')
+    @classmethod
+    def _normalize(cls, data):
+        if isinstance(data, dict):
+            for key in ('base_column', 'excel_column', 'default_value'):
+                val = data.get(key)
+                if isinstance(val, str):
+                    val = val.strip()
+                    if val == '':
+                        val = None
+                    data[key] = val
+            # is_required 兼容转换
+            val = data.get('is_required')
+            if isinstance(val, str):
+                lowered = val.strip().lower()
+                if lowered in {'true', '1', 'y', 'yes'}:
+                    data['is_required'] = True
+                elif lowered in {'false', '0', 'n', 'no'}:
+                    data['is_required'] = False
+        return data
+
+    @model_validator(mode='after')
+    def _validate(self):
+        if self.selected and not (self.base_column and self.base_column.strip()):
+            raise ValueError('选中字段必须提供数据库字段名')
+        if self.is_required and not (self.excel_column and self.excel_column.strip()):
+            raise ValueError('必传字段必须提供excel字段名')
+        return self
+

 class ImportModel(BaseModel):
    model_config = ConfigDict(alias_generator=to_camel, from_attributes=True)
@@ -19,3 +48,29 @@ class ImportModel(BaseModel):
    filed_info: Optional[list[ImportFieldModel]] = Field(description='字段关联表')
    file_name: Optional[str] = Field(description='文件名')

+    @model_validator(mode='before')
+    @classmethod
+    def _normalize(cls, data):
+        if isinstance(data, dict):
+            for key in ('table_name', 'sheet_name', 'file_name'):
+                val = data.get(key)
+                if isinstance(val, str):
+                    val = val.strip()
+                    if val == '':
+                        val = None
+                    data[key] = val
+        return data
+
+    @model_validator(mode='after')
+    def _validate(self):
+        # excel_column 不重复（忽略 None）
+        if self.filed_info:
+            seen = set()
+            for f in self.filed_info:
+                if f.excel_column:
+                    key = f.excel_column.strip()
+                    if key in seen:
+                        raise ValueError('excel字段名存在重复')
+                    seen.add(key)
+        return self
+
@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-

 from typing import Optional
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

 from app.core.base_schema import BaseSchema

@@ -12,6 +12,65 @@ class DemoCreateSchema(BaseModel):
    status: bool = Field(True, description="是否启用(True:启用 False:禁用)")
    description: Optional[str] = Field(default=None, max_length=255, description="描述")

+    @field_validator('name')
+    @classmethod
+    def _validate_name(cls, v: str) -> str:
+        v = v.strip()
+        if not v:
+            raise ValueError('名称不能为空')
+        return v
+
+    @model_validator(mode='before')
+    @classmethod
+    def _normalize(cls, data):
+        if isinstance(data, dict):
+            for key in ('name', 'description'):
+                val = data.get(key)
+                if isinstance(val, str):
+                    val = val.strip()
+                    if key == 'description' and val == '':
+                        val = None
+                    data[key] = val
+            # status兼容
+            val = data.get('status')
+            if isinstance(val, str):
+                lowered = val.strip().lower()
+                if lowered in {'true', '1', 'y', 'yes'}:
+                    data['status'] = True
+                elif lowered in {'false', '0', 'n', 'no'}:
+                    data['status'] = False
+            elif isinstance(val, int):
+                data['status'] = bool(val)
+        return data
+
+    @model_validator(mode='wrap')
+    @classmethod
+    def _wrap(cls, data, handler):
+        # 进一步处理：压缩名称/描述中的多余空白，并支持更多 status 同义词
+        if isinstance(data, dict):
+            name = data.get('name')
+            if isinstance(name, str):
+                data['name'] = ' '.join(name.split())
+            status_val = data.get('status')
+            if isinstance(status_val, str):
+                lowered = status_val.strip().lower()
+                if lowered in {'enabled', 'enable', 'on'}:
+                    data['status'] = True
+                elif lowered in {'disabled', 'disable', 'off'}:
+                    data['status'] = False
+            desc = data.get('description')
+            if isinstance(desc, str):
+                data['description'] = ' '.join(desc.split())
+        result = handler(data)
+        return result
+
+    @model_validator(mode='after')
+    def _check_disabled_requires_description(self):
+        # 业务示例：禁用时必须填写描述
+        if self.status is False and (self.description is None or (isinstance(self.description, str) and self.description.strip() == '')):
+            raise ValueError('禁用时必须填写描述')
+        return self
+

 class DemoUpdateSchema(DemoCreateSchema):
    """更新模型"""
@@ -2,8 +2,9 @@

 from typing import Optional, List, Dict, Any
 from datetime import datetime
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 from pathlib import Path
+from urllib.parse import urlparse


 class ResourceItemSchema(BaseModel):
@@ -20,6 +21,33 @@ class ResourceItemSchema(BaseModel):
    modified_time: Optional[datetime] = Field(None, description="修改时间")
    is_hidden: bool = Field(False, description="是否为隐藏文件")

+    @field_validator('file_url')
+    @classmethod
+    def _validate_file_url(cls, v: str) -> str:
+        v = v.strip()
+        parsed = urlparse(v)
+        if parsed.scheme not in ('http', 'https'):
+            raise ValueError('文件URL必须为 http/https')
+        return v
+
+    @field_validator('relative_path')
+    @classmethod
+    def _validate_relative_path(cls, v: str) -> str:
+        v = v.strip()
+        if '..' in v or v.startswith('\\'):
+            raise ValueError('相对路径包含不安全字符')
+        return v
+
+    @model_validator(mode='after')
+    def _validate_flags(self):
+        if self.is_file and self.is_dir:
+            raise ValueError('不能同时为文件和目录')
+        if not self.is_file and not self.is_dir:
+            raise ValueError('必须是文件或目录之一')
+        # 根据名称自动修正隐藏标记
+        self.is_hidden = self.name.startswith('.')
+        return self
+

 class ResourceDirectorySchema(BaseModel):
    """资源目录模型"""
@@ -78,6 +106,14 @@ class ResourceRenameSchema(BaseModel):
            raise ValueError("参数不能为空")
        return value.strip()

+    @field_validator('new_name')
+    @classmethod
+    def _validate_new_name(cls, v: str) -> str:
+        v = v.strip()
+        if '..' in v or '/' in v or '\\' in v:
+            raise ValueError('新名称包含不安全字符')
+        return v
+

 class ResourceCreateDirSchema(BaseModel):
    """创建目录模型"""
@@ -5,6 +5,7 @@ import shutil
 from datetime import datetime
 from typing import List, Dict, Any, Optional
 from pathlib import Path
+from urllib.parse import urlparse
 from fastapi import UploadFile

 from app.core.exceptions import CustomException
@@ -61,6 +62,26 @@ class ResourceService:
        if not path:
            return resource_root
        
+        # 支持前端传递的完整URL或以STATIC_URL/ROOT_PATH+STATIC_URL开头的URL路径，转换为相对资源路径
+        if isinstance(path, str):
+            static_prefix = settings.STATIC_URL.rstrip('/')
+            root_prefix = settings.ROOT_PATH.rstrip('/') if getattr(settings, 'ROOT_PATH', '') else ''
+            root_static_prefix = f"{root_prefix}{static_prefix}" if root_prefix else static_prefix
+            
+            def strip_prefix(p: str) -> str:
+                if p.startswith(root_static_prefix):
+                    return p[len(root_static_prefix):].lstrip('/')
+                if p.startswith(static_prefix):
+                    return p[len(static_prefix):].lstrip('/')
+                return p
+            
+            if path.startswith('http://') or path.startswith('https://'):
+                parsed = urlparse(path)
+                url_path = parsed.path or ''
+                path = strip_prefix(url_path)
+            else:
+                path = strip_prefix(path)
+        
        # 清理路径，移除危险字符
        path = path.strip().replace('..', '').replace('//', '/')
        
@@ -78,9 +99,9 @@ class ResourceService:
            raise CustomException(msg=f'访问路径不在允许范围内: {path}')
        
        # 防止路径遍历攻击
-        if '..' in safe_path or safe_path.count('/') > cls.MAX_PATH_DEPTH:  # 限制最大目录深度
+        if '..' in safe_path or safe_path.count('/') > cls.MAX_PATH_DEPTH:
            raise CustomException(msg=f'不安全的路径格式: {path}')
-            
+        
        return safe_path
    
    @classmethod
@@ -522,14 +543,14 @@ class ResourceService:
    @classmethod
    async def download_file_service(cls, file_path: str, base_url: Optional[str] = None) -> str:
        """
-        下载文件（返回文件路径）
+        下载文件（返回本地文件系统路径）
        
        参数:
-        - file_path (str): 文件路径。
-        - base_url (Optional[str]): 基础URL，用于生成完整URL。
+        - file_path (str): 文件路径（可为相对路径、绝对路径或完整URL）。
+        - base_url (Optional[str]): 基础URL，用于生成完整URL（不再直接返回URL）。
        
        返回:
-        - str: 文件访问URL。
+        - str: 本地文件系统路径。
        """
        try:
            safe_path = cls._get_safe_path(file_path)
@@ -540,10 +561,9 @@ class ResourceService:
            if not os.path.isfile(safe_path):
                raise CustomException(msg='路径不是文件')
            
-            # 生成HTTP URL路径而不是返回文件系统路径
-            http_url = cls._generate_http_url(safe_path, base_url)
-            logger.info(f"生成文件访问URL: {http_url}")
-            return http_url
+            # 返回本地文件路径给 FileResponse 使用
+            logger.info(f"定位文件路径: {safe_path}")
+            return safe_path
            
        except CustomException:
            raise
@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-

 from typing import Optional, List
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

 from app.core.base_schema import BaseSchema

@@ -23,6 +23,45 @@ class DeptCreateSchema(BaseModel):
        value = value.replace(" ", "")
        return value

+    @field_validator('code')
+    @classmethod
+    def validate_code(cls, value: Optional[str]):
+        if value is None:
+            return value
+        v = value.strip()
+        if v == "":
+            return None
+        import re
+        if not re.match(r'^[A-Za-z][A-Za-z0-9_]*$', v):
+            raise ValueError("部门编码必须以字母开头，且仅包含字母/数字/下划线")
+        return v
+
+    @model_validator(mode='before')
+    @classmethod
+    def _normalize(cls, data):
+        if isinstance(data, dict):
+            for key in ('code', 'description'):
+                val = data.get(key)
+                if isinstance(val, str):
+                    val = val.strip()
+                    if key == 'description' and val == '':
+                        val = None
+                    data[key] = val
+            pid = data.get('parent_id')
+            if isinstance(pid, str) and pid.strip().isdigit():
+                data['parent_id'] = int(pid.strip())
+            # status兼容
+            status_val = data.get('status')
+            if isinstance(status_val, str):
+                lowered = status_val.strip().lower()
+                if lowered in {'true', '1', 'y', 'yes'}:
+                    data['status'] = True
+                elif lowered in {'false', '0', 'n', 'no'}:
+                    data['status'] = False
+            elif isinstance(status_val, int):
+                data['status'] = bool(status_val)
+        return data
+

 class DeptUpdateSchema(DeptCreateSchema):
    """部门更新模型"""
@@ -56,6 +56,7 @@ class DictDataCreateSchema(BaseModel):
    description: Optional[str] = Field(default=None, max_length=255, description="描述")

    @field_validator('dict_label')
+    @classmethod
    def validate_dict_label(cls, value: str):
        if not value or value.strip() == '':
            raise ValueError('字典标签不能为空')
@@ -1,9 +1,10 @@
 # -*- coding: utf-8 -*-

 from typing import Optional
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

 from app.core.base_schema import BaseSchema
+import re


 class OperationLogCreateSchema(BaseModel):
@@ -22,6 +23,55 @@ class OperationLogCreateSchema(BaseModel):
    description: Optional[str] = Field(default=None, max_length=255, description="描述")
    creator_id: Optional[int] = Field(default=None, description="创建人ID")

+    @model_validator(mode='before')
+    @classmethod
+    def _normalize(cls, values):
+        if isinstance(values, dict):
+            # 字符串去空格
+            for k in ["request_path", "request_method", "request_payload", "request_ip", "login_location", "request_os", "request_browser", "response_json", "process_time", "description"]:
+                if k in values and isinstance(values[k], str):
+                    values[k] = values[k].strip() or None if values[k].strip() == "" and k in {"request_payload", "response_json", "description"} else values[k].strip()
+            # 方法大写
+            if "request_method" in values and isinstance(values["request_method"], str):
+                values["request_method"] = values["request_method"].strip().upper()
+            # 响应码转整数
+            if "response_code" in values and isinstance(values["response_code"], str):
+                try:
+                    values["response_code"] = int(values["response_code"].strip())
+                except Exception:
+                    pass
+        return values
+
+    @field_validator("type")
+    @classmethod
+    def _validate_type(cls, value: Optional[int]):
+        if value is None:
+            return value
+        if value not in {1, 2}:
+            raise ValueError("日志类型仅支持 1(登录) 或 2(操作)")
+        return value
+
+    @field_validator("request_method")
+    @classmethod
+    def _validate_method(cls, value: Optional[str]):
+        if value is None:
+            return value
+        allowed = {"GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD"}
+        if value.upper() not in allowed:
+            raise ValueError(f"请求方法必须为 {', '.join(sorted(allowed))}")
+        return value.upper()
+
+    @field_validator("request_ip")
+    @classmethod
+    def _validate_ip(cls, value: Optional[str]):
+        if value is None or value == "":
+            return value
+        ipv4 = r"^(25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)){3}$"
+        ipv6 = r"^([0-9a-fA-F]{1,4}:){7}[0-9a-fA-F]{1,4}$"
+        if not re.match(ipv4, value) and not re.match(ipv6, value):
+            raise ValueError("请求IP必须为有效的IPv4或IPv6地址")
+        return value
+

 class OperationLogOutSchema(OperationLogCreateSchema, BaseSchema):
    """日志响应模型"""
@@ -28,6 +28,36 @@ class MenuCreateSchema(BaseModel):
    parent_id: Optional[int] = Field(default=None, ge=1, description="父菜单ID")
    description: Optional[str] = Field(default=None, max_length=255, description="描述")

+    @model_validator(mode='before')
+    @classmethod
+    def _normalize(cls, values):
+        if isinstance(values, dict):
+            # 字符串去空格
+            for k in ["name", "icon", "permission", "route_name", "route_path", "component_path", "redirect", "title", "description"]:
+                if k in values and isinstance(values[k], str):
+                    values[k] = values[k].strip() or None if values[k].strip() == "" else values[k].strip()
+            # 布尔兼容
+            for k in ["status", "keep_alive", "hidden", "always_show", "affix"]:
+                if k in values and isinstance(values[k], str):
+                    values[k] = values[k].strip().lower() in {"true", "1", "yes", "y"}
+            # 父ID转整型
+            if "parent_id" in values and isinstance(values["parent_id"], str):
+                try:
+                    values["parent_id"] = int(values["parent_id"].strip())
+                except Exception:
+                    pass
+            # 路由名/路径规范
+            import re
+            if "route_name" in values and isinstance(values["route_name"], str):
+                rn = values["route_name"]
+                if rn and not re.match(r"^[A-Za-z][A-Za-z0-9_.-]{1,99}$", rn):
+                    raise ValueError("路由名称需字母开头，仅含字母/数字/_ . -")
+            if "route_path" in values and isinstance(values["route_path"], str):
+                rp = values["route_path"]
+                if rp and not rp.startswith("/"):
+                    raise ValueError("路由路径需以 / 开头")
+        return values
+
    @model_validator(mode='after')
    def validate_fields(self):
        return menu_request_validator(self)
@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-

 from typing import Optional
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

 from app.core.base_schema import BaseSchema

@@ -14,6 +14,42 @@ class NoticeCreateSchema(BaseModel):
    status: bool = Field(default=True, description="是否启用(True:启用 False:禁用)")
    description: Optional[str] = Field(default=None, max_length=255, description="描述")

+    @model_validator(mode='before')
+    @classmethod
+    def _normalize(cls, values):
+        if isinstance(values, dict):
+            # 字符串去空格
+            for k in ["notice_title", "notice_type", "notice_content", "description"]:
+                if k in values and isinstance(values[k], str):
+                    values[k] = values[k].strip() or None if values[k].strip() == "" and k == "description" else values[k].strip()
+            # 布尔兼容
+            if "status" in values and isinstance(values["status"], str):
+                values["status"] = values["status"].strip().lower() in {"true", "1", "yes", "y"}
+            # 类型映射
+            mapping = {"1": "1", "2": "2", "通知": "1", "公告": "2", "notice": "1", "announcement": "2"}
+            if "notice_type" in values and isinstance(values["notice_type"], str):
+                v = values["notice_type"].strip().lower()
+                if v in mapping:
+                    values["notice_type"] = mapping[v]
+        return values
+
+    @field_validator("notice_type")
+    @classmethod
+    def _validate_notice_type(cls, value: str):
+        if value not in {"1", "2"}:
+            raise ValueError("公告类型仅支持 '1'(通知) 或 '2'(公告)")
+        return value
+
+    @model_validator(mode='after')
+    def _validate_after(self):
+        if not self.notice_title.strip():
+            raise ValueError("公告标题不能为空")
+        if not self.notice_content.strip():
+            raise ValueError("公告内容不能为空")
+        if self.status is False and (not self.description or not str(self.description).strip()):
+            raise ValueError("禁用状态下必须填写描述")
+        return self
+

 class NoticeUpdateSchema(NoticeCreateSchema):
    """公告通知更新模型"""
@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-

 from typing import Optional
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

 from app.core.base_schema import BaseSchema

@@ -15,6 +15,43 @@ class ParamsCreateSchema(BaseModel):
    status: bool = Field(default=True, description="状态(True:正常 False:停用)")
    description: Optional[str] = Field(default=None, max_length=500, description="描述")

+    @model_validator(mode='before')
+    @classmethod
+    def _normalize(cls, data):
+        """前置归一化：字符串去空格、空串转 None、布尔兼容转换，并规范键为小写。"""
+        if isinstance(data, dict):
+            for key in ('config_name', 'config_key', 'config_value', 'description'):
+                val = data.get(key)
+                if isinstance(val, str):
+                    val = val.strip()
+                    if key in ('config_value', 'description') and val == '':
+                        val = None
+                    data[key] = val
+            # 规范键为小写
+            if isinstance(data.get('config_key'), str):
+                data['config_key'] = data['config_key'].lower()
+            # 规范布尔
+            for bkey in ('config_type', 'status'):
+                val = data.get(bkey)
+                if isinstance(val, str):
+                    lowered = val.strip().lower()
+                    if lowered in {'true', '1', 'y', 'yes'}:
+                        data[bkey] = True
+                    elif lowered in {'false', '0', 'n', 'no'}:
+                        data[bkey] = False
+                elif isinstance(val, int):
+                    data[bkey] = bool(val)
+        return data
+
+    @field_validator('config_key')
+    @classmethod
+    def _validate_config_key(cls, v: str) -> str:
+        v = v.strip().lower()
+        import re
+        if not re.match(r'^[a-z][a-z0-9_.-]*$', v):
+            raise ValueError('参数键名必须以小写字母开头，仅包含小写字母/数字/_.-')
+        return v
+

 class ParamsUpdateSchema(ParamsCreateSchema):
    """配置更新模型"""
@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-

 from typing import Optional
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

 from app.core.base_schema import BaseSchema
 from app.core.validator import DateTimeStr
@@ -13,6 +13,41 @@ class PositionCreateSchema(BaseModel):
    status: bool = Field(default=True, description="是否启用(True:启用 False:禁用)")
    description: Optional[str] = Field(default=None, max_length=255, description="描述")

+    @field_validator('name')
+    @classmethod
+    def _validate_name(cls, v: str) -> str:
+        v = v.strip()
+        if not v:
+            raise ValueError('岗位名称不能为空')
+        return v
+
+    @model_validator(mode='before')
+    @classmethod
+    def _normalize(cls, data):
+        if isinstance(data, dict):
+            for key in ('name', 'description'):
+                val = data.get(key)
+                if isinstance(val, str):
+                    val = val.strip()
+                    if key == 'description' and val == '':
+                        val = None
+                    data[key] = val
+            # order字符串转为整数
+            order_val = data.get('order')
+            if isinstance(order_val, str) and order_val.strip().isdigit():
+                data['order'] = int(order_val.strip())
+            # status兼容
+            status_val = data.get('status')
+            if isinstance(status_val, str):
+                lowered = status_val.strip().lower()
+                if lowered in {'true', '1', 'y', 'yes'}:
+                    data['status'] = True
+                elif lowered in {'false', '0', 'n', 'no'}:
+                    data['status'] = False
+            elif isinstance(status_val, int):
+                data['status'] = bool(status_val)
+        return data
+

 class PositionUpdateSchema(PositionCreateSchema):
    """岗位更新模型"""
@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-

 from typing import List, Optional
-from pydantic import BaseModel, ConfigDict, Field, model_validator
+from pydantic import BaseModel, ConfigDict, Field, model_validator, field_validator

 from app.core.base_schema import BaseSchema
 from app.core.validator import role_permission_request_validator
@@ -19,6 +19,49 @@ class RoleCreateSchema(BaseModel):
    status: bool = Field(default=True, description="是否启用")
    description: Optional[str] = Field(default=None, max_length=255, description="描述")

+    @field_validator("code")
+    @classmethod
+    def validate_code(cls, value: Optional[str]):
+        if value is None:
+            return value
+        import re
+        v = value.strip()
+        if not re.match(r"^[A-Za-z][A-Za-z0-9_]{1,39}$", v):
+            raise ValueError("角色编码需字母开头，允许字母/数字/下划线，长度2-40")
+        return v
+
+    @field_validator("name")
+    @classmethod
+    def validate_name(cls, value: str):
+        v = value.strip()
+        if not v:
+            raise ValueError("角色名称不能为空")
+        return v
+
+    @model_validator(mode='before')
+    @classmethod
+    def _normalize(cls, values):
+        if isinstance(values, dict):
+            for k in ["name", "code", "description"]:
+                if k in values and isinstance(values[k], str):
+                    values[k] = values[k].strip() or None if values[k].strip() == "" else values[k].strip()
+            # bool 兼容
+            if "status" in values and isinstance(values["status"], str):
+                values["status"] = values["status"].strip().lower() in {"true", "1", "yes", "y"}
+            # 数字兼容
+            if "order" in values and isinstance(values["order"], str):
+                try:
+                    values["order"] = int(values["order"].strip())
+                except Exception:
+                    pass
+        return values
+
+    @model_validator(mode='after')
+    def _validate_after(self):
+        if self.status is False and (not self.description or not str(self.description).strip()):
+            raise ValueError("禁用状态下必须填写描述")
+        return self
+

 class RolePermissionSettingSchema(BaseModel):
    """角色权限配置模型"""
@@ -27,6 +70,18 @@ class RolePermissionSettingSchema(BaseModel):
    menu_ids: List[int] = Field(default_factory=list, description='菜单ID列表')
    dept_ids: List[int] = Field(default_factory=list, description='部门ID列表')
    
+    @model_validator(mode='before')
+    @classmethod
+    def _normalize(cls, values):
+        if isinstance(values, dict):
+            for k in ["role_ids", "menu_ids", "dept_ids"]:
+                if k in values and values[k] is not None:
+                    try:
+                        values[k] = list({int(x) for x in values[k]})
+                    except Exception:
+                        pass
+        return values
+    
    @model_validator(mode='after')
    def validate_fields(self):
        """验证权限配置字段"""
@@ -1,11 +1,12 @@
 # -*- coding: utf-8 -*-

 from typing import Optional, List
-from pydantic import BaseModel, ConfigDict, Field, EmailStr, field_validator
+from pydantic import BaseModel, ConfigDict, Field, EmailStr, field_validator, model_validator

 from app.core.validator import DateTimeStr, mobile_validator
 from app.core.base_schema import BaseSchema, CommonSchema
 from app.api.v1.module_system.role.schema import RoleOutSchema
+from urllib.parse import urlparse

 class CurrentUserUpdateSchema(BaseModel):
    """基础用户信息"""
@@ -20,6 +21,16 @@ class CurrentUserUpdateSchema(BaseModel):
    def validate_mobile(cls, value: Optional[str]):
        return mobile_validator(value)

+    @field_validator("avatar")
+    @classmethod
+    def validate_avatar(cls, value: Optional[str]):
+        if not value:
+            return value
+        parsed = urlparse(value)
+        if parsed.scheme in ("http", "https") and parsed.netloc:
+            return value
+        raise ValueError("头像地址需为有效的HTTP/HTTPS URL")
+

 class UserRegisterSchema(BaseModel):
    """注册"""
@@ -36,6 +47,36 @@ class UserRegisterSchema(BaseModel):
    def validate_mobile(cls, value: Optional[str]):
        return mobile_validator(value)

+    @field_validator("username")
+    @classmethod
+    def validate_username(cls, value: str):
+        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='before')
+    @classmethod
+    def _normalize(cls, values):
+        if isinstance(values, dict):
+            for k in ["name", "username", "password", "description"]:
+                if k in values and isinstance(values[k], str):
+                    values[k] = values[k].strip() or values[k]
+            # role_ids 去重并转为 int
+            if "role_ids" in values and values["role_ids"] is not None:
+                try:
+                    values["role_ids"] = list[int]({int(x) for x in values["role_ids"]})
+                except Exception:
+                    pass
+            # mobile 空串转 None
+            if "mobile" in values and isinstance(values["mobile"], str) and values["mobile"].strip() == "":
+                values["mobile"] = None
+        return values
+

 class UserForgetPasswordSchema(BaseModel):
    """忘记密码"""
@@ -75,6 +116,35 @@ class UserCreateSchema(CurrentUserUpdateSchema):
    role_ids: Optional[List[int]] = Field(default=[], description='角色ID')
    position_ids: Optional[List[int]] = Field(default=[], description='岗位ID')

+    @model_validator(mode='before')
+    @classmethod
+    def _normalize(cls, values):
+        if isinstance(values, dict):
+            # 字符串去空格和空串转 None
+            for k in ["username", "password", "description", "name"]:
+                if k in values and isinstance(values[k], str):
+                    values[k] = values[k].strip() or None if values[k].strip() == "" else values[k].strip()
+            # bool 兼容
+            for k in ["status", "is_superuser"]:
+                if k in values:
+                    v = values[k]
+                    if isinstance(v, str):
+                        values[k] = v.strip().lower() in {"true", "1", "yes", "y"}
+            # 列表转 int 去重
+            for k in ["role_ids", "position_ids"]:
+                if k in values and values[k] is not None:
+                    try:
+                        values[k] = list({int(x) for x in values[k]})
+                    except Exception:
+                        pass
+        return values
+
+    @model_validator(mode='after')
+    def _validate_after(self):
+        if self.status is False and (not self.description or not str(self.description).strip()):
+            raise ValueError("禁用状态下必须填写备注描述")
+        return self
+

 class UserUpdateSchema(UserCreateSchema):
    """更新"""