from datetime import datetime from fastapi import APIRouter, Depends, Form, Request, Body from pydantic_validation_decorator import ValidateFields from sqlalchemy.ext.asyncio import AsyncSession from config.enums import BusinessType from config.get_db import get_db from module_admin.annotation.log_annotation import Log from module_admin.aspect.interface_auth import CheckUserInterfaceAuth from module_admin.entity.vo.user_vo import CurrentUserModel from module_admin.service.login_service import LoginService from module_admin.service.door_service import DoorService from module_admin.service.haikang_service import HaiKangService from module_admin.entity.vo.door_vo import DeleteDoorModel, DoorModel, DoorPageQueryModel from utils.common_util import bytes2file_response from utils.log_util import logger from utils.page_util import PageResponseModel from utils.response_util import ResponseUtil import time doorController = APIRouter(prefix='/system/door', dependencies=[Depends(LoginService.get_current_user)]) """ 实时监控页面权限 access:monitor:view ,dependencies=[Depends(CheckUserInterfaceAuth('access:monitor:view'))] 识别记录表权限 access:record:list , dependencies=[Depends(CheckUserInterfaceAuth('access:record:list'))] 门禁设备管理 access:device:list, dependencies=[Depends(CheckUserInterfaceAuth('access:device:list'))] """ # 机器人控制开门 @doorController.post('/control_door') async def control_door(request: Request, door_index_code: str = Body(), control_type: int = Body()): print("door_index_code: ", door_index_code) print("control_type: ", control_type) start_time = time.time() result = await HaiKangService.door_do_control_service(door_index_code, control_type) print("end_time: ", time.time() - start_time) if result[0]: logger.info('门控制成功') return ResponseUtil.success(data=result[1]) logger.error('门控制失败') return ResponseUtil.error(msg=result[1]) # 获取门禁设备列表 @doorController.get( '/list', response_model=PageResponseModel # , dependencies=[Depends(CheckUserInterfaceAuth('system:door:list'))] , dependencies=[Depends(CheckUserInterfaceAuth('access:device:list'))] ) async def get_system_door_list( request: Request, door_page_query: DoorPageQueryModel = Depends(DoorPageQueryModel.as_query), query_db: AsyncSession = Depends(get_db), ): # 获取分页数据 door_page_query_result = await DoorService.get_door_list_services(query_db, door_page_query, is_page=True) logger.info('获取成功') return ResponseUtil.success(model_content=door_page_query_result) # 添加门禁设备 @doorController.post('/add' # , dependencies=[Depends(CheckUserInterfaceAuth('system:door:add'))] , dependencies=[Depends(CheckUserInterfaceAuth('access:device:list'))] ) @ValidateFields(validate_model='add_door') @Log(title='门禁设备', business_type=BusinessType.INSERT) async def add_system_door( request: Request, add_door: DoorModel, query_db: AsyncSession = Depends(get_db), current_user: CurrentUserModel = Depends(LoginService.get_current_user), ): add_door.create_time = datetime.now() add_door.create_by = current_user.user.user_name add_door.update_time = datetime.now() add_door.update_by = current_user.user.user_name add_door_result = await DoorService.add_door_services(query_db, add_door) logger.info(add_door_result.message) return ResponseUtil.success(msg=add_door_result.message) @doorController.put('/edit' # , dependencies=[Depends(CheckUserInterfaceAuth('system:door:edit'))] , dependencies=[Depends(CheckUserInterfaceAuth('access:device:list'))] ) @ValidateFields(validate_model='edit_door') @Log(title='门禁设备', business_type=BusinessType.UPDATE) async def edit_system_door( request: Request, edit_door: DoorModel, query_db: AsyncSession = Depends(get_db), current_user: CurrentUserModel = Depends(LoginService.get_current_user), ): edit_door.update_by = current_user.user.user_name edit_door.update_time = datetime.now() edit_door_result = await DoorService.edit_door_services(query_db, edit_door) logger.info(edit_door_result.message) return ResponseUtil.success(msg=edit_door_result.message) @doorController.delete('/{ids}' # , dependencies=[Depends(CheckUserInterfaceAuth('system:door:remove'))] , dependencies=[Depends(CheckUserInterfaceAuth('access:device:list'))] ) @Log(title='门禁设备', business_type=BusinessType.DELETE) async def delete_system_door(request: Request, ids: str, query_db: AsyncSession = Depends(get_db)): delete_door = DeleteDoorModel(ids=ids) delete_door_result = await DoorService.delete_door_services(query_db, delete_door) logger.info(delete_door_result.message) return ResponseUtil.success(msg=delete_door_result.message) # 获取设备状态 @doorController.get('/door_status', dependencies=[Depends(CheckUserInterfaceAuth('access:monitor:view'))]) async def get_door_status(request: Request, query_db: AsyncSession = Depends(get_db)): result = await DoorService.get_door_status_service(query_db) if result[0]: return ResponseUtil.success(data=result[1]) return ResponseUtil.error(msg=result[1]) # 获取视频流地址 @doorController.get('/video_uri/{robot_id}' ,dependencies=[Depends(CheckUserInterfaceAuth('access:monitor:view'))]) async def get_video_uri(request: Request , robot_id: int , query_db: AsyncSession = Depends(get_db) ): result = await DoorService.get_video_uri_service(robot_id, query_db) return ResponseUtil.success(data=result) # @doorController.get( # '/{id}', response_model=DoorModel, dependencies=[Depends(CheckUserInterfaceAuth('system:door:query'))] # ) # async def query_detail_system_door(request: Request, id: int, query_db: AsyncSession = Depends(get_db)): # door_detail_result = await DoorService.door_detail_services(query_db, id) # logger.info(f'获取id为{id}的信息成功') # return ResponseUtil.success(data=door_detail_result) # @doorController.post('/export', dependencies=[Depends(CheckUserInterfaceAuth('system:door:export'))]) # @Log(title='门禁设备', business_type=BusinessType.EXPORT) # async def export_system_door_list( # request: Request, # door_page_query: DoorPageQueryModel = Form(), # query_db: AsyncSession = Depends(get_db), # ): # # 获取全量数据 # door_query_result = await DoorService.get_door_list_services(query_db, door_page_query, is_page=False) # door_export_result = await DoorService.export_door_list_services(door_query_result) # logger.info('导出成功') # return ResponseUtil.streaming(data=bytes2file_response(door_export_result))