Spaces:

jun2024
/

fachuan

Configuration error

fachuan / apps /core /utils /validators.py

jun2024

Upload 2241 files

f8b03c1 verified 7 months ago

11.1 kB

"""
通用数据验证工具模块
提供常用的验证函数和验证器
"""

import re
from datetime import date, datetime
from decimal import Decimal, InvalidOperation
from typing import Any

from apps.core.exceptions import ValidationException


class Validators:
    """通用验证器集合"""

    # 正则表达式模式
    PHONE_PATTERN = re.compile(r"^1[3-9]\d{9}$")
    EMAIL_PATTERN = re.compile(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
    ID_CARD_PATTERN = re.compile(r"^[1-9]\d{5}(18|19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[\dXx]$")
    SOCIAL_CREDIT_CODE_PATTERN = re.compile(r"^[0-9A-HJ-NPQRTUWXY]{2}\d{6}[0-9A-HJ-NPQRTUWXY]{10}$")

    @classmethod
    def validate_phone(cls, phone: str | None, field_name: str = "phone") -> str | None:
        """
        验证手机号码
        Args:
            phone: 手机号码
            field_name: 字段名（用于错误信息）
        Returns:
            验证后的手机号码
        Raises:
            ValidationException: 验证失败
        """
        if not phone:
            return None

        phone = phone.strip()
        if not phone:
            return None
        if not cls.PHONE_PATTERN.match(phone):
            raise ValidationException("手机号码格式不正确", errors={field_name: "请输入有效的11位手机号码"})
        return phone

    @classmethod
    def validate_email(cls, email: str | None, field_name: str = "email") -> str | None:
        """验证邮箱地址"""
        if not email:
            return None

        email = email.strip().lower()
        if not cls.EMAIL_PATTERN.match(email):
            raise ValidationException("邮箱格式不正确", errors={field_name: "请输入有效的邮箱地址"})
        return email

    @classmethod
    def validate_id_card(cls, id_card: str | None, field_name: str = "id_card") -> str | None:
        """验证身份证号码"""
        if not id_card:
            return None

        id_card = id_card.strip().upper()
        if not cls.ID_CARD_PATTERN.match(id_card):
            raise ValidationException("身份证号码格式不正确", errors={field_name: "请输入有效的18位身份证号码"})

        # 校验码验证
        if not cls._verify_id_card_checksum(id_card):
            raise ValidationException("身份证号码校验失败", errors={field_name: "身份证号码校验码不正确"})

        return id_card

    @classmethod
    def _verify_id_card_checksum(cls, id_card: str) -> bool:
        """验证身份证校验码"""
        weights = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
        check_codes = "10X98765432"

        try:
            total = sum(int(id_card[i]) * weights[i] for i in range(17))
            return check_codes[total % 11] == id_card[17]
        except (ValueError, IndexError):
            return False

    @classmethod
    def validate_social_credit_code(cls, code: str | None, field_name: str = "social_credit_code") -> str | None:
        """验证统一社会信用代码"""
        if not code:
            return None

        code = code.strip().upper()
        if not cls.SOCIAL_CREDIT_CODE_PATTERN.match(code):
            raise ValidationException(
                "统一社会信用代码格式不正确", errors={field_name: "请输入有效的18位统一社会信用代码"}
            )
        return code

    @classmethod
    def validate_required(cls, value: Any, field_name: str) -> Any:
        """验证必填字段"""
        if value is None or (isinstance(value, str) and not value.strip()):
            raise ValidationException(f"{field_name} 不能为空", errors={field_name: "此字段为必填项"})
        return value

    @classmethod
    def validate_length(
        cls,
        value: str | None,
        field_name: str,
        min_length: int | None = None,
        max_length: int | None = None,
    ) -> str | None:
        """验证字符串长度"""
        if value is None or value == "":
            return None

        length = len(value)

        if min_length is not None and length < min_length:
            raise ValidationException(f"{field_name} 长度不足", errors={field_name: f"最少需要 {min_length} 个字符"})

        if max_length is not None and length > max_length:
            raise ValidationException(f"{field_name} 长度超限", errors={field_name: f"最多允许 {max_length} 个字符"})

        return value

    @classmethod
    def validate_range(
        cls,
        value: float | None,
        field_name: str,
        min_value: float | None = None,
        max_value: float | None = None,
    ) -> float | None:
        """验证数值范围"""
        if value is None:
            return value

        if min_value is not None and value < min_value:
            raise ValidationException(f"{field_name} 值过小", errors={field_name: f"最小值为 {min_value}"})

        if max_value is not None and value > max_value:
            raise ValidationException(f"{field_name} 值过大", errors={field_name: f"最大值为 {max_value}"})

        return value

    @classmethod
    def validate_decimal(
        cls,
        value: Any,
        field_name: str,
        max_digits: int = 14,
        decimal_places: int = 2,
    ) -> Decimal | None:
        """验证并转换为 Decimal"""
        if value is None:
            return None

        try:
            decimal_value = Decimal(str(value))
        except (InvalidOperation, ValueError) as e:
            raise ValidationException(f"{field_name} 格式不正确", errors={field_name: "请输入有效的数字"}) from e

        # 检查精度
        sign, digits, exponent = decimal_value.as_tuple()
        total_digits = len(digits)
        decimal_digits = -exponent if isinstance(exponent, int) and exponent < 0 else 0
        integer_digits = total_digits - decimal_digits

        if integer_digits + decimal_places > max_digits:
            raise ValidationException(
                f"{field_name} 数值过大", errors={field_name: f"整数部分最多 {max_digits - decimal_places} 位"}
            )

        if decimal_digits > decimal_places:
            raise ValidationException(
                f"{field_name} 小数位数过多", errors={field_name: f"最多保留 {decimal_places} 位小数"}
            )

        return decimal_value

    @classmethod
    def validate_date(
        cls,
        value: Any,
        field_name: str,
        min_date: date | None = None,
        max_date: date | None = None,
    ) -> date | None:
        """验证日期"""
        if value is None:
            return None

        if isinstance(value, datetime):
            value = value.date()
        elif isinstance(value, str):
            try:
                value = datetime.strptime(value, "%Y-%m-%d").date()
            except ValueError as e:
                raise ValidationException(
                    f"{field_name} 日期格式不正确", errors={field_name: "请使用 YYYY-MM-DD 格式"}
                ) from e
        elif not isinstance(value, date):
            raise ValidationException(f"{field_name} 类型不正确", errors={field_name: "请提供有效的日期"})

        if min_date and value < min_date:
            raise ValidationException(f"{field_name} 日期过早", errors={field_name: f"日期不能早于 {min_date}"})

        if max_date and value > max_date:
            raise ValidationException(f"{field_name} 日期过晚", errors={field_name: f"日期不能晚于 {max_date}"})

        return value  # type: ignore[no-any-return]

    @classmethod
    def validate_in_choices(
        cls,
        value: Any,
        field_name: str,
        choices: list[Any],
        allow_none: bool = True,
    ) -> Any:
        """验证值是否在选项列表中"""
        if value is None:
            if allow_none:
                return None
            raise ValidationException(f"{field_name} 不能为空", errors={field_name: "此字段为必填项"})

        if value not in choices:
            raise ValidationException(
                f"{field_name} 值无效", errors={field_name: f"有效选项: {', '.join(str(c) for c in choices)}"}
            )

        return value

    # 可执行文件 magic bytes（PE/ELF/Mach-O）
    EXECUTABLE_MAGIC: tuple[bytes, ...] = (
        b"MZ",  # Windows PE
        b"\x7fELF",  # Linux ELF
        b"\xfe\xed\xfa\xce",  # Mach-O 32-bit
        b"\xfe\xed\xfa\xcf",  # Mach-O 64-bit
        b"\xce\xfa\xed\xfe",  # Mach-O 32-bit LE
        b"\xcf\xfa\xed\xfe",  # Mach-O 64-bit LE
    )

    @classmethod
    def validate_uploaded_file(
        cls,
        uploaded_file: Any,
        allowed_extensions: list[str] | None = None,
        max_size_mb: float | None = None,
        max_size_bytes: int | None = None,
        field_name: str = "file",
    ) -> Any:
        """
        验证上传文件的格式和大小
        Args:
            uploaded_file: 上传的文件对象
            allowed_extensions: 允许的扩展名列表，如 [".pdf", ".jpg"]
            max_size_mb: 最大文件大小（MB），None 表示不限制
            max_size_bytes: 最大文件大小（字节），None 表示不限制
            field_name: 字段名（用于错误信息）
        Returns:
            验证通过的文件对象
        Raises:
            ValidationException: 验证失败
        """
        if not uploaded_file:
            raise ValidationException("请选择要上传的文件", errors={field_name: "文件不能为空"})

        if allowed_extensions:
            filename: str = getattr(uploaded_file, "name", "") or ""
            ext = ("." + filename.rsplit(".", 1)[-1].lower()) if "." in filename else ""
            if ext not in allowed_extensions:
                raise ValidationException(
                    f"不支持的文件格式: {ext}",
                    errors={field_name: f"允许的格式: {', '.join(allowed_extensions)}"},
                )

        size: int = getattr(uploaded_file, "size", 0) or 0
        if max_size_bytes is not None and size > max_size_bytes:
            raise ValidationException(
                "文件大小超限",
                errors={field_name: f"文件大小不能超过 {max_size_bytes} 字节"},
            )
        if max_size_mb is not None and size > max_size_mb * 1024 * 1024:
            raise ValidationException(
                "文件大小超限",
                errors={field_name: f"文件大小不能超过 {max_size_mb} MB"},
            )

        # 检测可执行文件 magic bytes
        try:
            header: bytes = uploaded_file.read(8)
            uploaded_file.seek(0)
            if any(header.startswith(magic) for magic in cls.EXECUTABLE_MAGIC):
                raise ValidationException(
                    "不允许上传可执行文件",
                    errors={field_name: "文件内容被识别为可执行文件"},
                )
        except (AttributeError, OSError):
            pass

        return uploaded_file

	"""
	通用数据验证工具模块
	提供常用的验证函数和验证器
	"""

	import re
	from datetime import date, datetime
	from decimal import Decimal, InvalidOperation
	from typing import Any

	from apps.core.exceptions import ValidationException


	class Validators:
	"""通用验证器集合"""

	# 正则表达式模式
	PHONE_PATTERN = re.compile(r"^1[3-9]\d{9}$")
	EMAIL_PATTERN = re.compile(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
	ID_CARD_PATTERN = re.compile(r"^[1-9]\d{5}(18\|19\|20)\d{2}(0[1-9]\|1[0-2])(0[1-9]\|[12]\d\|3[01])\d{3}[\dXx]$")
	SOCIAL_CREDIT_CODE_PATTERN = re.compile(r"^[0-9A-HJ-NPQRTUWXY]{2}\d{6}[0-9A-HJ-NPQRTUWXY]{10}$")

	@classmethod
	def validate_phone(cls, phone: str \| None, field_name: str = "phone") -> str \| None:
	"""
	验证手机号码

	Args:
	phone: 手机号码
	field_name: 字段名（用于错误信息）

	Returns:
	验证后的手机号码

	Raises:
	ValidationException: 验证失败
	"""
	if not phone:
	return None

	phone = phone.strip()
	if not phone:
	return None
	if not cls.PHONE_PATTERN.match(phone):
	raise ValidationException("手机号码格式不正确", errors={field_name: "请输入有效的11位手机号码"})
	return phone

	@classmethod
	def validate_email(cls, email: str \| None, field_name: str = "email") -> str \| None:
	"""验证邮箱地址"""
	if not email:
	return None

	email = email.strip().lower()
	if not cls.EMAIL_PATTERN.match(email):
	raise ValidationException("邮箱格式不正确", errors={field_name: "请输入有效的邮箱地址"})
	return email

	@classmethod
	def validate_id_card(cls, id_card: str \| None, field_name: str = "id_card") -> str \| None:
	"""验证身份证号码"""
	if not id_card:
	return None

	id_card = id_card.strip().upper()
	if not cls.ID_CARD_PATTERN.match(id_card):
	raise ValidationException("身份证号码格式不正确", errors={field_name: "请输入有效的18位身份证号码"})

	# 校验码验证
	if not cls._verify_id_card_checksum(id_card):
	raise ValidationException("身份证号码校验失败", errors={field_name: "身份证号码校验码不正确"})

	return id_card

	@classmethod
	def _verify_id_card_checksum(cls, id_card: str) -> bool:
	"""验证身份证校验码"""
	weights = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
	check_codes = "10X98765432"

	try:
	total = sum(int(id_card[i]) * weights[i] for i in range(17))
	return check_codes[total % 11] == id_card[17]
	except (ValueError, IndexError):
	return False

	@classmethod
	def validate_social_credit_code(cls, code: str \| None, field_name: str = "social_credit_code") -> str \| None:
	"""验证统一社会信用代码"""
	if not code:
	return None

	code = code.strip().upper()
	if not cls.SOCIAL_CREDIT_CODE_PATTERN.match(code):
	raise ValidationException(
	"统一社会信用代码格式不正确", errors={field_name: "请输入有效的18位统一社会信用代码"}
	)
	return code

	@classmethod
	def validate_required(cls, value: Any, field_name: str) -> Any:
	"""验证必填字段"""
	if value is None or (isinstance(value, str) and not value.strip()):
	raise ValidationException(f"{field_name} 不能为空", errors={field_name: "此字段为必填项"})
	return value

	@classmethod
	def validate_length(
	cls,
	value: str \| None,
	field_name: str,
	min_length: int \| None = None,
	max_length: int \| None = None,
	) -> str \| None:
	"""验证字符串长度"""
	if value is None or value == "":
	return None

	length = len(value)

	if min_length is not None and length < min_length:
	raise ValidationException(f"{field_name} 长度不足", errors={field_name: f"最少需要 {min_length} 个字符"})

	if max_length is not None and length > max_length:
	raise ValidationException(f"{field_name} 长度超限", errors={field_name: f"最多允许 {max_length} 个字符"})

	return value

	@classmethod
	def validate_range(
	cls,
	value: float \| None,
	field_name: str,
	min_value: float \| None = None,
	max_value: float \| None = None,
	) -> float \| None:
	"""验证数值范围"""
	if value is None:
	return value

	if min_value is not None and value < min_value:
	raise ValidationException(f"{field_name} 值过小", errors={field_name: f"最小值为 {min_value}"})

	if max_value is not None and value > max_value:
	raise ValidationException(f"{field_name} 值过大", errors={field_name: f"最大值为 {max_value}"})

	return value

	@classmethod
	def validate_decimal(
	cls,
	value: Any,
	field_name: str,
	max_digits: int = 14,
	decimal_places: int = 2,
	) -> Decimal \| None:
	"""验证并转换为 Decimal"""
	if value is None:
	return None

	try:
	decimal_value = Decimal(str(value))
	except (InvalidOperation, ValueError) as e:
	raise ValidationException(f"{field_name} 格式不正确", errors={field_name: "请输入有效的数字"}) from e

	# 检查精度
	sign, digits, exponent = decimal_value.as_tuple()
	total_digits = len(digits)
	decimal_digits = -exponent if isinstance(exponent, int) and exponent < 0 else 0
	integer_digits = total_digits - decimal_digits

	if integer_digits + decimal_places > max_digits:
	raise ValidationException(
	f"{field_name} 数值过大", errors={field_name: f"整数部分最多 {max_digits - decimal_places} 位"}
	)

	if decimal_digits > decimal_places:
	raise ValidationException(
	f"{field_name} 小数位数过多", errors={field_name: f"最多保留 {decimal_places} 位小数"}
	)

	return decimal_value

	@classmethod
	def validate_date(
	cls,
	value: Any,
	field_name: str,
	min_date: date \| None = None,
	max_date: date \| None = None,
	) -> date \| None:
	"""验证日期"""
	if value is None:
	return None

	if isinstance(value, datetime):
	value = value.date()
	elif isinstance(value, str):
	try:
	value = datetime.strptime(value, "%Y-%m-%d").date()
	except ValueError as e:
	raise ValidationException(
	f"{field_name} 日期格式不正确", errors={field_name: "请使用 YYYY-MM-DD 格式"}
	) from e
	elif not isinstance(value, date):
	raise ValidationException(f"{field_name} 类型不正确", errors={field_name: "请提供有效的日期"})

	if min_date and value < min_date:
	raise ValidationException(f"{field_name} 日期过早", errors={field_name: f"日期不能早于 {min_date}"})

	if max_date and value > max_date:
	raise ValidationException(f"{field_name} 日期过晚", errors={field_name: f"日期不能晚于 {max_date}"})

	return value # type: ignore[no-any-return]

	@classmethod
	def validate_in_choices(
	cls,
	value: Any,
	field_name: str,
	choices: list[Any],
	allow_none: bool = True,
	) -> Any:
	"""验证值是否在选项列表中"""
	if value is None:
	if allow_none:
	return None
	raise ValidationException(f"{field_name} 不能为空", errors={field_name: "此字段为必填项"})

	if value not in choices:
	raise ValidationException(
	f"{field_name} 值无效", errors={field_name: f"有效选项: {', '.join(str(c) for c in choices)}"}
	)

	return value

	# 可执行文件 magic bytes（PE/ELF/Mach-O）
	EXECUTABLE_MAGIC: tuple[bytes, ...] = (
	b"MZ", # Windows PE
	b"\x7fELF", # Linux ELF
	b"\xfe\xed\xfa\xce", # Mach-O 32-bit
	b"\xfe\xed\xfa\xcf", # Mach-O 64-bit
	b"\xce\xfa\xed\xfe", # Mach-O 32-bit LE
	b"\xcf\xfa\xed\xfe", # Mach-O 64-bit LE
	)

	@classmethod
	def validate_uploaded_file(
	cls,
	uploaded_file: Any,
	allowed_extensions: list[str] \| None = None,
	max_size_mb: float \| None = None,
	max_size_bytes: int \| None = None,
	field_name: str = "file",
	) -> Any:
	"""
	验证上传文件的格式和大小

	Args:
	uploaded_file: 上传的文件对象
	allowed_extensions: 允许的扩展名列表，如 [".pdf", ".jpg"]
	max_size_mb: 最大文件大小（MB），None 表示不限制
	max_size_bytes: 最大文件大小（字节），None 表示不限制
	field_name: 字段名（用于错误信息）

	Returns:
	验证通过的文件对象

	Raises:
	ValidationException: 验证失败
	"""
	if not uploaded_file:
	raise ValidationException("请选择要上传的文件", errors={field_name: "文件不能为空"})

	if allowed_extensions:
	filename: str = getattr(uploaded_file, "name", "") or ""
	ext = ("." + filename.rsplit(".", 1)[-1].lower()) if "." in filename else ""
	if ext not in allowed_extensions:
	raise ValidationException(
	f"不支持的文件格式: {ext}",
	errors={field_name: f"允许的格式: {', '.join(allowed_extensions)}"},
	)

	size: int = getattr(uploaded_file, "size", 0) or 0
	if max_size_bytes is not None and size > max_size_bytes:
	raise ValidationException(
	"文件大小超限",
	errors={field_name: f"文件大小不能超过 {max_size_bytes} 字节"},
	)
	if max_size_mb is not None and size > max_size_mb * 1024 * 1024:
	raise ValidationException(
	"文件大小超限",
	errors={field_name: f"文件大小不能超过 {max_size_mb} MB"},
	)

	# 检测可执行文件 magic bytes
	try:
	header: bytes = uploaded_file.read(8)
	uploaded_file.seek(0)
	if any(header.startswith(magic) for magic in cls.EXECUTABLE_MAGIC):
	raise ValidationException(
	"不允许上传可执行文件",
	errors={field_name: "文件内容被识别为可执行文件"},
	)
	except (AttributeError, OSError):
	pass

	return uploaded_file