import re from datetime import date, datetime, time from typing import Annotated, Any from pydantic import AfterValidator, PlainSerializer, WithJsonSchema from app.common.constant import DATE_DISPLAY_FMT, DATETIME_DISPLAY_FMT, RET, TIME_DISPLAY_FMT from app.core.exceptions import CustomException # 自定义日期时间字符串类型 DateTimeStr = Annotated[ datetime, AfterValidator(lambda x: datetime_validator(x)), PlainSerializer( lambda x: x.strftime(DATETIME_DISPLAY_FMT) if isinstance(x, datetime) else str(x), return_type=str, when_used="json", ), WithJsonSchema({"type": "string"}, mode="serialization"), ] # 自定义日期字符串类型 DateStr = Annotated[ date, AfterValidator(lambda x: date_validator(x)), PlainSerializer( lambda x: x.strftime(DATE_DISPLAY_FMT) if isinstance(x, date) else str(x), return_type=str, when_used="json", ), WithJsonSchema({"type": "string"}, mode="serialization"), ] # 自定义时间字符串类型 TimeStr = Annotated[ time, AfterValidator(lambda x: time_validator(x)), PlainSerializer( lambda x: x.strftime(TIME_DISPLAY_FMT) if isinstance(x, time) else str(x), return_type=str, when_used="json", ), WithJsonSchema({"type": "string"}, mode="serialization"), ] # 自定义手机号类型 Telephone = Annotated[ str, AfterValidator(lambda x: mobile_validator(x)), PlainSerializer(lambda x: x, return_type=str), WithJsonSchema({"type": "string"}, mode="serialization"), ] # 自定义邮箱类型 Email = Annotated[ str, AfterValidator(lambda x: email_validator(x)), PlainSerializer(lambda x: x, return_type=str), WithJsonSchema({"type": "string"}, mode="serialization"), ] def datetime_validator(value: str | datetime) -> datetime: """ 日期格式验证器。 参数: - value (str | datetime): 日期值。 返回: - datetime: 格式化后的日期。 异常: - CustomException: 日期格式无效时抛出。 """ try: if isinstance(value, str): return datetime.strptime(value, DATETIME_DISPLAY_FMT) if isinstance(value, datetime): return value except Exception: raise CustomException(code=RET.ERROR.code, msg="无效的日期格式") def date_validator(value: str | date) -> date: """ 日期格式验证器。 参数: - value (str | date): 日期值。 返回: - date: 格式化后的日期。 异常: - CustomException: 日期格式无效时抛出。 """ try: if isinstance(value, str): return datetime.strptime(value, DATE_DISPLAY_FMT).date() if isinstance(value, date): return value except Exception: raise CustomException(code=RET.ERROR.code, msg="无效的日期格式") def time_validator(value: str | time) -> time: """ 时间格式验证器。 参数: - value (str | time): 时间值。 返回: - time: 格式化后的时间。 异常: - CustomException: 时间格式无效时抛出。 """ try: if isinstance(value, str): return datetime.strptime(value, TIME_DISPLAY_FMT).time() if isinstance(value, time): return value except Exception: raise CustomException(code=RET.ERROR.code, msg="无效的时间格式") def email_validator(value: str) -> str: """ 邮箱地址验证器。 参数: - value (str): 邮箱地址。 返回: - str: 验证后的邮箱地址。 异常: - CustomException: 邮箱格式无效时抛出。 """ if not value: raise CustomException(code=RET.ERROR.code, msg="邮箱地址不能为空") regex = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$" if not re.match(regex, value): raise CustomException(code=RET.ERROR.code, msg="邮箱地址格式不正确") return value def mobile_validator(value: str | None) -> str | None: """ 手机号验证器。 参数: - value (str | None): 手机号。 返回: - str | None: 验证后的手机号。 异常: - CustomException: 手机号格式无效时抛出。 """ if not value: return value if len(value) != 11 or not value.isdigit(): raise CustomException(code=RET.ERROR.code, msg="手机号格式不正确") regex = r"^1(3\d|4[4-9]|5[0-35-9]|6[67]|7[013-8]|8[0-9]|9[0-9])\d{8}$" if not re.match(regex, value): raise CustomException(code=RET.ERROR.code, msg="手机号格式不正确") return value def validate_required_code(value: str | None) -> str: """ 必填编码校验：字母开头，总长 2–16，仅含字母/数字/下划线。 参数: - value (str | None): 编码。 返回: - str: 去空白后的编码。 异常: - ValueError: 为空或格式不合法。 """ if value is None or not str(value).strip(): raise ValueError("编码不能为空") v = value.strip() if not re.match(r"^[A-Za-z][A-Za-z0-9_]{1,15}$", v): raise ValueError("编码需字母开头，允许字母/数字/下划线，长度2-16") return v def code_validator(value: str | None) -> str | None: """ 可选编码验证器（为空则跳过）。 参数: - value (str | None): 编码。 返回: - str | None: 验证后的编码；未填写时返回 None。 异常: - CustomException: 已填写但格式无效时抛出。 """ if not value: return value v = value.strip() if not v: return None if not re.match(r"^[A-Za-z][A-Za-z0-9_]{1,15}$", v): raise CustomException( code=RET.ERROR.code, msg="编码需字母开头，允许字母/数字/下划线，长度2-16", ) return v def menu_request_validator(data: Any) -> Any: """ 菜单请求数据验证器。 参数: - data (Any): 请求数据。 返回: - Any: 验证后的请求数据。 异常: - CustomException: 请求数据无效时抛出。 """ menu_types = {1: "目录", 2: "功能", 3: "权限", 4: "外链"} if data.type not in menu_types: raise CustomException( code=RET.ERROR.code, msg=f"菜单类型必须为: {','.join(map(str, menu_types.keys()))}", ) if data.type in [1, 2]: if not data.route_name: raise CustomException(code=RET.ERROR.code, msg="路由名称不能为空") if not data.route_path: raise CustomException(code=RET.ERROR.code, msg="路由路径不能为空") if data.type == 1 and not (data.redirect and str(data.redirect).strip()): raise CustomException(code=RET.ERROR.code, msg="目录类型必须填写重定向地址") if data.type == 2 and not data.component_path: raise CustomException(code=RET.ERROR.code, msg="组件路径不能为空") return data def role_permission_request_validator(data: Any) -> Any: """ 角色权限设置数据验证器。 参数: - data (Any): 请求数据。 返回: - Any: 验证后的请求数据。 异常: - CustomException: 请求数据无效时抛出。 """ data_scopes = { 1: "仅本人数据权限", 2: "本部门数据权限", 3: "本部门及以下数据权限", 4: "全部数据权限", 5: "自定义数据权限", } if data.data_scope not in data_scopes: raise CustomException( code=RET.ERROR.code, msg=f"数据权限范围必须为: {','.join(map(str, data_scopes.keys()))}", ) if not data.role_ids: raise CustomException(code=RET.ERROR.code, msg="角色不能为空") return data