TermControlAgent/app/services/adb_service.py

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服务类 - 实现ADB协议相关操作"""
    
    def __init__(self):
        self.adb_client = None
        self._initialize_adb()
    
    def _initialize_adb(self):
        """初始化ADB客户端"""
        try:
            self.adb_client = adb.connect()
            logger.info("ADB客户端初始化成功")
        except Exception as e:
            logger.error(f"ADB客户端初始化失败: {e}")
            self.adb_client = None
    
    async def _get_device(self, device_id: str):
        """获取ADB设备对象"""
        if not self.adb_client:
            self._initialize_adb()
        
        try:
            device = self.adb_client.device(device_id)
            return device
        except Exception as e:
            logger.error(f"获取ADB设备失败: {device_id}, 错误: {e}")
            raise ValueError(f"设备 {device_id} 不存在或未连接")
    
    async def click(self, device_id: str, x: int, y: int, duration: int = 100) -> dict:
        """执行点击操作"""
        try:
            device = await self._get_device(device_id)
            
            # 使用adb shell input命令执行点击
            if duration > 0:
                # 长按
                cmd = f"input touchscreen swipe {x} {y} {x} {y} {duration}"
            else:
                # 点击
                cmd = f"input tap {x} {y}"
            
            result = device.shell(cmd)
            
            logger.info(f"ADB点击操作成功: {device_id} -> ({x}, {y})")
            return {
                "success": True,
                "message": "点击操作执行成功",
                "coordinates": {"x": x, "y": y},
                "duration": duration
            }
        except Exception as e:
            logger.error(f"ADB点击操作失败: {device_id}, 错误: {e}")
            return {
                "success": False,
                "message": f"点击操作失败: {str(e)}"
            }
    
    async def input_text(self, device_id: str, text: str, clear_first: bool = False) -> dict:
        """输入文本"""
        try:
            device = await self._get_device(device_id)
            
            if clear_first:
                # 先清空输入框
                device.shell("input keyevent KEYCODE_MOVE_END")
                device.shell("input keyevent KEYCODE_DEL")
            
            # 输入文本
            result = device.shell(f"input text '{text}'")
            
            logger.info(f"ADB文本输入成功: {device_id} -> {text}")
            return {
                "success": True,
                "message": "文本输入成功",
                "text": text
            }
        except Exception as e:
            logger.error(f"ADB文本输入失败: {device_id}, 错误: {e}")
            return {
                "success": False,
                "message": f"文本输入失败: {str(e)}"
            }
    
    async def screenshot(self, device_id: str) -> ScreenshotResponse:
        """获取屏幕截图"""
        try:
            device = await self._get_device(device_id)
            
            # 获取截图
            screenshot_data = device.screenshot()
            
            # 转换为Base64
            import io
            from PIL import Image
            
            img_buffer = io.BytesIO()
            screenshot_data.save(img_buffer, format='PNG')
            img_data = base64.b64encode(img_buffer.getvalue()).decode('utf-8')
            
            logger.info(f"ADB截图成功: {device_id}")
            return ScreenshotResponse(
                image_data=img_data,
                format="png",
                width=screenshot_data.width,
                height=screenshot_data.height
            )
        except Exception as e:
            logger.error(f"ADB截图失败: {device_id}, 错误: {e}")
            raise ValueError(f"截图失败: {str(e)}")
    
    async def install_apk(self, device_id: str, apk_path: str, package_name: Optional[str] = None) -> InstallResponse:
        """安装APK"""
        try:
            device = await self._get_device(device_id)
            
            # 安装APK
            result = device.install(apk_path)
            
            if result:
                logger.info(f"ADB安装APK成功: {device_id} -> {apk_path}")
                return InstallResponse(
                    success=True,
                    message="APK安装成功",
                    package_name=package_name
                )
            else:
                logger.error(f"ADB安装APK失败: {device_id} -> {apk_path}")
                return InstallResponse(
                    success=False,
                    message="APK安装失败"
                )
        except Exception as e:
            logger.error(f"ADB安装APK异常: {device_id}, 错误: {e}")
            return InstallResponse(
                success=False,
                message=f"APK安装异常: {str(e)}"
            )
    
    async def get_logcat(self, device_id: str, package_name: Optional[str] = None, 
                        level: str = "V", max_lines: int = 100) -> LogcatResponse:
        """获取日志"""
        try:
            device = await self._get_device(device_id)
            
            # 构建logcat命令
            cmd = f"logcat -d -v time"
            if package_name:
                cmd += f" | grep {package_name}"
            if level != "V":
                cmd += f" *:{level}"
            
            # 执行命令
            result = device.shell(cmd)
            
            # 处理结果
            lines = result.strip().split('\n') if result.strip() else []
            logs = lines[-max_lines:] if len(lines) > max_lines else lines
            
            logger.info(f"ADB获取日志成功: {device_id}")
            return LogcatResponse(
                logs=logs,
                total_lines=len(logs)
            )
        except Exception as e:
            logger.error(f"ADB获取日志失败: {device_id}, 错误: {e}")
            return LogcatResponse(
                logs=[],
                total_lines=0
            )
    
    async def get_device_info(self, device_id: str) -> dict:
        """获取设备信息"""
        try:
            device = await self._get_device(device_id)
            
            # 获取设备信息
            model = device.shell("getprop ro.product.model").strip()
            android_version = device.shell("getprop ro.build.version.release").strip()
            sdk_version = int(device.shell("getprop ro.build.version.sdk").strip())
            
            return {
                "device_id": device_id,
                "model": model,
                "android_version": android_version,
                "sdk_version": sdk_version,
                "status": "online"
            }
        except Exception as e:
            logger.error(f"获取设备信息失败: {device_id}, 错误: {e}")
            return {
                "device_id": device_id,
                "model": "Unknown",
                "android_version": "Unknown",
                "sdk_version": 0,
                "status": "error"
            }
    
    async def list_devices(self) -> List[str]:
        """列出所有ADB设备"""
        try:
            if not self.adb_client:
                self._initialize_adb()
            
            devices = self.adb_client.device_list()
            device_ids = [device.serial for device in devices]
            
            logger.info(f"ADB设备列表: {device_ids}")
            return device_ids
        except Exception as e:
            logger.error(f"获取ADB设备列表失败: {e}")
            return []