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):
"""
初始化ADB控制器
:param device_id: 指定设备ID（多设备同时连接时必填）
:param timeout: 单条ADB命令超时时间（秒）
"""
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
)
# 跳过标题行，只保留状态为device的设备
raw_devices = result.stdout.strip().split('\n')[1:]
return [
line.split('\t')[0]
for line in raw_devices
if line.strip() and 'device' in line and 'offline' not in line
]
except subprocess.TimeoutExpired:
print("⚠️ ADB命令超时，请检查服务是否正常")
return []
except Exception as e:
print(f"❌ 获取设备列表失败: {str(e)}")
return []
def _select_current_device(self) -> str:
"""内部方法：自动或手动选择当前设备"""
if not self.available_devices:
raise Exception("❌ 未找到可用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:
"""
内部方法：执行带设备ID的ADB命令
:param cmd_list: 不带adb前缀的命令列表（比如['install', 'xxx.apk']）
:return: 命令执行后的标准输出，失败返回空字符串
"""
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"❌ 命令执行异常: {str(e)}")
return ""
# ------------------- 以下是常用的公开方法 -------------------
def get_third_packages(self) -> List[str]:
"""获取所有第三方应用的包名（App爬虫首选）"""
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个第三方应用
third_pkgs = adb.get_third_packages()[:3]
print(f"📱 前3个第三方应用: {third_pkgs}")
# 2. 连续3次截屏，间隔2秒
for i in range(3):
adb.capture_screen()
if i < 2:
time.sleep(2)
except Exception as e:
print(f"❌ 示例运行失败: {str(e)}")
