204 lines
6.6 KiB
Python
204 lines
6.6 KiB
Python
"""风控只读查询参数。"""
|
|
|
|
from datetime import date, datetime
|
|
from email.utils import parseaddr
|
|
from typing import Literal
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator
|
|
|
|
RiskLevel = Literal["低", "中", "高"]
|
|
BehaviorLevel = Literal["normal", "slight", "attention", "high", "immediate"]
|
|
RiskEvidenceSource = Literal[
|
|
"customers",
|
|
"products",
|
|
"transactions",
|
|
"capital_flows",
|
|
"holdings",
|
|
"login_records",
|
|
"alerts",
|
|
"notifications",
|
|
]
|
|
|
|
|
|
class RiskAlertPageQuery(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
keyword: str | None = Field(default=None, max_length=128)
|
|
customer_no: str | None = Field(default=None, max_length=32)
|
|
product_code: str | None = Field(default=None, max_length=32)
|
|
product_name: str | None = Field(default=None, max_length=128)
|
|
risk_level: RiskLevel | None = None
|
|
rule_code: str | None = Field(default=None, pattern=r"^RW-[0-9]{3}$")
|
|
start_time: datetime | None = None
|
|
end_time: datetime | None = None
|
|
cursor: str | None = Field(default=None, max_length=512)
|
|
limit: int = Field(default=5, ge=1, le=5)
|
|
|
|
@field_validator("keyword", "customer_no", "product_code", "product_name", "rule_code")
|
|
@classmethod
|
|
def text_must_not_be_blank(cls, value: str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
value = value.strip()
|
|
if not value:
|
|
raise ValueError("筛选条件不能为空")
|
|
return value
|
|
|
|
@field_validator("end_time")
|
|
@classmethod
|
|
def end_time_must_not_precede_start(
|
|
cls,
|
|
value: datetime | None,
|
|
info: ValidationInfo,
|
|
) -> datetime | None:
|
|
start_time = info.data.get("start_time")
|
|
if value is not None and start_time is not None and value < start_time:
|
|
raise ValueError("结束时间不能早于开始时间")
|
|
return value
|
|
|
|
class RiskEvidencePageQuery(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
keyword: str | None = Field(default=None, max_length=128)
|
|
behavior_level: BehaviorLevel | None = None
|
|
send_status: str | None = Field(default=None, max_length=32)
|
|
start_time: datetime | None = None
|
|
end_time: datetime | None = None
|
|
cursor: str | None = Field(default=None, max_length=512)
|
|
limit: int = Field(default=10, ge=1, le=10)
|
|
|
|
@field_validator("keyword", "send_status")
|
|
@classmethod
|
|
def text_must_not_be_blank(cls, value: str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
value = value.strip()
|
|
if not value:
|
|
raise ValueError("筛选条件不能为空")
|
|
return value
|
|
|
|
@field_validator("end_time")
|
|
@classmethod
|
|
def end_time_must_not_precede_start(
|
|
cls,
|
|
value: datetime | None,
|
|
info: ValidationInfo,
|
|
) -> datetime | None:
|
|
start_time = info.data.get("start_time")
|
|
if value is not None and start_time is not None and value < start_time:
|
|
raise ValueError("结束时间不能早于开始时间")
|
|
return value
|
|
|
|
|
|
class RiskNotificationPageQuery(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
keyword: str | None = Field(default=None, max_length=128)
|
|
send_status: str | None = Field(default=None, max_length=32)
|
|
start_time: datetime | None = None
|
|
end_time: datetime | None = None
|
|
cursor: str | None = Field(default=None, max_length=512)
|
|
limit: int = Field(default=10, ge=1, le=10)
|
|
|
|
@field_validator("keyword", "send_status")
|
|
@classmethod
|
|
def text_must_not_be_blank(cls, value: str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
value = value.strip()
|
|
if not value:
|
|
raise ValueError("筛选条件不能为空")
|
|
return value
|
|
|
|
@field_validator("end_time")
|
|
@classmethod
|
|
def end_time_must_not_precede_start(
|
|
cls,
|
|
value: datetime | None,
|
|
info: ValidationInfo,
|
|
) -> datetime | None:
|
|
start_time = info.data.get("start_time")
|
|
if value is not None and start_time is not None and value < start_time:
|
|
raise ValueError("结束时间不能早于开始时间")
|
|
return value
|
|
|
|
|
|
class RiskAlertExclusionRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
reason: str = Field(min_length=1, max_length=500)
|
|
|
|
@field_validator("reason")
|
|
@classmethod
|
|
def reason_must_not_be_blank(cls, value: str) -> str:
|
|
value = value.strip()
|
|
if not value:
|
|
raise ValueError("误报理由不能为空")
|
|
return value
|
|
|
|
|
|
class RiskAlertResolutionRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
resolution: str = Field(min_length=1, max_length=500)
|
|
|
|
@field_validator("resolution")
|
|
@classmethod
|
|
def resolution_must_not_be_blank(cls, value: str) -> str:
|
|
value = value.strip()
|
|
if not value:
|
|
raise ValueError("处置结论不能为空")
|
|
return value
|
|
|
|
|
|
class RiskAlertEscalationRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
reason: str = Field(min_length=1, max_length=500)
|
|
|
|
@field_validator("reason")
|
|
@classmethod
|
|
def reason_must_not_be_blank(cls, value: str) -> str:
|
|
value = value.strip()
|
|
if not value:
|
|
raise ValueError("升级理由不能为空")
|
|
return value
|
|
|
|
|
|
class RiskDailyReportGenerateRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
report_date: date | None = None
|
|
|
|
|
|
class RiskDailyReportMailRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
recipients: list[str] = Field(min_length=1, max_length=10)
|
|
subject: str = Field(min_length=1, max_length=128)
|
|
content: str = Field(min_length=1, max_length=20000)
|
|
|
|
@field_validator("recipients")
|
|
@classmethod
|
|
def recipients_must_be_valid(cls, values: list[str]) -> list[str]:
|
|
result: list[str] = []
|
|
seen: set[str] = set()
|
|
for value in values:
|
|
address = value.strip()
|
|
_, parsed = parseaddr(address)
|
|
if not address or parsed != address or "@" not in address or len(address) > 254:
|
|
raise ValueError("收件邮箱格式无效")
|
|
normalized = address.lower()
|
|
if normalized not in seen:
|
|
result.append(address)
|
|
seen.add(normalized)
|
|
return result
|
|
|
|
@field_validator("subject", "content")
|
|
@classmethod
|
|
def text_must_not_be_blank(cls, value: str) -> str:
|
|
value = value.strip()
|
|
if not value:
|
|
raise ValueError("邮件主题和正文不能为空")
|
|
return value
|