191 lines
7.8 KiB
Python
191 lines
7.8 KiB
Python
import html
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import time
|
||
from pathlib import Path
|
||
|
||
from server.services.device import DeviceProfile
|
||
|
||
PACKAGE = "com.leivideo.raptdrama"
|
||
PREFS = f"/data/data/{PACKAGE}/shared_prefs/FlutterSharedPreferences.xml"
|
||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||
|
||
|
||
class Adb:
|
||
def __init__(self, executable: str) -> None:
|
||
self.executable = executable
|
||
|
||
@classmethod
|
||
def locate(cls, explicit: str = "") -> "Adb":
|
||
if explicit:
|
||
if not os.path.isfile(explicit):
|
||
raise RuntimeError(f"找不到 adb：{explicit}")
|
||
return cls(explicit)
|
||
name = "adb.exe" if os.name == "nt" else "adb"
|
||
path = REPO_ROOT / "tools" / "platform-tools" / name
|
||
if not path.is_file():
|
||
raise RuntimeError(f"项目内没有 adb：{path}")
|
||
return cls(str(path))
|
||
|
||
def list_devices(self) -> list[dict]:
|
||
proc = self._run(["devices", "-l"])
|
||
if proc.returncode not in (0, None):
|
||
err = proc.stderr.decode("utf-8", "replace").strip() or "adb devices 失败"
|
||
raise RuntimeError(err)
|
||
items = []
|
||
for line in proc.stdout.decode("utf-8", "replace").splitlines()[1:]:
|
||
parts = line.split()
|
||
if len(parts) < 2 or parts[1] not in ("device", "unauthorized", "offline"):
|
||
continue
|
||
serial, state = parts[0], parts[1]
|
||
model = ""
|
||
for part in parts[2:]:
|
||
if part.startswith("model:"):
|
||
model = part.split(":", 1)[1].replace("_", " ")
|
||
if state == "device" and not model:
|
||
_, text, _ = self._shell(serial, ["getprop", "ro.product.model"], timeout=8)
|
||
model = text.strip()
|
||
items.append({
|
||
"serial": serial,
|
||
"model": model or "未知型号",
|
||
"rooted": self._rooted(serial) if state == "device" else False,
|
||
"state": state,
|
||
})
|
||
return items
|
||
|
||
def read_profile(self, serial: str) -> DeviceProfile:
|
||
serial = self._ready(serial)
|
||
profile = DeviceProfile.builtin()
|
||
brand = self._prop(serial, "ro.product.brand") or self._prop(serial, "ro.product.manufacturer")
|
||
model = self._prop(serial, "ro.product.model")
|
||
os_version = self._prop(serial, "ro.build.version.release")
|
||
characteristics = self._prop(serial, "ro.build.characteristics")
|
||
qemu = self._prop(serial, "ro.kernel.qemu") or self._prop(serial, "ro.boot.qemu")
|
||
_, android_id, err = self._shell(serial, ["settings", "get", "secure", "android_id"], timeout=8)
|
||
if err == "timeout":
|
||
raise RuntimeError("读取设备信息超时，请确认设备仍连接后重试。")
|
||
android_id = android_id.strip()
|
||
version_name, version_code = self._app_version(serial)
|
||
if brand:
|
||
profile.brand = brand
|
||
if model:
|
||
profile.model = model
|
||
if os_version:
|
||
profile.os_version = os_version
|
||
if version_name:
|
||
profile.app_version = version_name
|
||
if version_code:
|
||
profile.app_build = version_code
|
||
if android_id and android_id.lower() != "null":
|
||
profile.device_id = f"rd_{android_id}"
|
||
profile.device_type = "pad" if "tablet" in characteristics else "phone"
|
||
profile.physical = not (qemu == "1" or serial.startswith("emulator-"))
|
||
return profile
|
||
|
||
def read_token(self, serial: str) -> str:
|
||
serial = self._ready(serial)
|
||
if not self._app_installed(serial):
|
||
raise RuntimeError("这台设备还没有安装 RaptDrama。请先安装 App 并登录，然后再读取登录信息。")
|
||
text = self._prefs(serial)
|
||
match = re.search(r'name="flutter\.user_data"[^>]*>(.*?)</string>', text, re.S)
|
||
if not match:
|
||
raise RuntimeError("登录信息里没有 user_data，请先在 App 里登录")
|
||
user = json.loads(html.unescape(match.group(1)))
|
||
token = user.get("accessToken") or ""
|
||
if not token:
|
||
raise RuntimeError("当前账号没有 accessToken，请先在 App 里登录")
|
||
return token
|
||
|
||
def _ready(self, serial: str) -> str:
|
||
serial = (serial or "").strip()
|
||
if ":" in serial:
|
||
self._run(["connect", serial], timeout=15)
|
||
ready = self._ready_serials()
|
||
if serial not in ready:
|
||
shown = "、".join(ready) or "无"
|
||
raise RuntimeError(f"设备 {serial} 未就绪。当前已连接：{shown}")
|
||
return serial
|
||
|
||
def _ready_serials(self) -> list[str]:
|
||
proc = self._run(["devices"])
|
||
serials = []
|
||
for line in proc.stdout.decode("utf-8", "replace").splitlines()[1:]:
|
||
parts = line.split()
|
||
if len(parts) >= 2 and parts[1] == "device":
|
||
serials.append(parts[0])
|
||
return serials
|
||
|
||
def _app_installed(self, serial: str) -> bool:
|
||
_, text, err = self._shell(serial, ["pm", "path", PACKAGE], timeout=8)
|
||
if err == "timeout":
|
||
raise RuntimeError("检查 App 是否安装超时，请确认设备仍连接后重试。")
|
||
return "package:" in text
|
||
|
||
def _app_version(self, serial: str) -> tuple[str, str]:
|
||
_, text, _ = self._shell(serial, ["dumpsys", "package", PACKAGE], timeout=15)
|
||
version_name = ""
|
||
version_code = ""
|
||
for line in text.splitlines():
|
||
line = line.strip()
|
||
if line.startswith("versionName="):
|
||
version_name = line.split("=", 1)[1].strip()
|
||
elif line.startswith("versionCode="):
|
||
version_code = line.split("=", 1)[1].split()[0]
|
||
return version_name, version_code
|
||
|
||
def _prop(self, serial: str, name: str) -> str:
|
||
_, text, err = self._shell(serial, ["getprop", name], timeout=8)
|
||
if err == "timeout":
|
||
raise RuntimeError("读取设备信息超时，请确认设备仍连接后重试。")
|
||
return text.strip()
|
||
|
||
def _rooted(self, serial: str) -> bool:
|
||
_, text, _ = self._shell(serial, ["id"], timeout=5)
|
||
if "uid=0" in text:
|
||
return True
|
||
for args in (["su", "-c", "id"], ["su", "0", "id"]):
|
||
_, text, _ = self._shell(serial, args, timeout=4)
|
||
if "uid=0" in text:
|
||
return True
|
||
return False
|
||
|
||
def _prefs(self, serial: str) -> str:
|
||
def usable(text: str) -> bool:
|
||
return "flutter.user_data" in text
|
||
|
||
_, text, _ = self._shell(serial, ["cat", PREFS])
|
||
if usable(text):
|
||
return text
|
||
self._run(["-s", serial, "root"], timeout=15)
|
||
if ":" in serial:
|
||
self._run(["connect", serial], timeout=15)
|
||
for _ in range(10):
|
||
time.sleep(0.5)
|
||
_, text, _ = self._shell(serial, ["cat", PREFS])
|
||
if usable(text):
|
||
return text
|
||
for args in (["su", "-c", f"cat {PREFS}"], ["su", "0", "cat", PREFS]):
|
||
_, text, _ = self._shell(serial, args, timeout=8)
|
||
if usable(text):
|
||
return text
|
||
raise RuntimeError(
|
||
"读不到应用私有目录。未 root 的设备无法获取登录信息。"
|
||
"请使用已 root 的手机，或带 root 的模拟器。"
|
||
)
|
||
|
||
def _shell(self, serial: str, args: list[str], timeout: int = 15) -> tuple[int, str, str]:
|
||
try:
|
||
proc = self._run(["-s", serial, "shell", *args], timeout=timeout)
|
||
except subprocess.TimeoutExpired:
|
||
return 1, "", "timeout"
|
||
return (
|
||
proc.returncode,
|
||
proc.stdout.decode("utf-8", "replace"),
|
||
proc.stderr.decode("utf-8", "replace"),
|
||
)
|
||
|
||
def _run(self, args: list[str], timeout: int = 20) -> subprocess.CompletedProcess:
|
||
return subprocess.run([self.executable, *args], capture_output=True, timeout=timeout)
|