214 lines
7.5 KiB
Python
214 lines
7.5 KiB
Python
"""设备池：SQLite devices 表 = 设备清单，本地 adb = 在线状态（STF 替代数据源）。
|
||
|
||
设计：
|
||
- list_configured() — devices 表里 enabled 的设备（清单，管理页维护）
|
||
- list_online() — 本机 `adb devices` 里 state=device 的设备（实时；
|
||
池内有 USB 设备时合并 220 远程 adb server 状态）
|
||
- list_ready() — configured ∩ online（任务调度用）
|
||
- CRUD — add/update/remove/set_enabled（设备池管理页用）
|
||
|
||
红线：全模块不 connect / 不 kill-server / 不 disconnect，遵守既有技术约束。
|
||
后台线程（task_manager 等）调用时由本模块自行推 app context，调用方无需关心。
|
||
"""
|
||
import time
|
||
import threading
|
||
import subprocess
|
||
|
||
from config import ADB_PATH, USB_ADB_HOST, USB_ADB_PORT
|
||
from core.adb_helper import _adb, _adb_remote
|
||
from core.logger import get_logger
|
||
from core.models import db, Device
|
||
|
||
_log = get_logger("core.device_pool")
|
||
|
||
_app = None
|
||
|
||
|
||
def init_app(app):
|
||
"""web_server 启动时调用：绑定 Flask app（供后台线程推 db context），
|
||
并后台刷新一次设备型号（首次启动/设备更换后型号可能变化）。"""
|
||
global _app
|
||
_app = app
|
||
try:
|
||
t = threading.Thread(target=_refresh_models_bg, daemon=True)
|
||
t.start()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _refresh_models_bg():
|
||
"""后台批量采集在线设备型号（启动时/手动触发）。失败静默，不影响启动。"""
|
||
time.sleep(3) # 等服务起来再查
|
||
try:
|
||
refresh_all_models()
|
||
except Exception as e:
|
||
_log.warning(f"启动型号采集失败（可稍后手动采集）: {e}")
|
||
|
||
|
||
def _ctx():
|
||
"""后台线程访问 db 需自行推 app context（与 task_manager._db 同思路）。"""
|
||
if _app is None:
|
||
raise RuntimeError("device_pool 未关联 Flask app（web_server 启动时调用 init_app）")
|
||
return _app.app_context()
|
||
|
||
|
||
# ================== 清单 ==================
|
||
def list_configured(enabled_only=True):
|
||
"""devices 表里的设备清单（serial 列表）。"""
|
||
with _ctx():
|
||
q = Device.query
|
||
if enabled_only:
|
||
q = q.filter_by(enabled=True)
|
||
return [d.serial for d in q.order_by(Device.serial).all()]
|
||
|
||
|
||
def list_devices():
|
||
"""全量设备行（设备池管理页用）。"""
|
||
with _ctx():
|
||
return [d.to_dict() for d in Device.query.order_by(Device.serial).all()]
|
||
|
||
|
||
# ================== 在线状态 ==================
|
||
def list_online():
|
||
"""本机 adb 中 state=device 的设备 + 池内 USB 设备（经 220 远程 adb server）。
|
||
|
||
池内无 USB 设备（serial 无冒号）时不查远程 server，零额外开销。
|
||
"""
|
||
out = _adb("devices") or ""
|
||
serials = []
|
||
for line in out.splitlines()[1:]:
|
||
parts = line.split()
|
||
if len(parts) >= 2 and parts[0] and parts[1] == "device":
|
||
serials.append(parts[0])
|
||
# 池内有 USB 设备时，补查远程 adb server（220，host 网络 5037 全接口监听）
|
||
try:
|
||
usb = [s for s in list_configured() if ":" not in s]
|
||
except Exception:
|
||
usb = []
|
||
if usb:
|
||
try:
|
||
out = _adb_remote(USB_ADB_HOST, USB_ADB_PORT, "devices") or ""
|
||
for line in out.splitlines()[1:]:
|
||
parts = line.split()
|
||
if len(parts) >= 2 and parts[0] and parts[1] == "device":
|
||
if parts[0] not in serials:
|
||
serials.append(parts[0])
|
||
except Exception:
|
||
pass
|
||
return serials
|
||
|
||
|
||
def is_online(serial):
|
||
"""设备当前是否在线（本机 adb 可达且状态正常）。"""
|
||
return serial in list_online()
|
||
|
||
|
||
def list_ready():
|
||
"""可调度设备：清单 ∩ 在线（取代 STF list_free_devices）。"""
|
||
online = set(list_online())
|
||
return [s for s in list_configured() if s in online]
|
||
|
||
|
||
# ================== 管理（CRUD） ==================
|
||
def add_device(serial, name="", note="", enabled=True):
|
||
"""添加/更新设备（upsert）。返回 True 新增 / False 已存在并更新。"""
|
||
serial = (serial or "").strip()
|
||
if not serial:
|
||
return False
|
||
with _ctx():
|
||
d = Device.query.get(serial)
|
||
if d:
|
||
d.name, d.note, d.enabled = name or "", note or "", enabled
|
||
db.session.commit()
|
||
return False
|
||
db.session.add(Device(serial=serial, name=name or "", note=note or "",
|
||
enabled=enabled,
|
||
created_at=time.strftime("%Y-%m-%d %H:%M")))
|
||
db.session.commit()
|
||
_log.info(f"设备池新增设备: {serial}")
|
||
return True
|
||
|
||
|
||
def remove_device(serial):
|
||
"""删除设备。返回是否删除成功。"""
|
||
with _ctx():
|
||
d = Device.query.get(serial)
|
||
if not d:
|
||
return False
|
||
db.session.delete(d)
|
||
db.session.commit()
|
||
_log.info(f"设备池删除设备: {serial}")
|
||
return True
|
||
|
||
|
||
def set_enabled(serial, enabled):
|
||
"""启用/停用设备（停用后不参与调度）。返回是否成功。"""
|
||
with _ctx():
|
||
d = Device.query.get(serial)
|
||
if not d:
|
||
return False
|
||
d.enabled = enabled
|
||
db.session.commit()
|
||
return True
|
||
|
||
|
||
# ================== 型号采集 ==================
|
||
def refresh_model(serial, timeout=8):
|
||
"""采集单台设备型号（getprop ro.product.model）并写回 DB。
|
||
|
||
IP:5555 设备走本机 adb；USB 设备先查本机 adb，查不到再查 220 远程 adb server。
|
||
采集失败返回空串（静默，不影响主流程）。返回采集到的型号。
|
||
"""
|
||
if not serial:
|
||
return ""
|
||
model = ""
|
||
if ":" in serial:
|
||
try:
|
||
r = subprocess.run([ADB_PATH, "-s", serial, "shell",
|
||
"getprop", "ro.product.model"],
|
||
capture_output=True, timeout=timeout)
|
||
model = (r.stdout or b"").decode("utf-8", errors="replace").strip()
|
||
except Exception:
|
||
model = ""
|
||
else:
|
||
# USB：先本机，再远程
|
||
try:
|
||
r = subprocess.run([ADB_PATH, "-s", serial, "shell",
|
||
"getprop", "ro.product.model"],
|
||
capture_output=True, timeout=timeout)
|
||
model = (r.stdout or b"").decode("utf-8", errors="replace").strip()
|
||
except Exception:
|
||
model = ""
|
||
if not model:
|
||
model = _adb_remote(USB_ADB_HOST, USB_ADB_PORT, "-s", serial,
|
||
"shell", "getprop", "ro.product.model")
|
||
model = model.strip() or ""
|
||
if not model:
|
||
return ""
|
||
with _ctx():
|
||
d = Device.query.get(serial)
|
||
if d and d.model != model:
|
||
d.model = model
|
||
db.session.commit()
|
||
_log.info(f"设备池采集型号: {serial} -> {model}")
|
||
return model
|
||
|
||
|
||
def refresh_all_models():
|
||
"""批量采集池内在线设备的型号（并发 10，后台线程调用）。返回成功数。"""
|
||
serials = list_online()
|
||
if not serials:
|
||
return 0
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
ok = 0
|
||
with ThreadPoolExecutor(max_workers=min(10, len(serials))) as pool:
|
||
futures = {pool.submit(refresh_model, s): s for s in serials}
|
||
for fut in as_completed(futures):
|
||
try:
|
||
if fut.result():
|
||
ok += 1
|
||
except Exception:
|
||
pass
|
||
_log.info(f"设备池批量采集型号完成: {ok}/{len(serials)} 台")
|
||
return ok
|