# -*- coding: utf-8 -*- import io import json from fastapi import UploadFile import pandas as pd from redis.asyncio.client import Redis from app.core.database import async_db_session from app.core.redis_crud import RedisCURD from app.common.enums import RedisInitKeyConfig from app.core.base_schema import BatchSetAvailable from app.core.exceptions import CustomException 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 BizVarDictCreateSchema, BizVarDictUpdateSchema, BizVarDictOutSchema, BizVarDictQueryParam from ..crane.crud import BizCraneCRUD from ..crane.model import BizCraneModel from ..gateway.crud import GatewayCRUD from ..gateway.model import GatewayModel from ..mec.crud import BizMecCRUD from ..vardict.crud import BizVarDictCRUD from ..vardict.schema import VarDictMecGroupSchema class BizVarDictService: """ 变量信息服务层 """ @classmethod async def detail_vardict_service(cls, auth: AuthSchema, id: int) -> dict: """详情""" obj = await BizVarDictCRUD(auth).get_by_id_vardict_crud(id=id) if not obj: raise CustomException(msg="该数据不存在") crane = await BizCraneCRUD(auth).get_by_id_crane_crud_for_no(obj.crane_no) gateway = await GatewayCRUD(auth).get_by_id_gateway_crud(obj.gateway_id) res = BizVarDictOutSchema.model_validate(obj).model_dump() res['crane_name'] = crane.crane_name res['gateway_name'] = gateway.gateway_name if gateway else "" return res @classmethod async def list_vardict_service(cls, auth: AuthSchema, search: BizVarDictQueryParam | None = None, order_by: list[dict] | None = None) -> list[dict]: """列表查询""" search_dict = search.__dict__ if search else None obj_list = await BizVarDictCRUD(auth).list_vardict_crud(search=search_dict, order_by=order_by) return [BizVarDictOutSchema.model_validate(obj).model_dump() for obj in obj_list] @classmethod async def page_vardict_service(cls, auth: AuthSchema, page_no: int, page_size: int, search: BizVarDictQueryParam | 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 BizVarDictCRUD(auth).page_vardict_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']) gateway_model: GatewayModel | None = await GatewayCRUD(auth).get_by_id_gateway_crud(id=item['gateway_id']) item['crane_name'] = crane_model.crane_name item['gateway_name'] = gateway_model.gateway_name if gateway_model else "" return result @classmethod async def create_vardict_service(cls, auth: AuthSchema, data: BizVarDictCreateSchema,redis: Redis) -> dict: """创建""" # 检查唯一性约束 obj = await BizVarDictCRUD(auth).create_vardict_crud(data=data) if obj: # 更新缓存中数据 await RedisCURD(redis).clear(f"{RedisInitKeyConfig.VAR_DICT.key}:*") return BizVarDictOutSchema.model_validate(obj).model_dump() @classmethod async def update_vardict_service(cls, auth: AuthSchema, id: int, data: BizVarDictUpdateSchema,redis: Redis) -> dict: """更新""" # 检查数据是否存在 obj = await BizVarDictCRUD(auth).get_by_id_vardict_crud(id=id) if not obj: raise CustomException(msg='更新失败，该数据不存在') # 检查唯一性约束 obj = await BizVarDictCRUD(auth).update_vardict_crud(id=id, data=data) if obj: # 更新缓存中数据 await RedisCURD(redis).clear(f"{RedisInitKeyConfig.VAR_DICT.key}:*") return BizVarDictOutSchema.model_validate(obj).model_dump() @classmethod async def delete_vardict_service(cls, auth: AuthSchema, ids: list[int],redis: Redis) -> None: """删除""" if len(ids) < 1: raise CustomException(msg='删除失败，删除对象不能为空') for id in ids: obj = await BizVarDictCRUD(auth).get_by_id_vardict_crud(id=id) if not obj: raise CustomException(msg=f'删除失败，ID为{id}的数据不存在') await BizVarDictCRUD(auth).delete_vardict_crud(ids=ids) # 更新缓存中数据 await RedisCURD(redis).clear(f"{RedisInitKeyConfig.VAR_DICT.key}:*") @classmethod async def set_available_vardict_service(cls, auth: AuthSchema, data: BatchSetAvailable,redis: Redis) -> None: """批量设置状态""" await BizVarDictCRUD(auth).set_available_vardict_crud(ids=data.ids, status=data.status) # 更新缓存中数据 await RedisCURD(redis).clear(f"{RedisInitKeyConfig.VAR_DICT.key}:*") @classmethod async def batch_export_vardict_service(cls, obj_list: list[dict]) -> bytes: """批量导出""" mapping_dict = { 'id': 'id ', 'crane_no': '', 'var_code': '变量code', 'var_name': '变量名称', 'mec_type': '所属机构', 'data_type': '数据类型', 'switch_type': '变量类型', 'addr': 'modbus地址', 'gateway_id': '网关', 'var_sort': '排序', 'var_group': '变量分组', 'var_category': '变量分类', 'translate': '绑定公式', 'device_no': '关联设备编号 ', 'is_reverse': '是否取反', 'is_top_show': '是否重点显示', 'is_save': '是否生成', 'is_calibration': '是否标定', 'is_overview_top_show': '是否首页重点显示', 'is_home_page_show': '是否首页显示', 'is_diagnose': '是否启用诊断专家', 'is_upload': '是否上传云平台', 'diagnosis_id': '关联诊断专家', 'status': '是否启用', 'description': '备注/描述', 'create_time': '创建时间 ', 'updated_time': '更新时间', 'created_id': '创建人ID', '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_vardict_service(cls, auth: AuthSchema, file: UploadFile, update_support: bool = False) -> str: """批量导入""" header_dict = { 'id ': 'id', '': 'crane_no', '变量code': 'var_code', '变量名称': 'var_name', '所属机构': 'mec_type', '数据类型': 'data_type', '变量类型': 'switch_type', 'modbus地址': 'addr', '网关': 'gateway_id', '排序': 'var_sort', '变量分组': 'var_group', '变量分类': 'var_category', '绑定公式': 'translate', '关联设备编号 ': 'device_no', '是否取反': 'is_reverse', '是否重点显示': 'is_top_show', '是否生成': 'is_save', '是否标定': 'is_calibration', '是否首页重点显示': 'is_overview_top_show', '是否首页显示': 'is_home_page_show', '是否启用诊断专家': 'is_diagnose', '是否上传云平台': 'is_upload', '关联诊断专家': 'diagnosis_id', '是否启用': 'status', '备注/描述': 'description', '创建时间 ': 'create_time', '更新时间': 'updated_time', '创建人ID': 'created_id', '更新人ID': 'updated_id', } 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'], "var_code": row['var_code'], "var_name": row['var_name'], "mec_type": row['mec_type'], "data_type": row['data_type'], "switch_type": row['switch_type'], "addr": row['addr'], "gateway_id": row['gateway_id'], "var_sort": row['var_sort'], "var_group": row['var_group'], "var_category": row['var_category'], "translate": row['translate'], "device_no": row['device_no'], "is_reverse": row['is_reverse'], "is_top_show": row['is_top_show'], "is_save": row['is_save'], "is_calibration": row['is_calibration'], "is_overview_top_show": row['is_overview_top_show'], "is_home_page_show": row['is_home_page_show'], "is_diagnose": row['is_diagnose'], "is_upload": row['is_upload'], "diagnosis_id": row['diagnosis_id'], "status": row['status'], "description": row['description'], "create_time": row['create_time'], "updated_time": row['updated_time'], "created_id": row['created_id'], "updated_id": row['updated_id'], } # 使用CreateSchema做校验后入库 create_schema = BizVarDictCreateSchema.model_validate(data) # 检查唯一性约束 await BizVarDictCRUD(auth).create_vardict_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_vardict_service(cls) -> bytes: """下载导入模板""" header_list = [ 'id ', '', '变量code', '变量名称', '所属机构', '数据类型', '变量类型', 'modbus地址', '网关', '排序', '变量分组', '变量分类', '绑定公式', '关联设备编号 ', '是否取反', '是否重点显示', '是否生成', '是否标定', '是否首页重点显示', '是否首页显示', '是否启用诊断专家', '是否上传云平台', '关联诊断专家', '是否启用', '备注/描述', '创建时间 ', '更新时间', '创建人ID', '更新人ID', ] selector_header_list = [] option_list = [] # 添加下拉选项 selector_header_list.append('所属机构') option_list.append({'所属机构': []}) selector_header_list.append('数据类型') option_list.append({'数据类型': []}) selector_header_list.append('变量类型') option_list.append({'变量类型': []}) selector_header_list.append('变量分类') option_list.append({'变量分类': []}) return ExcelUtil.get_excel_template( header_list=header_list, selector_header_list=selector_header_list, option_list=option_list ) @classmethod async def get_vardict_group_service(cls,auth: AuthSchema, redis: Redis,id: int): """ 从缓存获取变量分组数据列表信息service 参数: - redis (Redis): Redis客户端 - id (int): 行车id 返回: - list[dict]: 变量分组数据列表 """ try: crane = await BizCraneCRUD(auth).get_by_id_crane_crud(id) redis_key = f"{RedisInitKeyConfig.VAR_DICT.key}:{crane.crane_no}" obj_list_dict = await RedisCURD(redis).get(redis_key) # 确保返回数据正确序列化 if obj_list_dict: if isinstance(obj_list_dict, str): try: return json.loads(obj_list_dict) except json.JSONDecodeError: log.warning(f"变量分组数据反序列化失败，尝试重新初始化缓存: {crane.crane_name}") elif isinstance(obj_list_dict, list): return obj_list_dict # 缓存不存在或格式错误时重新初始化 await cls.init_vardict_group_service(redis) obj_list_dict = await RedisCURD(redis).get(redis_key) if not obj_list_dict: raise CustomException(msg="变量分组数据不存在") # 再次确保返回数据正确序列化 if isinstance(obj_list_dict, str): try: return json.loads(obj_list_dict) except json.JSONDecodeError: raise CustomException(msg="变量分组数据格式错误") return obj_list_dict except CustomException: raise except Exception as e: log.error(f"获取变量分组数据缓存失败: {str(e)}") raise CustomException(msg=f"获取变量分组数据失败: {str(e)}") @classmethod async def init_vardict_group_service(cls, redis: Redis): """ 应用初始化: 获取所有天车变量数据信息并缓存service 参数: - redis (Redis): Redis客户端 返回: - None """ try: async with async_db_session() as session: async with session.begin(): # 在初始化过程中，不需要检查数据权限 auth = AuthSchema(db=session, check_data_scope=False) crane_list = await BizCraneCRUD(auth).list(search={'status':'1'},order_by=[{'order':'asc'}]) if not crane_list: log.warning("未找到任何天车数据") return success_count = 0 fail_count = 0 for crane in crane_list: crane_no = crane.crane_no crane_name = crane.crane_name try: varDictMecGroupSchema: list[VarDictMecGroupSchema] = [] mec_list = await BizMecCRUD(auth).list(search={'crane_no':crane_no,'status':'1'},order_by=[{'sort':'asc'}]) for mec in mec_list: # 获取分组数据 varDicts = await BizVarDictCRUD(auth).list(search={'crane_no':crane_no,'mec_type':mec.mec_type,'status':'1'},order_by=[{'var_sort':'asc'}]) if not varDicts: continue varDictMecGroupSchema.append( VarDictMecGroupSchema(mec_type=mec.mec_type, varList_simple=varDicts)) # 保存到Redis并设置过期时间 redis_key = f"{RedisInitKeyConfig.VAR_DICT.key}:{crane_no}" var_dict_list = [item.model_dump() for item in varDictMecGroupSchema] value = json.dumps(var_dict_list, ensure_ascii=False) await RedisCURD(redis).set( key=redis_key, value=value, ) success_count += 1 log.info(f"✅ 变量数据缓存成功: {crane_name}") except Exception as e: fail_count += 1 log.error(f"❌ 初始化变量数据失败 [{crane_name}]: {e}") log.info(f"变量数据初始化完成 - 成功: {success_count}, 失败: {fail_count}") except Exception as e: log.error(f"变量数据初始化过程发生错误: {e}") # 只在严重错误时抛出异常，允许单个字典加载失败 raise CustomException(msg=f"变量数据初始化失败: {str(e)}")