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

# -*- coding: utf-8 -*-
import httpx
from fastapi import APIRouter, Depends, UploadFile, Body, Path, Query
from fastapi.responses import StreamingResponse, JSONResponse
from redis.asyncio.client import Redis
from app.common.response import SuccessResponse, StreamResponse
from app.config.setting import settings
from app.core.dependencies import AuthPermission, redis_getter
from app.api.v1.module_system.auth.schema import AuthSchema
from app.core.base_params import PaginationQueryParam
from app.utils.common_util import bytes2file_response
from app.core.logger import log
from app.core.base_schema import BatchSetAvailable

from .service import BizVarDictService
from .schema import BizVarDictCreateSchema, BizVarDictUpdateSchema, BizVarDictQueryParam

BizVarDictRouter = APIRouter(prefix='/vardict', tags=["变量信息模块"]) 

@BizVarDictRouter.get("/detail/{id}", summary="获取变量信息详情", description="获取变量信息详情")
async def get_vardict_detail_controller(
    id: int = Path(..., description="ID"),
    auth: AuthSchema = Depends(AuthPermission(["module_business:vardict:query"]))
) -> JSONResponse:
    """获取变量信息详情接口"""
    result_dict = await BizVarDictService.detail_vardict_service(auth=auth, id=id)
    log.info(f"获取变量信息详情成功 {id}")
    return SuccessResponse(data=result_dict, msg="获取变量信息详情成功")

@BizVarDictRouter.get("/list", summary="查询变量信息列表", description="查询变量信息列表")
async def get_vardict_list_controller(
    page: PaginationQueryParam = Depends(),
    search: BizVarDictQueryParam = Depends(),
    auth: AuthSchema = Depends(AuthPermission(["module_business:vardict:query"]))
) -> JSONResponse:
    """查询变量信息列表接口（数据库分页）"""
    result_dict = await BizVarDictService.page_vardict_service(
        auth=auth,
        page_no=page.page_no if page.page_no is not None else 1,
        page_size=page.page_size if page.page_size is not None else 10,
        search=search,
        order_by=page.order_by
    )
    log.info("查询变量信息列表成功")
    return SuccessResponse(data=result_dict, msg="查询变量信息列表成功")

@BizVarDictRouter.post("/create", summary="创建变量信息", description="创建变量信息")
async def create_vardict_controller(
    data: BizVarDictCreateSchema,
    redis: Redis = Depends(redis_getter),
    auth: AuthSchema = Depends(AuthPermission(["module_business:vardict:create"]))
) -> JSONResponse:
    """创建变量信息接口"""
    result_dict = await BizVarDictService.create_vardict_service(auth=auth, data=data,redis=redis)
    log.info("创建变量信息成功")
    return SuccessResponse(data=result_dict, msg="创建变量信息成功")

@BizVarDictRouter.put("/update/{id}", summary="修改变量信息", description="修改变量信息")
async def update_vardict_controller(
    data: BizVarDictUpdateSchema,
    id: int = Path(..., description="ID"),
    redis: Redis = Depends(redis_getter),
    auth: AuthSchema = Depends(AuthPermission(["module_business:vardict:update"]))
) -> JSONResponse:
    """修改变量信息接口"""
    result_dict = await BizVarDictService.update_vardict_service(auth=auth, id=id, data=data,redis=redis)
    log.info("修改变量信息成功")
    return SuccessResponse(data=result_dict, msg="修改变量信息成功")

@BizVarDictRouter.delete("/delete", summary="删除变量信息", description="删除变量信息")
async def delete_vardict_controller(
    ids: list[int] = Body(..., description="ID列表"),
    redis: Redis = Depends(redis_getter),
    auth: AuthSchema = Depends(AuthPermission(["module_business:vardict:delete"]))
) -> JSONResponse:
    """删除变量信息接口"""
    await BizVarDictService.delete_vardict_service(auth=auth, ids=ids,redis=redis)
    log.info(f"删除变量信息成功: {ids}")
    return SuccessResponse(msg="删除变量信息成功")

@BizVarDictRouter.patch("/available/setting", summary="批量修改变量信息状态", description="批量修改变量信息状态")
async def batch_set_available_vardict_controller(
    data: BatchSetAvailable,
    auth: AuthSchema = Depends(AuthPermission(["module_business:vardict:patch"]))
) -> JSONResponse:
    """批量修改变量信息状态接口"""
    await BizVarDictService.set_available_vardict_service(auth=auth, data=data)
    log.info(f"批量修改变量信息状态成功: {data.ids}")
    return SuccessResponse(msg="批量修改变量信息状态成功")

@BizVarDictRouter.post('/export', summary="导出变量信息", description="导出变量信息")
async def export_vardict_list_controller(
    search: BizVarDictQueryParam = Depends(),
    auth: AuthSchema = Depends(AuthPermission(["module_business:vardict:export"]))
) -> StreamingResponse:
    """导出变量信息接口"""
    result_dict_list = await BizVarDictService.list_vardict_service(search=search, auth=auth)
    export_result = await BizVarDictService.batch_export_vardict_service(obj_list=result_dict_list)
    log.info('导出变量信息成功')

    return StreamResponse(
        data=bytes2file_response(export_result),
        media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
        headers={
            'Content-Disposition': 'attachment; filename=biz_var_dict.xlsx'
        }
    )

