hd-crane-gs/backend/app/api/v1/module_business/vardict/schema.py

# -*- coding: utf-8 -*-

from pydantic import BaseModel, ConfigDict, Field
from fastapi import Query

from app.core.validator import DateTimeStr
from app.core.base_schema import BaseSchema, UserBySchema

class BizVarDictCreateSchema(BaseModel):
    """
    变量信息新增模型
    """
    crane_no: str = Field(default=..., description='行车编号')
    var_code: str = Field(default=..., description='变量code')
    var_name: str = Field(default=..., description='变量名称')
    mec_type: int = Field(default=..., description='所属机构')
    data_type: int = Field(default=..., description='数据类型')
    switch_type: int | None = Field(default=None, description='变量类型')
    addr: str = Field(default=..., description='采集地址')
    gateway_id: int = Field(default=..., description='网关id')
    var_sort: int | None = Field(default=None, description='排序')
    var_group: int | None = Field(default=None, description='变量分组')
    var_category: int = Field(default=..., description='变量分类')
    translate: str | None = Field(default=None, description='绑定公式')
    device_no: str | None = Field(default=None, description='关联设备编号 ')
    is_reverse: int | None = Field(default=None, description='是否取反')
    is_top_show: int = Field(default=..., description='是否重点显示')
    is_save: int = Field(default=..., description='是否生成')
    is_calibration: int | None = Field(default=None, description='是否标定')
    is_overview_top_show: int = Field(default=..., description='是否首页重点显示')
    is_home_page_show: int = Field(default=..., description='是否首页显示')
    is_diagnose: int | None = Field(default=None, description='是否启用诊断专家')
    is_upload: int = Field(default=..., description='是否上传云平台')
    diagnosis_id: str | None = Field(default=None, description='关联诊断专家')
    status: str = Field(default="1", description='是否启用')
    description: str | None = Field(default=None, max_length=255, description='备注/描述')


class BizVarDictUpdateSchema(BizVarDictCreateSchema):
    """
    变量信息更新模型
    """
    ...


class BizVarDictOutSchema(BizVarDictCreateSchema, BaseSchema, UserBySchema):
    """
    变量信息响应模型
    """
    model_config = ConfigDict(from_attributes=True)


class BizVarDictQueryParam:
    """变量信息查询参数"""

    def __init__(
        self,
        var_code: str | None = Query(None, description="变量code"),
        var_name: str | None = Query(None, description="变量名称"),
        addr: str | None = Query(None, description="modbus地址"),
        device_no: str | None = Query(None, description="关联设备编号 "),
        diagnosis_id: str | None = Query(None, description="关联诊断专家"),
        status: str | None = Query(None, description="是否启用"),
        crane_no: str | None = Query(None, description=""),
        mec_type: int | None = Query(None, description="所属机构"),
        data_type: int | None = Query(None, description="数据类型"),
        switch_type: int | None = Query(None, description="变量类型"),
        gateway_id: int | None = Query(None, description="网关"),
        var_sort: int | None = Query(None, description="排序"),
        var_group: int | None = Query(None, description="变量分组"),
        var_category: int | None = Query(None, description="变量分类"),
        translate: str | None = Query(None, description="绑定公式"),
        is_reverse: int | None = Query(None, description="是否取反"),
        is_top_show: int | None = Query(None, description="是否重点显示"),
        is_save: int | None = Query(None, description="是否生成"),
        is_calibration: int | None = Query(None, description="是否标定"),
        is_overview_top_show: int | None = Query(None, description="是否首页重点显示"),
        is_home_page_show: int | None = Query(None, description="是否首页显示"),
        is_diagnose: int | None = Query(None, description="是否启用诊断专家"),
        is_upload: int | None = Query(None, description="是否上传云平台"),
        created_id: int | None = Query(None, description="创建人ID"),
        updated_id: int | None = Query(None, description="更新人ID"),
        created_time: list[DateTimeStr] | None = Query(None, description="创建时间范围", examples=["2025-01-01 00:00:00", "2025-12-31 23:59:59"]),
        updated_time: list[DateTimeStr] | None = Query(None, description="更新时间范围", examples=["2025-01-01 00:00:00", "2025-12-31 23:59:59"]),
        # 新增：场景区分参数，默认 False（代表后台内部请求），True 代表前端接口请求
        is_api_request: bool | None = Query(False, description="是否为前端接口请求（用于切换 switch_type 查询方式）")

    ) -> None:
        
        # 精确查询字段
        self.crane_no = crane_no
        # 模糊查询字段
        self.var_code = ("like", var_code)
        # 模糊查询字段
        self.var_name = ("like", var_name)
        # 精确查询字段
        self.mec_type = mec_type
        # 精确查询字段
        self.data_type = data_type
        if is_api_request:
            self.switch_type = (">=", switch_type)
        else:
            self.switch_type = switch_type
        # 模糊查询字段
        self.addr = ("like", addr)
        # 精确查询字段
        self.gateway_id = gateway_id
        # 精确查询字段
        self.var_sort = var_sort
        # 精确查询字段
        self.var_group = var_group
        # 精确查询字段
        self.var_category = var_category
        # 精确查询字段
        self.translate = translate
        # 模糊查询字段
        self.device_no = ("like", device_no)
        # 精确查询字段
        self.is_reverse = is_reverse
        # 精确查询字段
        self.is_top_show = is_top_show
        # 精确查询字段
        self.is_save = is_save
        # 精确查询字段
        self.is_calibration = is_calibration
        # 精确查询字段
        self.is_overview_top_show = is_overview_top_show
        # 精确查询字段
        self.is_home_page_show = is_home_page_show
        # 精确查询字段
        self.is_diagnose = is_diagnose
        # 精确查询字段
        self.is_upload = is_upload
        # 模糊查询字段
        self.diagnosis_id = ("like", diagnosis_id)
        # 模糊查询字段
        self.status = ("like", status)
        # 精确查询字段
        self.created_id = created_id
        # 精确查询字段
        self.updated_id = updated_id
        # 时间范围查询
        if created_time and len(created_time) == 2:
            self.created_time = ("between", (created_time[0], created_time[1]))
        if updated_time and len(updated_time) == 2:
            self.updated_time = ("between", (updated_time[0], updated_time[1]))

class VarDictMecGroupSchema(BaseModel):
    """
    行车信息页面数据模型
    """
    mec_type: str = Field(default=..., description='所属机构')
    alarm_varList: list[BizVarDictOutSchema] | None = Field(default=None, description='报警变量数据')
    digital_varList: list[BizVarDictOutSchema] | None = Field(default=None, description='开关量变量数据')
    analog_varList: list[BizVarDictOutSchema] | None = Field(default=None, description='模拟量变量数据')
    varList_simple: list[BizVarDictOutSchema] | None = Field(default=None, description='变量数据')