#ADB (Android Debug Bridge) 详解

import subprocess
import time
import os
from typing import Optional, List

class LightADB:
    """轻量级 ADB 控制器，专注 App 爬虫 / 日常自动化"""

    def __init__(self, device_id: Optional[str] = None, timeout: int = 30):
        self.device_id = device_id
        self.timeout = timeout
        self.available_devices = self._fetch_devices()
        self.current_device = self._select_current_device()
        print(f"✅ ADB 初始化完成，当前操作设备: {self.current_device}")

    def _fetch_devices(self) -> List[str]:
        """获取所有已授权在线设备的 ID"""
        try:
            result = subprocess.run(
                ['adb', 'devices'],
                capture_output=True, text=True, timeout=self.timeout
            )
            raw_lines = result.stdout.strip().split('\n')[1:]
            return [
                line.split('\t')[0]
                for line in raw_lines
                if '\tdevice' in line
            ]
        except subprocess.TimeoutExpired:
            print("⚠️ ADB 命令超时，请检查服务是否正常")
            return []
        except Exception as e:
            print(f"❌ 获取设备列表失败: {e}")
            return []

    def _select_current_device(self) -> str:
        """选择当前要操作的设备"""
        if not self.available_devices:
            raise RuntimeError("❌ 未找到可用 ADB 设备，请检查连接和授权。")
        if self.device_id and self.device_id in self.available_devices:
            return self.device_id
        return self.available_devices[0]

    def _exec_cmd(self, cmd_list: List[str]) -> str:
        """执行 adb -s <serial> <cmd> 命令并返回输出"""
        full_cmd = ['adb', '-s', self.current_device] + cmd_list
        try:
            result = subprocess.run(
                full_cmd,
                capture_output=True, text=True, timeout=self.timeout
            )
            if result.returncode != 0:
                print(f"⚠️ 命令执行失败: {' '.join(full_cmd)}\n{result.stderr.strip()}")
                return ""
            return result.stdout.strip()
        except Exception as e:
            print(f"❌ 命令异常: {e}")
            return ""

    # ---------- 常用操作方法 ----------
    def get_third_packages(self) -> List[str]:
        """获取所有第三方应用包名"""
        output = self._exec_cmd(['shell', 'pm', 'list', 'packages', '-3'])
        return [line.split(':')[1] for line in output.split('\n') if line.strip()]

    def capture_screen(self, save_dir: str = "./adb_screenshots") -> Optional[str]:
        """
        手机截屏并自动拉取到本地
        :param save_dir: 本地保存目录，会自动创建
        :return: 成功时返回图片路径，失败返回 None
        """
        os.makedirs(save_dir, exist_ok=True)
        timestamp = int(time.time())
        temp_phone_path = f"/sdcard/adb_temp_{timestamp}.png"
        local_path = os.path.join(save_dir, f"screen_{timestamp}.png")

        # 1. 手机端截图
        self._exec_cmd(['shell', 'screencap', temp_phone_path])
        # 2. 拉取到电脑
        pull_res = self._exec_cmd(['pull', temp_phone_path, local_path])
        # 3. 删除手机临时文件
        self._exec_cmd(['shell', 'rm', temp_phone_path])

        if "pulled" in pull_res.lower():
            print(f"✅ 截屏已保存: {local_path}")
            return local_path
        return None

# ================= 使用示例 =================
if __name__ == "__main__":
    try:
        adb = LightADB()

        # 1. 查看前 3 个第三方应用
        packages = adb.get_third_packages()[:3]
        print(f"📱 前 3 个第三方应用: {packages}")

        # 2. 每隔 2 秒截屏一次，连续截 3 次
        for i in range(3):
            adb.capture_screen()
            if i < 2:
                time.sleep(2)

    except Exception as e:
        print(f"❌ 示例运行失败: {e}")