ETest-Vue-FastAPI/ruoyi-fastapi-backend/module_admin/system/service/test_eut_service.py

from fastapi import Request
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List
from config.constant import CommonConstant
from exceptions.exception import ServiceException
from module_admin.entity.vo.common_vo import CrudResponseModel
from module_admin.service.dict_service import DictDataService
from module_admin.system.dao.test_eut_dao import Test_eutDao
from module_admin.system.entity.vo.test_eut_vo import DeleteTest_eutModel, Test_eutModel, Test_eutPageQueryModel
from utils.common_util import CamelCaseUtil
from utils.excel_util import ExcelUtil


class Test_eutService:
    """
    产品管理模块服务层
    """

    @classmethod
    async def get_test_eut_list_services(
        cls, query_db: AsyncSession, query_object: Test_eutPageQueryModel, is_page: bool = False, order_id_from_work_order: int = None
    ):
        """
        获取产品管理列表信息service

        :param query_db: orm对象
        :param query_object: 查询参数对象
        :param is_page: 是否开启分页
        :param order_id_from_work_order: 通过工单的order_id查询样品（用于从工单生成的订单）
        :return: 产品管理列表信息对象
        """
        test_eut_list_result = await Test_eutDao.get_test_eut_list(query_db, query_object, is_page, order_id_from_work_order)

        return test_eut_list_result


    @classmethod
    async def add_test_eut_services(cls, query_db: AsyncSession, page_object: Test_eutModel):
        """
        新增产品管理信息service

        :param query_db: orm对象
        :param page_object: 新增产品管理对象
        :return: 新增产品管理校验结果
        """
        try:
            await Test_eutDao.add_test_eut_dao(query_db, page_object)
            await query_db.commit()
            return CrudResponseModel(is_success=True, message='新增成功')
        except Exception as e:
            await query_db.rollback()
            raise e

    @classmethod
    async def edit_test_eut_services(cls, query_db: AsyncSession, page_object: Test_eutModel):
        """
        编辑产品管理信息service

        :param query_db: orm对象
        :param page_object: 编辑产品管理对象
        :return: 编辑产品管理校验结果
        """
        edit_test_eut = page_object.model_dump(
            exclude_unset=True,
            exclude={'eut_type_name', 'test_flow_name'}
        )
        test_eut_info = await cls.test_eut_detail_services(query_db, page_object.id)
        if test_eut_info.id:
            try:
                await Test_eutDao.edit_test_eut_dao(query_db, edit_test_eut)
                await query_db.commit()
                return CrudResponseModel(is_success=True, message='更新成功')
            except Exception as e:
                await query_db.rollback()
                raise e
        else:
            raise ServiceException(message='产品管理不存在')

    @classmethod
    async def delete_test_eut_services(cls, query_db: AsyncSession, page_object: DeleteTest_eutModel):
        """
        删除产品管理信息service

        :param query_db: orm对象
        :param page_object: 删除产品管理对象
        :return: 删除产品管理校验结果
        """
        if page_object.ids:
            id_list = page_object.ids.split(',')
            try:
                for id in id_list:
                    await Test_eutDao.delete_test_eut_dao(query_db, Test_eutModel(id=id))
                await query_db.commit()
                return CrudResponseModel(is_success=True, message='删除成功')
            except Exception as e:
                await query_db.rollback()
                raise e
        else:
            raise ServiceException(message='传入产品ID为空')

    @classmethod
    async def test_eut_detail_services(cls, query_db: AsyncSession, id: int):
        """
        获取产品管理详细信息service

        :param query_db: orm对象
        :param id: 产品ID
        :return: 产品ID对应的信息
        """
        test_eut = await Test_eutDao.get_test_eut_detail_by_id(query_db, id=id)
        if test_eut:
            result = Test_eutModel(**CamelCaseUtil.transform_result(test_eut))
        else:
            result = Test_eutModel(**dict())

        return result

    @classmethod
    async def get_test_flow_list_services(cls, query_db: AsyncSession):
        """
        获取测试流程列表service
        
        :param query_db: orm对象
        :return: 测试流程列表
        """
        return await Test_eutDao.get_test_flow_list(query_db)
        
    @classmethod
    async def get_eut_type_list_services(cls, query_db: AsyncSession):
        """
        获取产品类别列表service
        
        :param query_db: orm对象
        :return: 产品类别列表
        """
        return await Test_eutDao.get_eut_type_list(query_db)
        
    @staticmethod
    async def export_test_eut_list_services(request: Request, test_eut_list: List):
        """
        导出产品管理信息service

        :param test_eut_list: 产品管理信息列表
        :return: 产品管理信息对应excel的二进制数据
        """
        # 创建一个映射字典，将英文键映射到中文键
        mapping_dict = {
            'id': '产品ID',
            'testOrderId': '订单ID',
            'testFlowName': '流程名称',
            'eutTypeName': '产品类别名称',
            'sn': 'sn号',
            'batchNo': '批次号',
            'version': '版本号',
            'sampleAppearance': '外观',
            'memo': '备注说明',
        }
        sys_sample_appearance_list = await DictDataService.query_dict_data_list_from_cache_services(
            request.app.state.redis, dict_type='sys_sample_appearance'
        )
        sys_sample_appearance_option = [dict(label=item.get('dictLabel'), value=item.get('dictValue')) for item in sys_sample_appearance_list]
        sys_sample_appearance_option_dict = {item.get('value'): item for item in sys_sample_appearance_option}
        for item in test_eut_list:
            if str(item.get('sampleAppearance')) in sys_sample_appearance_option_dict.keys():
                item['sampleAppearance'] = sys_sample_appearance_option_dict.get(str(item.get('sampleAppearance'))).get('label')
        binary_data = ExcelUtil.export_list2excel(test_eut_list, mapping_dict)

        return binary_data