hd-crane-gs/backend/app/api/v1/module_business/crane/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 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 BizCraneService
from .schema import BizCraneCreateSchema, BizCraneUpdateSchema, BizCraneQueryParam

BizCraneRouter = APIRouter(prefix='/crane', tags=["行车信息模块"]) 

@BizCraneRouter.get("/detail/{id}", summary="获取行车信息详情", description="获取行车信息详情")
async def get_crane_detail_controller(
    id: int = Path(..., description="ID"),
    auth: AuthSchema = Depends(AuthPermission(["module_business:crane:query"]))
) -> JSONResponse:
    """获取行车信息详情接口"""
    result_dict = await BizCraneService.detail_crane_service(auth=auth, id=id)
    log.info(f"获取行车信息详情成功 {id}")
    return SuccessResponse(data=result_dict, msg="获取行车信息详情成功")

@BizCraneRouter.get("/detailForNo/{crane_no}", summary="获取行车信息详情", description="获取行车信息详情")
async def get_crane_detail_controller(
    crane_no: str = Path(..., description="CraneNo"),
    auth: AuthSchema = Depends(AuthPermission(["module_business:crane:query"]))
) -> JSONResponse:
    """获取行车信息详情接口"""
    result_dict = await BizCraneService.detail_crane_service_for_no(auth=auth, crane_no=crane_no)
    log.info(f"获取行车信息详情成功 {id}")
    return SuccessResponse(data=result_dict, msg="获取行车信息详情成功")

@BizCraneRouter.get("/list", summary="查询行车信息列表", description="查询行车信息列表")
async def get_crane_list_controller(
    page: PaginationQueryParam = Depends(),
    search: BizCraneQueryParam = Depends(),
    auth: AuthSchema = Depends(AuthPermission(["module_business:crane:query"]))
) -> JSONResponse:
    """查询行车信息列表接口（数据库分页）"""
    result_dict = await BizCraneService.page_crane_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="查询行车信息列表成功")

@BizCraneRouter.get("/list_status", summary="查询行车信息列表", description="查询行车信息列表")
async def get_crane_list_status_controller(
    page: PaginationQueryParam = Depends(),
    search: BizCraneQueryParam = Depends(),
    auth: AuthSchema = Depends(AuthPermission(["module_business:crane:query"]))
) -> JSONResponse:
    """查询行车信息列表接口（数据库分页）"""
    result_dict = await BizCraneService.page_crane_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
    )
    #请求采集接口获取状态信息
    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['items']:
                    crane_no = item['crane_no']
                    crane_data = json_data.get('data').get(crane_no)
                    if crane_data:
                        item['work_status'] = crane_data.get('data').get('status').get('status')
    log.info("查询行车信息列表成功")
    return SuccessResponse(data=result_dict, msg="查询行车信息列表成功")

@BizCraneRouter.post("/create", summary="创建行车信息", description="创建行车信息")
async def create_crane_controller(
    data: BizCraneCreateSchema,
    redis: Redis = Depends(redis_getter),
    auth: AuthSchema = Depends(AuthPermission(["module_business:crane:create"]))
) -> JSONResponse:
    """创建行车信息接口"""
    result_dict = await BizCraneService.create_crane_service(auth=auth, data=data,redis=redis)
    log.info("创建行车信息成功")
    return SuccessResponse(data=result_dict, msg="创建行车信息成功")

@BizCraneRouter.put("/update/{id}", summary="修改行车信息", description="修改行车信息")
async def update_crane_controller(
    data: BizCraneUpdateSchema,
    id: int = Path(..., description="ID"),
    redis: Redis = Depends(redis_getter),
    auth: AuthSchema = Depends(AuthPermission(["module_business:crane:update"]))
) -> JSONResponse:
    """修改行车信息接口"""
    result_dict = await BizCraneService.update_crane_service(auth=auth, id=id, data=data,redis=redis)
    log.info("修改行车信息成功")
    return SuccessResponse(data=result_dict, msg="修改行车信息成功")

@BizCraneRouter.delete("/delete", summary="删除行车信息", description="删除行车信息")
async def delete_crane_controller(
    ids: list[int] = Body(..., description="ID列表"),
    redis: Redis = Depends(redis_getter),
    auth: AuthSchema = Depends(AuthPermission(["module_business:crane:delete"]))
) -> JSONResponse:
    """删除行车信息接口"""
    await BizCraneService.delete_crane_service(auth=auth, ids=ids,redis=redis)
    log.info(f"删除行车信息成功: {ids}")
    return SuccessResponse(msg="删除行车信息成功")

@BizCraneRouter.patch("/available/setting", summary="批量修改行车信息状态", description="批量修改行车信息状态")
async def batch_set_available_crane_controller(
    data: BatchSetAvailable,
    redis: Redis = Depends(redis_getter),
    auth: AuthSchema = Depends(AuthPermission(["module_business:crane:patch"]))
) -> JSONResponse:
    """批量修改行车信息状态接口"""
    await BizCraneService.set_available_crane_service(auth=auth, data=data,redis=redis)
    log.info(f"批量修改行车信息状态成功: {data.ids}")
    return SuccessResponse(msg="批量修改行车信息状态成功")

@BizCraneRouter.post('/export', summary="导出行车信息", description="导出行车信息")
async def export_crane_list_controller(
    search: BizCraneQueryParam = Depends(),
    auth: AuthSchema = Depends(AuthPermission(["module_business:crane:export"]))
) -> StreamingResponse:
    """导出行车信息接口"""
    result_dict_list = await BizCraneService.list_crane_service(search=search, auth=auth)
    export_result = await BizCraneService.batch_export_crane_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_crane.xlsx'
        }
    )

@BizCraneRouter.post('/import', summary="导入行车信息", description="导入行车信息")
async def import_crane_list_controller(
    file: UploadFile,
    auth: AuthSchema = Depends(AuthPermission(["module_business:crane:import"]))
) -> JSONResponse:
    """导入行车信息接口"""
    batch_import_result = await BizCraneService.batch_import_crane_service(file=file, auth=auth, update_support=True)
    log.info("导入行车信息成功")
    
    return SuccessResponse(data=batch_import_result, msg="导入行车信息成功")

@BizCraneRouter.post('/download/template', summary="获取行车信息导入模板", description="获取行车信息导入模板", dependencies=[Depends(AuthPermission(["module_business:crane:download"]))])
async def export_crane_template_controller() -> StreamingResponse:
    """获取行车信息导入模板接口"""
    import_template_result = await BizCraneService.import_template_download_crane_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_crane_template.xlsx'}
    )