vent_agent_2.0/extensions/main_fan/schemas.py

"""主通风机筛查输入、审计结果与平台公共结果。"""
from typing import Any, Literal

from pydantic import AliasChoices, BaseModel, ConfigDict, Field, field_validator, model_validator

from algorithms.main_fan.models import Screening, ScreeningInput

from runtime.business_models import JudgmentResult
from runtime.result_models import ResultCode


class AgentEvidence(BaseModel):
    """Agent 的来源引用；宿主只绑定来源，不代替模型判断原文含义。"""
    model_config = ConfigDict(extra="forbid", strict=True)
    source: str = Field(min_length=1, description="本轮原件、input.json/data.json 或实时/时序查询工具返回的可信文件虚拟路径")
    page: int | None = Field(default=None, ge=1, description="原件物理页码；JSON 数据引用不用填")
    pointer: str = Field(default="", description="原始 JSON 中的字段路径或 JSON Pointer")
    excerpt: str = Field(default="", description="实际读取的原文或数据说明")

    @field_validator('source', mode='before')
    @classmethod
    def workspace_reference(cls, value):
        # 代码中的相对工作区路径与工具虚拟路径指向同一个来源，不必让模型为前缀重新计算。
        # 宿主仍会与本轮来源清单精确匹配，不能借此引用其他文件或任务。
        if isinstance(value, str) and value and not value.startswith('/'):
            return '/workspace/' + value.removeprefix('./')
        return value


class AgentAssessment(BaseModel):
    """沙箱脚本交付契约；只约束交付格式，不规定算法和中间事实结构。"""
    model_config = ConfigDict(extra="forbid", strict=True)
    status: Literal["completed", "needs_materials"] = "completed"
    result_code: ResultCode
    summary: str = Field(min_length=1, max_length=4000)
    data_detail: str = Field(default="", max_length=12000)
    missing_materials: list[str] = Field(default_factory=list, max_length=50)
    evidence: list[AgentEvidence] = Field(default_factory=list, max_length=200)
    details: dict[str, Any] = Field(default_factory=dict)

    @model_validator(mode="after")
    def check_delivery(self):
        if not self.summary.strip():
            raise ValueError("研判摘要不能为空白")
        if self.result_code != "UNDETERMINED" and not self.evidence:
            raise ValueError("确定结论必须引用本轮原件或原始数据")
        if self.status == "needs_materials" and self.result_code != "UNDETERMINED":
            raise ValueError("缺少关键材料时应提交 UNDETERMINED")
        return self


class MainFanInput(ScreeningInput):
    target: str = Field(default="", max_length=256, description="可仅传目标名称或配置中的风机标识，例如众源煤业1号主通风机")
    mine_code: str = Field(default="", min_length=1, validation_alias=AliasChoices("mine_code","mineCode"), description="可由服务端按target解析")
    mine_name: str = Field(default="", min_length=1, validation_alias=AliasChoices("mine_name","mineName"), description="可由服务端按target解析")
    fan_name: str = ""
    dept_id: str = Field(default='', max_length=64, description='批次绑定的真实部门编号；取数工具校验同矿范围')
    task_description: str = "矿井主通风机研判"
    materials: list[dict[str, Any]] = Field(default_factory=list, max_length=20)
    measurements: dict[str, Any] = Field(default_factory=dict)
    use_gateway: bool = False
    document_cache_policy: Literal["auto", "refresh", "off"] = "auto"
    paired_assessment: bool = True

    @model_validator(mode="before")
    @classmethod
    def resolve_business_target(cls, value):
        from extensions.main_fan.targets import resolve_request
        return resolve_request(value)


class MainFanResult(JudgmentResult):
    status: Literal["completed", "needs_materials"]
    judgment_scope: Literal["preliminary_screening", "main_fan_consistency"] = "main_fan_consistency"
    analysis_id: str = ""
    analysis: Screening | None = None
    result_format: Literal["report"] | None = None
    sources: list[dict[str, str]] = Field(default_factory=list)
    assessment: dict[str, Any] | None = None
    workflow_id: str = ""

    @model_validator(mode="after")
    def screening_receipt(self):
        if self.status == "completed" and not self.analysis_id:
            raise ValueError("已完成筛查必须包含实际工具生成的 analysis_id；完整原件由结果验证器恢复")
        return self