File size: 4,051 Bytes
b4f2238 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | """Validation helpers for reminder services."""
from __future__ import annotations
import json
from datetime import datetime
from typing import TYPE_CHECKING, Any
from django.utils import timezone
from django.utils.translation import gettext_lazy as _
from apps.core.exceptions import ValidationException
from ..models import ReminderType
if TYPE_CHECKING:
from django.utils.functional import Promise
from ..ports import CaseLogTargetQueryPort, CaseTargetQueryPort, ContractTargetQueryPort
def normalize_target_id(value: int | None, *, field_name: str | Promise) -> int | None:
if value is None:
return None
if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
raise ValidationException(_("%(field_name)s 必须为正整数") % {"field_name": field_name})
return value
def validate_positive_id(value: int, *, field_name: str | Promise) -> None:
"""校验 ID 为正整数（非 bool）。"""
if isinstance(value, bool) or value <= 0:
raise ValidationException(_("%(field_name)s 必须为正整数") % {"field_name": field_name})
def validate_binding_exclusive(
*,
contract_id: int | None,
case_id: int | None,
case_log_id: int | None,
) -> None:
bound_count = sum(target_id is not None for target_id in (contract_id, case_id, case_log_id))
if bound_count > 1:
raise ValidationException(_("合同、案件、案件日志最多只能绑定一个"))
def validate_fk_exists(
*,
contract_id: int | None,
case_id: int | None,
case_log_id: int | None,
contract_target_query: ContractTargetQueryPort | None = None,
case_target_query: CaseTargetQueryPort | None = None,
case_log_target_query: CaseLogTargetQueryPort | None = None,
) -> None:
"""校验外键引用的记录是否存在。"""
if contract_id is not None:
if contract_target_query is None:
raise RuntimeError("ContractTargetQueryPort not provided")
if not contract_target_query.exists(contract_id):
raise ValidationException(_("合同 %(id)s 不存在") % {"id": contract_id})
if case_id is not None:
if case_target_query is None:
raise RuntimeError("CaseTargetQueryPort not provided")
if not case_target_query.exists(case_id):
raise ValidationException(_("案件 %(id)s 不存在") % {"id": case_id})
if case_log_id is not None:
if case_log_target_query is None:
raise RuntimeError("CaseLogTargetQueryPort not provided")
if not case_log_target_query.exists(case_log_id):
raise ValidationException(_("案件日志 %(id)s 不存在") % {"id": case_log_id})
def normalize_reminder_type(reminder_type: str) -> str:
value = str(reminder_type).strip()
if not value:
raise ValidationException(_("提醒类型不能为空"))
if value not in ReminderType.values:
raise ValidationException(_("无效的提醒类型"))
return value
# 与 Reminder.content max_length=255 保持同步
_CONTENT_MAX_LENGTH = 255
def normalize_content(content: str) -> str:
value = str(content).strip()
if not value:
raise ValidationException(_("提醒事项不能为空"))
if len(value) > _CONTENT_MAX_LENGTH:
raise ValidationException(_("提醒事项不能超过 %(max)d 个字符") % {"max": _CONTENT_MAX_LENGTH})
return value
def normalize_due_at(due_at: datetime) -> datetime:
if not isinstance(due_at, datetime):
raise ValidationException(_("到期时间格式不正确"))
if timezone.is_naive(due_at):
return timezone.make_aware(due_at)
return due_at
def normalize_metadata(metadata: Any) -> dict[str, Any]:
if metadata is None:
return {}
if not isinstance(metadata, dict):
raise ValidationException(_("扩展数据必须为 JSON 对象"))
try:
json.dumps(metadata)
except (TypeError, ValueError):
raise ValidationException(_("扩展数据包含不可序列化的值"))
return metadata
|