@BizVarDictRouter.post('/import', summary="导入变量信息", description="导入变量信息")
async def import_vardict_list_controller(
    file: UploadFile,
    auth: AuthSchema = Depends(AuthPermission(["module_business:vardict:import"]))
) -> JSONResponse:
    """导入变量信息接口"""
    batch_import_result = await BizVarDictService.batch_import_vardict_service(file=file, auth=auth, update_support=True)
    log.info("导入变量信息成功")
    
    return SuccessResponse(data=batch_import_result, msg="导入变量信息成功")

@BizVarDictRouter.post('/download/template', summary="获取变量信息导入模板", description="获取变量信息导入模板", dependencies=[Depends(AuthPermission(["module_business:vardict:download"]))])
async def export_vardict_template_controller() -> StreamingResponse:
    """获取变量信息导入模板接口"""
    import_template_result = await BizVarDictService.import_template_download_vardict_service()
    log.info('获取变量信息导入模板成功')

    return StreamResponse(
        data=bytes2file_response(import_template_result),
        media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
        headers={'Content-Disposition': 'attachment; filename=biz_var_dict_template.xlsx'}
    )

@BizVarDictRouter.get("/list_alarms", summary="查询变量信息列表", description="查询变量信息列表")
async def get_vardict_list_alarms_controller(
    search: BizVarDictQueryParam = Depends(),
    auth: AuthSchema = Depends(AuthPermission(["module_business:vardict:query"]))
) -> JSONResponse:
    """查询变量信息列表接口（数据库分页）"""
    result_dict = await BizVarDictService.vardict_alarms_list(auth=auth,crane_no=search.crane_no)
    #请求采集接口获取状态信息
    async with httpx.AsyncClient() as client:
        response = await client.get(
            url=settings.COLLECT_DATA_FULL,
            params={},
            timeout=2
        )
        if response.status_code == 200:
            json_data = response.json()
            if json_data['code'] == 200 and json_data['data']:
                for item in result_dict:
                    item['value'] = False
                    crane_no = item['crane_no']
                    alarm = json_data.get('data').get(crane_no).get('data').get('alarm').get(item['var_code'])
                    if alarm:
                        item['value'] = alarm.get('value')
    log.info("查询变量信息列表成功")
    return SuccessResponse(data=result_dict, msg="查询变量信息列表成功")

@BizVarDictRouter.get("/list_analog", summary="查询变量信息列表", description="查询变量信息列表")
async def get_vardict_list_analog_controller(
    search: BizVarDictQueryParam = Depends(),
    auth: AuthSchema = Depends(AuthPermission(["module_business:vardict:query"]))
) -> JSONResponse:
    """查询变量信息列表接口（数据库分页）"""
    result_dict = await BizVarDictService.vardict_analog_list(auth=auth,crane_no=search.crane_no)
    #请求采集接口获取状态信息
    async with httpx.AsyncClient() as client:
        response = await client.get(
            url=settings.COLLECT_DATA_FULL,
            params={},
            timeout=2
        )
        if response.status_code == 200:
            json_data = response.json()
            if json_data['code'] == 200 and json_data['data']:
                for item in result_dict:
                    item['value'] = False
                    crane_no = item['crane_no']
                    analog = json_data.get('data').get(crane_no).get('data').get('analog').get(item['var_code'])
                    if analog:
                        item['value'] = analog.get('value')
    log.info("查询变量信息列表成功")
    return SuccessResponse(data=result_dict, msg="查询变量信息列表成功")

@BizVarDictRouter.get("/varDictMecGroup/{crane_no}", summary="获取变量信息分组数据", description="获取变量信息分组数据")
async def get_vardict_mec_group_controller(
    crane_no: str = Path(..., description="crane_no"),
    redis: Redis = Depends(redis_getter),
    auth: AuthSchema = Depends(AuthPermission(["module_business:crane:query"]))
) -> JSONResponse:

    result_dict = await BizVarDictService.get_vardict_group_service(
        redis=redis, crane_no=crane_no,auth=auth
    )
    if not result_dict:
        log.info(f"获取变量信息分组数据成功：{result_dict}")
        return SuccessResponse(data=result_dict, msg="获取变量分组数据成功")
    #请求采集接口获取状态信息
    async with httpx.AsyncClient() as client:
        response = await client.get(
            url=settings.COLLECT_DATA_FULL,
            params={},
            timeout=2
        )
        if response.status_code == 200:
            json_data = response.json()
            if json_data['code'] == 200 and json_data['data']:
                json_analog = json_data.get('data').get(crane_no).get('data').get('analog')
                json_digital = json_data.get('data').get(crane_no).get('data').get('digital')
                for var_dict in result_dict:
                    for key,inner_dict in var_dict.items():
                        if key != 'mec_type' and key != 'alarm_varList' and key != 'mecVarList_simple':
                            for item in inner_dict:
                                if key == 'digital_varList':
                                    item['value'] = json_digital.get(item.get('var_code')).get('value')
                                else:
                                    item['value'] = json_analog.get(item.get('var_code')).get('value')
    log.info(f"获取变量信息分组数据成功：{result_dict}")
    return SuccessResponse(data=result_dict, msg="获取变量分组数据成功")