""" ADB服务模块 """ import asyncio import base64 from typing import Optional, List from adbutils import adb from app.core.device.manager import device_manager from app.schemas.adb import ( ClickRequest, InputRequest, ScreenshotResponse, InstallRequest, InstallResponse, LogcatRequest, LogcatResponse ) from app.utils.log import get_logger logger = get_logger(__name__) class AdbService: """ADB服务类""" def __init__(self): self.device_manager = device_manager async def get_devices(self): """获取所有ADB设备""" try: devices = await self.device_manager.get_all_devices() adb_devices = [d for d in devices if d.protocol_type == "adb"] return adb_devices except Exception as e: logger.error(f"获取ADB设备失败: {e}") return [] async def execute_command(self, device_id: str, command: str): """在指定设备上执行ADB命令""" try: device = await self.device_manager.get_device(device_id) if not device: raise ValueError(f"设备 {device_id} 不存在") if device.protocol_type != "adb": raise ValueError(f"设备 {device_id} 不是ADB设备") # 这里应该实现具体的ADB命令执行逻辑 logger.info(f"在设备 {device_id} 上执行命令: {command}") return {"success": True, "command": command, "device_id": device_id} except Exception as e: logger.error(f"执行ADB命令失败: {e}") return {"success": False, "error": str(e)}