vent_agent_2.0/runtime/business_models.py

"""业务输入附件及平台结果的公共契约，不绑定特定研判业务。"""
from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, Field, model_validator
from runtime.result_models import PlatformResultPayload


class RemoteFile(BaseModel):
    model_config = ConfigDict(extra="forbid")
    file_id: str = Field(min_length=1, max_length=128)
    version: str = Field(min_length=1, max_length=128)
    filename: str = Field(min_length=1, max_length=255)
    path: str = Field(min_length=1, description="文件系统内相对接口路径，以 / 开头")
    params: dict[str, Any] = Field(default_factory=dict)


class Clue(BaseModel):
    model_config = ConfigDict(extra="forbid", populate_by_name=True)
    clue_type: Literal["text", "table", "image", "file", "video", "curve"] = Field(alias="clueType")
    clue_name: str = Field(default="", alias="clueName")
    clue_content: str = Field(default="", alias="clueContent")


class ResultArtifact(BaseModel):
    model_config = ConfigDict(extra="forbid")
    filename: str = Field(min_length=1, max_length=255)
    path: str = Field(pattern=r"^/workspace/", max_length=1024,
                      description="本次会话中由业务工具生成的文件路径")


class JudgmentResult(BaseModel):
    model_config = ConfigDict(extra="forbid")
    status: Literal["completed", "needs_materials", "not_implemented"]
    mine_code: str = ""
    mine_name: str = ""
    check_object: str = Field(min_length=1)
    result_title: str = Field(min_length=1)
    result: str = Field(min_length=1)
    check_process: str = ""
    alarm_level: Literal["红", "橙", "黄", "蓝"] | None = None
    missing_materials: list[str] = Field(default_factory=list)
    clues: list[Clue] = Field(default_factory=list)
    artifacts: list[ResultArtifact] = Field(default_factory=list, max_length=30)
    platform_payload: PlatformResultPayload | None = None

    @model_validator(mode="after")
    def completed_identity(self):
        if any(not getattr(self, key).strip() for key in ("check_object", "result_title", "result")):
            raise ValueError("研判对象、标题和结论不能仅为空白")
        if self.status == "completed" and (not self.mine_code.strip() or not self.mine_name.strip()):
            raise ValueError("正式结果必须具有矿井编码和名称")
        return self


def platform_result(value: dict[str, Any]) -> JudgmentResult:
    """保留业务专有结果供审计，平台映射只取公共契约字段。"""
    return JudgmentResult.model_validate({key: item for key, item in value.items()
                                         if key in JudgmentResult.model_fields})