from fastapi import APIRouter, HTTPException, status, UploadFile, File from typing import Optional from app.core.device.manager import device_manager from app.core.device.dispatcher import device_dispatcher from app.schemas.adb import ( ClickRequest, InputRequest, ScreenshotResponse, InstallRequest, InstallResponse, LogcatRequest, LogcatResponse ) from app.utils.log import get_logger logger = get_logger(__name__) router = APIRouter() @router.post("/devices/{device_id}/click", summary="执行点击操作") async def click_device(device_id: str, click_request: ClickRequest): """在指定设备上执行点击操作""" try: # 检查设备是否存在 device = await device_manager.get_device(device_id) if not device: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"设备 {device_id} 不存在" ) # 检查协议是否支持点击操作 if device.protocol_type not in ['adb', 'atx']: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"协议 {device.protocol_type} 不支持点击操作" ) # 执行点击操作 result = await device_dispatcher.execute_operation( device_id=device_id, protocol_type=device.protocol_type, operation="click", x=click_request.x, y=click_request.y, duration=click_request.duration ) return result except HTTPException: raise except Exception as e: logger.error(f"点击操作失败: {device_id}, 错误: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"点击操作失败: {str(e)}" ) @router.post("/devices/{device_id}/input", summary="输入文本") async def input_text(device_id: str, input_request: InputRequest): """在指定设备上输入文本""" try: # 检查设备是否存在 device = await device_manager.get_device(device_id) if not device: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"设备 {device_id} 不存在" ) # 检查协议是否支持输入操作 if device.protocol_type not in ['adb', 'atx']: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"协议 {device.protocol_type} 不支持输入操作" ) # 执行输入操作 if device.protocol_type == 'adb': result = await device_dispatcher.execute_operation( device_id=device_id, protocol_type=device.protocol_type, operation="input_text", text=input_request.text, clear_first=input_request.clear_first ) else: # atx result = await device_dispatcher.execute_operation( device_id=device_id, protocol_type=device.protocol_type, operation="input_text", text=input_request.text, clear_first=input_request.clear_first ) return result except HTTPException: raise except Exception as e: logger.error(f"输入操作失败: {device_id}, 错误: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"输入操作失败: {str(e)}" ) @router.get("/devices/{device_id}/screenshot", response_model=ScreenshotResponse, summary="获取屏幕截图") async def get_screenshot(device_id: str): """获取指定设备的屏幕截图""" try: # 检查设备是否存在 device = await device_manager.get_device(device_id) if not device: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"设备 {device_id} 不存在" ) # 检查协议是否支持截图操作 if device.protocol_type not in ['adb', 'atx']: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"协议 {device.protocol_type} 不支持截图操作" ) # 执行截图操作 result = await device_dispatcher.execute_operation( device_id=device_id, protocol_type=device.protocol_type, operation="screenshot" ) return result except HTTPException: raise except Exception as e: logger.error(f"截图操作失败: {device_id}, 错误: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"截图操作失败: {str(e)}" ) @router.post("/devices/{device_id}/install", response_model=InstallResponse, summary="安装APK") async def install_apk(device_id: str, install_request: InstallRequest): """在指定设备上安装APK""" try: # 检查设备是否存在 device = await device_manager.get_device(device_id) if not device: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"设备 {device_id} 不存在" ) # 检查协议是否支持安装操作 if device.protocol_type not in ['adb', 'hdc']: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"协议 {device.protocol_type} 不支持安装操作" ) # 执行安装操作 result = await device_dispatcher.execute_operation( device_id=device_id, protocol_type=device.protocol_type, operation="install_apk", apk_path=install_request.apk_path, package_name=install_request.package_name ) return result except HTTPException: raise except Exception as e: logger.error(f"安装APK失败: {device_id}, 错误: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"安装APK失败: {str(e)}" ) @router.get("/devices/{device_id}/logcat", response_model=LogcatResponse, summary="获取日志") async def get_logcat( device_id: str, package_name: Optional[str] = None, level: str = "V", max_lines: int = 100 ): """获取指定设备的日志""" try: # 检查设备是否存在 device = await device_manager.get_device(device_id) if not device: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"设备 {device_id} 不存在" ) # 检查协议是否支持日志操作 if device.protocol_type not in ['adb']: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"协议 {device.protocol_type} 不支持日志操作" ) # 执行日志获取操作 result = await device_dispatcher.execute_operation( device_id=device_id, protocol_type=device.protocol_type, operation="get_logcat", package_name=package_name, level=level, max_lines=max_lines ) return result except HTTPException: raise except Exception as e: logger.error(f"获取日志失败: {device_id}, 错误: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"获取日志失败: {str(e)}" ) @router.get("/devices/{device_id}/info", summary="获取设备信息") async def get_device_info(device_id: str): """获取指定设备的详细信息""" try: # 检查设备是否存在 device = await device_manager.get_device(device_id) if not device: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"设备 {device_id} 不存在" ) # 获取设备信息 result = await device_dispatcher.execute_operation( device_id=device_id, protocol_type=device.protocol_type, operation="get_device_info" ) return result except HTTPException: raise except Exception as e: logger.error(f"获取设备信息失败: {device_id}, 错误: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"获取设备信息失败: {str(e)}" )