hd-crane-gs/backend/app/api/v1/module_business/mec/service.py

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

import io
from fastapi import UploadFile
import pandas as pd
from redis.asyncio import Redis

from app.common.enums import RedisInitKeyConfig
from app.core.base_schema import BatchSetAvailable
from app.core.exceptions import CustomException
from app.core.redis_crud import RedisCURD
from app.utils.excel_util import ExcelUtil
from app.core.logger import log
from app.api.v1.module_system.auth.schema import AuthSchema
from .schema import BizMecCreateSchema, BizMecUpdateSchema, BizMecOutSchema, BizMecQueryParam
from .crud import BizMecCRUD
from ..crane.crud import BizCraneCRUD
from ..crane.model import BizCraneModel


class BizMecService:
    """
    机构信息服务层
    """
    
    @classmethod
    async def detail_mec_service(cls, auth: AuthSchema, id: int) -> dict:
        """详情"""
        obj = await BizMecCRUD(auth).get_by_id_mec_crud(id=id)
        if not obj:
            raise CustomException(msg="该数据不存在")
        crane = await BizCraneCRUD(auth).get_by_id_crane_crud_for_no(obj.crane_no)
        res = BizMecOutSchema.model_validate(obj).model_dump()
        res['crane_name'] = crane.crane_name
        return res

    @classmethod
    async def detail_mec_service_for_no(cls, auth: AuthSchema, mec_no: str) -> dict:
        """详情"""
        obj = await BizMecCRUD(auth).get_by_id_mec_crud_for_no(mec_no=mec_no)
        if not obj:
            raise CustomException(msg="该数据不存在")
        return BizMecOutSchema.model_validate(obj).model_dump()
    
    @classmethod
    async def list_mec_service(cls, auth: AuthSchema, search: BizMecQueryParam | None = None, order_by: list[dict] | None = None) -> list[dict]:
        """列表查询"""
        search_dict = search.__dict__ if search else None
        obj_list = await BizMecCRUD(auth).list_mec_crud(search=search_dict, order_by=order_by)
        return [BizMecOutSchema.model_validate(obj).model_dump() for obj in obj_list]

    @classmethod
    async def page_mec_service(cls, auth: AuthSchema, page_no: int, page_size: int, search: BizMecQueryParam | None = None, order_by: list[dict] | None = None) -> dict:
        """分页查询（数据库分页）"""
        search_dict = search.__dict__ if search else {}
        order_by_list = order_by or [{'id': 'asc'}]
        offset = (page_no - 1) * page_size
        result = await BizMecCRUD(auth).page_mec_crud(
            offset=offset,
            limit=page_size,
            order_by=order_by_list,
            search=search_dict
        )
        for item in result.get('items'):
            crane_model: BizCraneModel | None = await BizCraneCRUD(auth).get_by_id_crane_crud_for_no(crane_no=item['crane_no'])
            item['crane_name'] = crane_model.crane_name
        return result
    
    @classmethod
    async def create_mec_service(cls, auth: AuthSchema, data: BizMecCreateSchema,redis: Redis) -> dict:
        """创建"""
        # 检查唯一性约束
        obj = await BizMecCRUD(auth).create_mec_crud(data=data)
        if obj:
            # 更新缓存中数据
            await RedisCURD(redis).clear(f"{RedisInitKeyConfig.VAR_DICT.key}:*")
        return BizMecOutSchema.model_validate(obj).model_dump()
    
    @classmethod
    async def update_mec_service(cls, auth: AuthSchema, id: int, data: BizMecUpdateSchema,redis: Redis) -> dict:
        """更新"""
        # 检查数据是否存在
        obj = await BizMecCRUD(auth).get_by_id_mec_crud(id=id)
        if not obj:
            raise CustomException(msg='更新失败，该数据不存在')
        
        # 检查唯一性约束
            
        obj = await BizMecCRUD(auth).update_mec_crud(id=id, data=data)
        if obj:
            # 更新缓存中数据
            await RedisCURD(redis).clear(f"{RedisInitKeyConfig.VAR_DICT.key}:*")
        return BizMecOutSchema.model_validate(obj).model_dump()
    
    @classmethod
    async def delete_mec_service(cls, auth: AuthSchema, ids: list[int],redis: Redis) -> None:
        """删除"""
        if len(ids) < 1:
            raise CustomException(msg='删除失败，删除对象不能为空')
        for id in ids:
            obj = await BizMecCRUD(auth).get_by_id_mec_crud(id=id)
            if not obj:
                raise CustomException(msg=f'删除失败，ID为{id}的数据不存在')
        await BizMecCRUD(auth).delete_mec_crud(ids=ids)
        # 更新缓存中数据
        await RedisCURD(redis).clear(f"{RedisInitKeyConfig.VAR_DICT.key}:*")
    
    @classmethod
    async def set_available_mec_service(cls, auth: AuthSchema, data: BatchSetAvailable,redis: Redis) -> None:
        """批量设置状态"""
        await BizMecCRUD(auth).set_available_mec_crud(ids=data.ids, status=data.status)
        # 更新缓存中数据
        await RedisCURD(redis).clear(f"{RedisInitKeyConfig.VAR_DICT.key}:*")
    
    @classmethod
    async def batch_export_mec_service(cls, obj_list: list[dict]) -> bytes:
        """批量导出"""
        mapping_dict = {
            'id': '编号 ',
            'crane_no': '起重机编号 ',
            'mec_no': '机构编号 ',
            'mec_category': '机构分类 MecCategory',
            'mec_type': '机构类型 MecType',
            'hoist_weight': '起升重量 起升机构',
            'hoist_height': '起升高度 起升机构',
            'hoist_speed': '起升速度 起升机构',
            'rope_count': '钢线绳受力根数 起升机构',
            'rope_max_pull': '钢丝绳最大拉力 起升机构',
            'rope_diameter': '钢丝绳直径 起升机构',
            'rope_model': '钢丝绳型号 起升机构',
            'car_speed': '速度 平移机构',
            'wheel_span': '轮距 平移机构',
            'track_span': '轨距 平移机构',
            'wheel_pressure': '轮压 平移机构',
            'track_model': '轨道型号 平移机构',
            'track_diameter': '车轨直径 平移机构',
            'track_count': '车轨数量 平移机构',
            'word_system': '工作制度 ',
            'reducer_model': '减速机型号 ',
            'reducer_trans_ratio': '减速机传动比 ',
            'reducer_total_trans_ratio': '减速机总传动比 ',
            'motor_model': '电机型号 ',
            'motor_voltage': '电机额定电压 ',
            'motor_current': '电机额定电流 ',
            'motor_power': '电机额定功率 ',
            'motor_speed': '电机转速 ',
            'motor_count': '电机数量 ',
            'bradk_model': '制动器型号 ',
            'brake_torque': '制动器制动力矩 ',
            'brake_wheel_diameter': '制动器轮径 ',
            'brake_count': '制动器数量 ',
            'sort': '排序 ',
            'uuid': 'UUID全局唯一标识',
            'status': '是否启用(0:启用 1:禁用)',
            'description': '备注/描述',
            'created_time': '创建时间',
            'updated_time': '更新时间',
            'hoist_weight_var_code': '重量关联变量',
            'stroke_var_code': '行程高度变量',
            'is_canvas_show': '是否首页动画展示',
            'is_canvas_move': '是否启用动画',
            'gear_select_var_codes': '机构选择变量',
            'vib_gear_value': '振动分析挡位',
            'updated_id': '更新者ID',
        }

        data = obj_list.copy()
        for item in data:
            # 状态转换
            if 'status' in item:
                item['status'] = '启用' if item.get('status') == '0' else '停用'
            # 创建者转换
            creator_info = item.get('creator')
            if isinstance(creator_info, dict):
                item['creator'] = creator_info.get('name', '未知')
            elif creator_info is None:
                item['creator'] = '未知'

        return ExcelUtil.export_list2excel(list_data=data, mapping_dict=mapping_dict)

    @classmethod
    async def batch_import_mec_service(cls, auth: AuthSchema, file: UploadFile, update_support: bool = False) -> str:
        """批量导入"""
        header_dict = {
            '编号 ': 'id',
            '起重机编号 ': 'crane_no',
            '机构编号 ': 'mec_no',
            '机构分类 MecCategory': 'mec_category',
            '机构类型 MecType': 'mec_type',
            '起升重量 起升机构': 'hoist_weight',
            '起升高度 起升机构': 'hoist_height',
            '起升速度 起升机构': 'hoist_speed',
            '钢线绳受力根数 起升机构': 'rope_count',
            '钢丝绳最大拉力 起升机构': 'rope_max_pull',
            '钢丝绳直径 起升机构': 'rope_diameter',
            '钢丝绳型号 起升机构': 'rope_model',
            '速度 平移机构': 'car_speed',
            '轮距 平移机构': 'wheel_span',
            '轨距 平移机构': 'track_span',
            '轮压 平移机构': 'wheel_pressure',
            '轨道型号 平移机构': 'track_model',
            '车轨直径 平移机构': 'track_diameter',
            '车轨数量 平移机构': 'track_count',
            '工作制度 ': 'word_system',
            '减速机型号 ': 'reducer_model',
            '减速机传动比 ': 'reducer_trans_ratio',
            '减速机总传动比 ': 'reducer_total_trans_ratio',
            '电机型号 ': 'motor_model',
            '电机额定电压 ': 'motor_voltage',
            '电机额定电流 ': 'motor_current',
            '电机额定功率 ': 'motor_power',
            '电机转速 ': 'motor_speed',
            '电机数量 ': 'motor_count',
            '制动器型号 ': 'bradk_model',
            '制动器制动力矩 ': 'brake_torque',
            '制动器轮径 ': 'brake_wheel_diameter',
            '制动器数量 ': 'brake_count',
            '排序 ': 'sort',
            'UUID全局唯一标识': 'uuid',
            '是否启用(0:启用 1:禁用)': 'status',
            '备注/描述': 'description',
            '创建时间': 'created_time',
            '更新时间': 'updated_time',
            '重量关联变量': 'hoist_weight_var_code',
            '行程高度变量': 'stroke_var_code',
            '是否首页动画展示': 'is_canvas_show',
            '是否启用动画': 'is_canvas_move',
            '机构选择变量': 'gear_select_var_codes',
            '振动分析挡位': 'vib_gear_value',
        }

        try:
            contents = await file.read()
            df = pd.read_excel(io.BytesIO(contents))
            await file.close()
            
            if df.empty:
                raise CustomException(msg="导入文件为空")
            
            missing_headers = [header for header in header_dict.keys() if header not in df.columns]
            if missing_headers:
                raise CustomException(msg=f"导入文件缺少必要的列: {', '.join(missing_headers)}")
            
            df.rename(columns=header_dict, inplace=True)
            
            # 验证必填字段
            
            error_msgs = []
            success_count = 0
            count = 0
            
            for index, row in df.iterrows():
                count += 1
                try:
                    data = {
                        "id": row['id'],
                        "crane_no": row['crane_no'],
                        "mec_no": row['mec_no'],
                        "mec_category": row['mec_category'],
                        "mec_type": row['mec_type'],
                        "hoist_weight": row['hoist_weight'],
                        "hoist_height": row['hoist_height'],
                        "hoist_speed": row['hoist_speed'],
                        "rope_count": row['rope_count'],
                        "rope_max_pull": row['rope_max_pull'],
                        "rope_diameter": row['rope_diameter'],
                        "rope_model": row['rope_model'],
                        "car_speed": row['car_speed'],
                        "wheel_span": row['wheel_span'],
                        "track_span": row['track_span'],
                        "wheel_pressure": row['wheel_pressure'],
                        "track_model": row['track_model'],
                        "track_diameter": row['track_diameter'],
                        "track_count": row['track_count'],
                        "word_system": row['word_system'],
                        "reducer_model": row['reducer_model'],
                        "reducer_trans_ratio": row['reducer_trans_ratio'],
                        "reducer_total_trans_ratio": row['reducer_total_trans_ratio'],
                        "motor_model": row['motor_model'],
                        "motor_voltage": row['motor_voltage'],
                        "motor_current": row['motor_current'],
                        "motor_power": row['motor_power'],
                        "motor_speed": row['motor_speed'],
                        "motor_count": row['motor_count'],
                        "bradk_model": row['bradk_model'],
                        "brake_torque": row['brake_torque'],
                        "brake_wheel_diameter": row['brake_wheel_diameter'],
                        "brake_count": row['brake_count'],
                        "sort": row['sort'],
                        "uuid": row['uuid'],
                        "status": row['status'],
                        "description": row['description'],
                        "created_time": row['created_time'],
                        "updated_time": row['updated_time'],
                        "hoist_weight_var_code": row['hoist_weight_var_code'],
                        "stroke_var_code": row['stroke_var_code'],
                        "is_canvas_show": row['is_canvas_show'],
                        "is_canvas_move": row['is_canvas_move'],
                        "gear_select_var_codes": row['gear_select_var_codes'],
                        "vib_gear_value": row['vib_gear_value'],
                    }
                    # 使用CreateSchema做校验后入库
                    create_schema = BizMecCreateSchema.model_validate(data)
                    
                    # 检查唯一性约束
                    
                    await BizMecCRUD(auth).create_mec_crud(data=create_schema)
                    success_count += 1
                except Exception as e:
                    error_msgs.append(f"第{count}行: {str(e)}")
                    continue

            result = f"成功导入 {success_count} 条数据"
            if error_msgs:
                result += "\n错误信息:\n" + "\n".join(error_msgs)
            return result
            
        except Exception as e:
            log.error(f"批量导入失败: {str(e)}")
            raise CustomException(msg=f"导入失败: {str(e)}")
    
    @classmethod
    async def import_template_download_mec_service(cls) -> bytes:
        """下载导入模板"""
        header_list = [
            '编号 ',
            '起重机编号 ',
            '机构编号 ',
            '机构分类 MecCategory',
            '机构类型 MecType',
            '起升重量 起升机构',
            '起升高度 起升机构',
            '起升速度 起升机构',
            '钢线绳受力根数 起升机构',
            '钢丝绳最大拉力 起升机构',
            '钢丝绳直径 起升机构',
            '钢丝绳型号 起升机构',
            '速度 平移机构',
            '轮距 平移机构',
            '轨距 平移机构',
            '轮压 平移机构',
            '轨道型号 平移机构',
            '车轨直径 平移机构',
            '车轨数量 平移机构',
            '工作制度 ',
            '减速机型号 ',
            '减速机传动比 ',
            '减速机总传动比 ',
            '电机型号 ',
            '电机额定电压 ',
            '电机额定电流 ',
            '电机额定功率 ',
            '电机转速 ',
            '电机数量 ',
            '制动器型号 ',
            '制动器制动力矩 ',
            '制动器轮径 ',
            '制动器数量 ',
            '排序 ',
            'UUID全局唯一标识',
            '是否启用(0:启用 1:禁用)',
            '备注/描述',
            '创建时间',
            '更新时间',
            '重量关联变量',
            '行程高度变量',
            '是否首页动画展示',
            '是否启用动画',
            '机构选择变量',
            '振动分析挡位',
        ]
        selector_header_list = []
        option_list = []
        
        # 添加下拉选项
        
        return ExcelUtil.get_excel_template(
            header_list=header_list,
            selector_header_list=selector_header_list,
            option_list=option_list
        )