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

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

import json
import asyncio
from starlette.websockets import WebSocketDisconnect
from fastapi import APIRouter, Depends, UploadFile, Body, Path, WebSocket
from fastapi.responses import StreamingResponse, JSONResponse
from app.common.response import SuccessResponse, StreamResponse
from app.core.dependencies import AuthPermission
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,
    auth: AuthSchema = Depends(AuthPermission(["module_business:vardict:create"]))
) -> JSONResponse:
    """创建变量信息接口"""
    result_dict = await BizVarDictService.create_vardict_service(auth=auth, data=data)
    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"),
    auth: AuthSchema = Depends(AuthPermission(["module_business:vardict:update"]))
) -> JSONResponse:
    """修改变量信息接口"""
    result_dict = await BizVarDictService.update_vardict_service(auth=auth, id=id, data=data)
    log.info("修改变量信息成功")
    return SuccessResponse(data=result_dict, msg="修改变量信息成功")

@BizVarDictRouter.delete("/delete", summary="删除变量信息", description="删除变量信息")
async def delete_vardict_controller(
    ids: list[int] = Body(..., description="ID列表"),
    auth: AuthSchema = Depends(AuthPermission(["module_business:vardict:delete"]))
) -> JSONResponse:
    """删除变量信息接口"""
    await BizVarDictService.delete_vardict_service(auth=auth, ids=ids)
    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.websocket("/ws/alert", name="变量报警信息", dependencies=[])
async def websocket_chat_controller(
        websocket: WebSocket,
):
    await websocket.accept()
    is_closed = False
    try:

        log.info("前端WebSocket连接成功，开始推送测试数据")
        # 构造测试数据（完全匹配前端预期的格式）
        test_messages = [
            # 测试报警消息（gc/alert）
            {
                "topic": "gc/alert",
                "data": [
                    {"switch_type": "2", "crane_name": "塔吊001", "msg": "轻度过载报警"},
                    {"switch_type": "3", "crane_name": "塔吊002", "msg": "中度倾斜报警"},
                    {"switch_type": "4", "crane_name": "塔吊003", "msg": "重度碰撞报警"}
                ]
            }
        ]
        # 循环推送测试数据（每隔3秒推一次，模拟实时更新）
        while True:
            if is_closed:
                break
            for msg in test_messages:
                # 将字典转为JSON字符串（前端能解析）
                await websocket.send_text(json.dumps(msg))
                log.info(f"已推送测试数据: {msg['topic']}")
                await asyncio.sleep(5)  # 每3秒推一条，可调整间隔
    except WebSocketDisconnect:
        is_closed = True
        log.info("前端主动断开WebSocket连接")
    except Exception as e:
        is_closed = True
        log.error(f"WebSocket异常: {str(e)}", exc_info=True)
    finally:
        if not is_closed:
            try:
                await websocket.close()
                log.info("关闭WebSocket连接")
            except RuntimeError:
                log.debug("WebSocket已关闭，无需重复操作")