"""统一业务事实。单位、来源和人工确认均显式记录。""" from datetime import date, datetime from typing import Literal from pydantic import AliasChoices, BaseModel, ConfigDict, Field, field_validator class FactModel(BaseModel): model_config = ConfigDict(extra="forbid", populate_by_name=True, allow_inf_nan=False, str_strip_whitespace=True) class Evidence(FactModel): source: str = "" page: int | None = Field(default=None, ge=1) label: str = "" excerpt: str = "" sha256: str = "" method: Literal["api", "manual_transcription", "model_extraction", "text_extraction", "synthetic"] = "api" class Device(FactModel): device_code: str = Field(min_length=1, validation_alias=AliasChoices("device_code", "deviceCode")) name: str = Field(min_length=1, validation_alias=AliasChoices("name", "deviceName")) mine_code: str = Field(default="", validation_alias=AliasChoices("mine_code", "mineCode")) mine_name: str = Field(default="", validation_alias=AliasChoices("mine_name", "mineName")) location: str = "" state: Literal["running", "stopped", "unknown"] = "unknown" fan_role: Literal["in_service", "standby", "upgrading", "replacing", "unknown"] = "unknown" availability_status: Literal["available", "unavailable", "unknown"] = "unknown" observed_at: datetime | None = None evidence: Evidence = Field(default_factory=Evidence) @field_validator("state", mode="before") @classmethod def state_label(cls, value): return {"开": "running", "停": "stopped"}.get(value, value) Metric = Literal["velocity", "flow", "static_pressure", "pressure", "shaft_power", "input_power", "temperature", "current", "voltage", "speed", "frequency", "unknown"] class MonitorPoint(FactModel): point_code: str = Field(min_length=1, validation_alias=AliasChoices("point_code", "pointCode")) name: str = Field(min_length=1, validation_alias=AliasChoices("name", "pointName")) mine_code: str = Field(default="", validation_alias=AliasChoices("mine_code", "mineCode")) mine_name: str = Field(default="", validation_alias=AliasChoices("mine_name", "mineName")) device_name: str = "" location: str = "" metric: Metric = "unknown" value: float | None = None unit: str = "" status: str = "unknown" observed_at: datetime | None = None evidence: Evidence = Field(default_factory=Evidence) class ConfirmedBinding(FactModel): point_code: str = Field(min_length=1) fan_id: str = Field(min_length=1) confirmed_by: str = Field(min_length=1) reason: str = Field(min_length=1) class PerformancePoint(FactModel): label: str = "" frequency_hz: float | None = Field(default=None, gt=0) blade_angle_deg: float | None = None # 现场实测与折算值不可混用；每条记录都是同一行的量。 condition: Literal["measured", "converted"] = "measured" flow_m3_s: float = Field(ge=0) static_pressure_pa: float = Field(ge=0) shaft_power_kw: float = Field(gt=0) input_power_kw: float | None = Field(default=None, gt=0) reported_efficiency_pct: float | None = Field(default=None, ge=0, le=100) evidence: Evidence = Field(default_factory=Evidence) class PerformanceReport(FactModel): report_id: str = Field(min_length=1) mine_name: str = Field(min_length=1) mine_code: str = "" fan_name: str = Field(min_length=1) serial_number: str = "" model: str = "" motor_model: str = "" rated_power_raw: str = Field(default="", description="铭牌额定功率原表达式，如2×185 kW；不以运行功率代替") rated_power_kw: float | None = Field(default=None, gt=0, description="仅直接载明的总额定功率；组合表达式由工具计算") rated_speed_rpm: float | None = Field(default=None, gt=0) rated_voltage_raw: str = "" manufacturer: str = "" impeller_diameter_m: float | None = Field(default=None, gt=0) ventilation_mode: str = "" structure_type: str = "" rated_flow_range_m3_min: list[float] = Field(default_factory=list, max_length=2) rated_pressure_range_pa: list[float] = Field(default_factory=list, max_length=2) blade_angle_deg: float | None = Field(default=None, description="测试叶片安装角；原表/或空白表示未载明，不填0") inspected_on: date | None = None recommended_next_on: date | None = None original_conclusion: str = "" basis: list[str] = Field(default_factory=list) points: list[PerformancePoint] = Field(default_factory=list, max_length=250) curves: list[PerformancePoint] = Field(default_factory=list, max_length=30, description="只录曲线图注印出的工况数字，不从曲线像素猜读，和数据表分开保存") metadata_evidence: dict[str, Evidence] = Field(default_factory=dict, description="每个身份/日期字段的原文与物理页码，日期须含检测/建议下次标签") evidence: Evidence = Field(default_factory=Evidence) class DocumentFacts(FactModel): reports: list[PerformanceReport] = Field(default_factory=list) devices: list[Device] = Field(default_factory=list) points: list[MonitorPoint] = Field(default_factory=list) warnings: list[str] = Field(default_factory=list) extraction_audit: dict = Field(default_factory=dict, description="仅由服务端填写，记录路由、缓存和模型调用") class SeriesSample(FactModel): observed_at: datetime value: float status: str = "normal" class OperatingSeries(FactModel): mine_code: str = "" mine_name: str = "" fan_id: str = Field(min_length=1) metric: Metric unit: str sensor_id: str = Field(min_length=1) mapping_confirmed: bool = False samples: list[SeriesSample] = Field(default_factory=list, max_length=200000) evidence: Evidence = Field(default_factory=Evidence) class SwitchRecord(FactModel): mine_code: str = "" mine_name: str = "" event_id: str = Field(min_length=1) switched_at: datetime from_fan_id: str = Field(min_length=1) to_fan_id: str = Field(min_length=1) from_stopped_at: datetime | None = None to_started_at: datetime | None = None reason: str = "unknown" evidence: Evidence = Field(default_factory=Evidence) class FanStateRecord(FactModel): mine_code: str = "" mine_name: str = "" fan_id: str = Field(min_length=1) observed_at: datetime state: Literal["running", "stopped", "unknown"] complete_fan_state: bool = False frequency_hz: float | None = Field(default=None, gt=0) blade_angle_deg: float | None = None event_type: Literal["state", "start_stop", "frequency_change", "angle_change", "fault"] = "state" evidence: Evidence = Field(default_factory=Evidence) class ScreeningInput(FactModel): mine_code: str = Field(min_length=1, validation_alias=AliasChoices("mine_code", "mineCode")) mine_name: str = Field(min_length=1, validation_alias=AliasChoices("mine_name", "mineName")) fan_id: str = Field(default="", validation_alias=AliasChoices("fan_id", "fanId")) fan_name: str = "" fan_pair: list[str] = Field(default_factory=list, max_length=2, description="两台风机的明确标识；不表示谁主用谁备用") operating_series: list[OperatingSeries] = Field(default_factory=list, max_length=200) switch_records: list[SwitchRecord] = Field(default_factory=list, max_length=1000) fan_state_records: list[FanStateRecord] = Field(default_factory=list, max_length=200000) evaluation_time: datetime | None = None devices: list[Device] = Field(default_factory=list, max_length=500) points: list[MonitorPoint] = Field(default_factory=list, max_length=3000) reports: list[PerformanceReport] = Field(default_factory=list, max_length=50) confirmed_bindings: list[ConfirmedBinding] = Field(default_factory=list) class Finding(FactModel): code: str severity: Literal["info", "review", "blocking"] message: str evidence: list[Evidence] = Field(default_factory=list) class Association(FactModel): point_code: str candidate_fan: str state: Literal["candidate", "confirmed", "unmatched", "ambiguous", "excluded"] confidence: Literal["none", "low", "medium", "confirmed"] basis: list[str] eligible_for_calculation: bool = False class Calculation(FactModel): scope: Literal["historical_report", "current_confirmed"] label: str formula: str = "eta_static = Q[m3/s] * p_static[Pa] / (1000 * P_shaft[kW]) * 100%" values: dict[str, float] efficiency_pct: float reported_efficiency_pct: float | None = None difference_percentage_points: float | None = None evidence: list[Evidence] = Field(default_factory=list) class Screening(FactModel): scope: Literal["preliminary_screening"] = "preliminary_screening" engineering_status: Literal["insufficient_data", "review_required"] has_in_scope_data: bool findings: list[Finding] associations: list[Association] calculations: list[Calculation] missing_materials: list[str] document_audits: list[dict] = Field(default_factory=list)