vent_agent_2.0/extensions/main_fan/batch_models.py

"""月度判识的矿级计划与报告档案契约；业务归属和算法由 Agent 判断。"""
from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, Field, model_validator


class HistoryPoint(BaseModel):
    model_config = ConfigDict(extra='forbid', str_strip_whitespace=True)
    system: Literal['safety', 'equipment']
    node_id: str = Field(min_length=1, max_length=128, pattern=r'^[A-Za-z0-9_-]+$')
    role: Literal['fan_state', 'flow']
    fan_id: str = ''
    label: str = ''
    unit: str = ''
    mapping_confirmed: bool = False
    mapping_reason: str = ''


class ShaftPlan(BaseModel):
    model_config = ConfigDict(extra='forbid', str_strip_whitespace=True)
    shaft_id: str = Field(min_length=1, max_length=80, pattern=r'^[A-Za-z0-9_-]+$')
    shaft_name: str = Field(min_length=1, max_length=160)
    identity_status: Literal['confirmed', 'candidate'] = 'candidate'
    identity_reason: str = Field(min_length=1, max_length=2000)
    fan_pair: list[str] = Field(default_factory=list, max_length=2)
    points: list[HistoryPoint] = Field(default_factory=list, max_length=32)
    limitations: list[str] = Field(default_factory=list, max_length=30)


class MinePlan(BaseModel):
    model_config = ConfigDict(extra='forbid', str_strip_whitespace=True)
    coverage: Literal['confirmed', 'partial'] = 'partial'
    shafts: list[ShaftPlan] = Field(min_length=1, max_length=20)
    limitations: list[str] = Field(default_factory=list, max_length=30)

    @model_validator(mode='after')
    def unique_shafts(self):
        if len({shaft.shaft_id for shaft in self.shafts}) != len(self.shafts):
            raise ValueError('同矿通风井任务标识不能重复')
        return self


class ReportArchiveFacts(BaseModel):
    """只约束完整读取声明和来源，具体提取事实结构由 Agent 设计。"""
    model_config = ConfigDict(extra='forbid')
    source_sha256: str = Field(pattern=r'^[0-9a-f]{64}$')
    page_count: int = Field(ge=1)
    read_pages: list[int] = Field(min_length=1)
    page_sources: dict[int, str] = Field(default_factory=dict, description='每个物理页实际读取的独立图片或文本文件路径')
    reports: list[dict[str, Any]] = Field(min_length=1)
    analysis: dict[str, Any] = Field(default_factory=dict)
    limitations: list[str] = Field(default_factory=list)


class BatchRequest(BaseModel):
    model_config = ConfigDict(extra='forbid')
    # 必须显式指定矿井；CLI的联调保护默认最多两个，无隐式全量执行。
    dept_ids: list[str] = Field(default_factory=list, max_length=500)
    all_producing_mines: bool = False
    reassess: bool = Field(default=False, description='在原批次和原时段内重新分析已完成结果，保留旧版本并复用已取得原件')
    snapshot_id: str | None = None
    cutoff: str | None = None
    lookback_days: int = Field(default=40, ge=1, le=40)
    user_name: str = Field(default='main-fan-batch', min_length=1, max_length=128)

    @model_validator(mode='after')
    def explicit_selection(self):
        if self.all_producing_mines == bool(self.dept_ids):
            raise ValueError('必须显式选择部门列表或全部正常生产煤矿，两种方式只能选一种')
        return self