"""结果接收平台的 JSON 契约;与取数网关及内部审计结构分开。""" from datetime import datetime from typing import Literal from urllib.parse import urlsplit from pydantic import BaseModel, ConfigDict, Field, field_validator ResultCode = Literal["NORMAL", "WARNING", "ABNORMAL", "UNDETERMINED"] class PlatformJudgment(BaseModel): model_config = ConfigDict(extra="forbid", strict=True) resultCode: ResultCode result: str = Field(min_length=1) reason: str = Field(min_length=1) evidenceLink: str = "" @field_validator("evidenceLink") @classmethod def evidence_url(cls, value): if value: parts = urlsplit(value) if parts.scheme not in {"http", "https"} or not parts.netloc or parts.username or parts.password: raise ValueError("evidenceLink 只能填写可访问的 HTTP(S) 证据页地址或空字符串") return value @field_validator("reason") @classmethod def report_text(cls, value): if "data:image/" in value or ";base64," in value: raise ValueError("报告正文不能包含图片编码") return value class PlatformResultPayload(BaseModel): model_config = ConfigDict(extra="forbid", strict=True) agentCode: str = Field(min_length=1, max_length=64) agentName: str = Field(min_length=1) taskId: str = Field(pattern=r"^[0-9a-f]{32}$") checkedAt: str mineCode: str = Field(min_length=1) mineName: str = Field(min_length=1) objectDetail: str = Field(min_length=1) dataDetail: str judgment: PlatformJudgment @field_validator("checkedAt") @classmethod def completion_time(cls, value): parsed = datetime.strptime(value, "%Y-%m-%d %H:%M:%S") if parsed.strftime("%Y-%m-%d %H:%M:%S") != value: raise ValueError("checkedAt 必须使用 yyyy-MM-dd HH:mm:ss(Asia/Shanghai)") return value @field_validator("agentCode", "agentName", "mineCode", "mineName", "objectDetail") @classmethod def nonblank(cls, value): if not value.strip(): raise ValueError("平台身份及研判对象不能为空白") return value