from fastapi import APIRouter, Depends, Form, Request, Response 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.dao.robot_role_pairing_dao import PairingDao from module_admin.entity.vo.user_vo import CurrentUserModel from module_admin.service.login_service import LoginService from module_admin.service.robot_role_pairing_service import PairingService from module_admin.entity.vo.robot_role_pairing_vo import DeletePairingModel, PairingModel, PairingPageQueryModel, \ ChangeRobotRoleModel from utils.common_util import bytes2file_response, export_list2excel from utils.log_util import logger from utils.page_util import PageResponseModel from utils.response_util import ResponseUtil from datetime import datetime pairingController = APIRouter(prefix='/system/robot_role' # , dependencies=[Depends(LoginService.get_current_user)] ) # 获取机器人角色 @pairingController.get( '/{robot_id}', ) async def get_system_role_pairing_list( request: Request, robot_id: int, query_db: AsyncSession = Depends(get_db), ): # 获取分页数据 pairing_list, style = await PairingService.get_role_pairing_list_services(query_db, robot_id) return ResponseUtil.success(data={"role": pairing_list, "style":style}) # 获取机器人角色新版 @pairingController.get( '/v1/{robot_id}', ) async def get_system_role_pairing_list_v1( request: Request, robot_id: int, query_db: AsyncSession = Depends(get_db), ): pairing_list, style = await PairingService.get_role_pairing_list_services_v1(query_db, robot_id) return ResponseUtil.success(data={"role": pairing_list, "style":style}) @pairingController.post( '/change', ) async def change_robot_role( request: Request, change_pairing: ChangeRobotRoleModel, query_db: AsyncSession = Depends(get_db), ): await PairingService.change_robot_role_services(query_db,change_pairing) return ResponseUtil.success() # @pairingController.get( # '/list', response_model=PageResponseModel, dependencies=[Depends(CheckUserInterfaceAuth('system:pairing:list'))] # ) # async def get_system_pairing_list( # request: Request, # pairing_page_query: PairingPageQueryModel = Depends(PairingPageQueryModel.as_query), # query_db: AsyncSession = Depends(get_db), # ): # # 获取分页数据 # pairing_page_query_result = await PairingService.get_pairing_list_services(query_db, pairing_page_query, is_page=True) # logger.info('获取成功') # # return ResponseUtil.success(model_content=pairing_page_query_result) # # # @pairingController.post('', dependencies=[Depends(CheckUserInterfaceAuth('system:pairing:add'))]) # @ValidateFields(validate_model='add_pairing') # @Log(title='角色-机器人-映射', business_type=BusinessType.INSERT) # async def add_system_pairing( # request: Request, # add_pairing: PairingModel, # query_db: AsyncSession = Depends(get_db), # current_user: CurrentUserModel = Depends(LoginService.get_current_user), # ): # add_pairing.update_time = datetime.now() # add_pairing.update_by = current_user.user.user_name # add_pairing_result = await PairingService.add_pairing_services(query_db, add_pairing) # logger.info(add_pairing_result.message) # # return ResponseUtil.success(msg=add_pairing_result.message) # # # @pairingController.put('', dependencies=[Depends(CheckUserInterfaceAuth('system:pairing:edit'))]) # @ValidateFields(validate_model='edit_pairing') # @Log(title='角色-机器人-映射', business_type=BusinessType.UPDATE) # async def edit_system_pairing( # request: Request, # edit_pairing: PairingModel, # query_db: AsyncSession = Depends(get_db), # current_user: CurrentUserModel = Depends(LoginService.get_current_user), # ): # edit_pairing.update_by = current_user.user.user_name # edit_pairing.update_time = datetime.now() # edit_pairing_result = await PairingService.edit_pairing_services(query_db, edit_pairing) # logger.info(edit_pairing_result.message) # # return ResponseUtil.success(msg=edit_pairing_result.message) # # # @pairingController.delete('/{pairing_ids}', dependencies=[Depends(CheckUserInterfaceAuth('system:pairing:remove'))]) # @Log(title='角色-机器人-映射', business_type=BusinessType.DELETE) # async def delete_system_pairing(request: Request, pairing_ids: str, query_db: AsyncSession = Depends(get_db)): # delete_pairing = DeletePairingModel(pairingIds=pairing_ids) # delete_pairing_result = await PairingService.delete_pairing_services(query_db, delete_pairing) # logger.info(delete_pairing_result.message) # # return ResponseUtil.success(msg=delete_pairing_result.message) # # # @pairingController.get( # '/{pairing_id}', response_model=PairingModel, dependencies=[Depends(CheckUserInterfaceAuth('system:pairing:query'))] # ) # async def query_detail_system_pairing(request: Request, pairing_id: int, query_db: AsyncSession = Depends(get_db)): # pairing_detail_result = await PairingService.pairing_detail_services(query_db, pairing_id) # logger.info(f'获取pairing_id为{pairing_id}的信息成功') # # return ResponseUtil.success(data=pairing_detail_result) # # # @pairingController.post('/export', dependencies=[Depends(CheckUserInterfaceAuth('system:pairing:export'))]) # @Log(title='角色-机器人-映射', business_type=BusinessType.EXPORT) # async def export_system_pairing_list( # request: Request, # pairing_page_query: PairingPageQueryModel = Form(), # query_db: AsyncSession = Depends(get_db), # ): # # 获取全量数据 # pairing_query_result = await PairingService.get_pairing_list_services(query_db, pairing_page_query, is_page=False) # pairing_export_result = await PairingService.export_pairing_list_services(pairing_query_result) # logger.info('导出成功') # # return Response( # # content=pairing_export_result, # # media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", # # headers={ # # "Content-Disposition": "attachment; filename=export.xlsx" # # } # # ) # return ResponseUtil.streaming(data=bytes2file_response(pairing_export_result))