butubb
0a1b4d6122
背景：设备池原先拿 serial（IP）当身份。设备一换 IP（DHCP 重新分配）旧记录就成了连不上的
僵尸条目（表现为"断联·自动重连中"但设备并没关机），分组与 serial 模式的任务还吊着死地址。
2026-09-11 实际发生：.70 变成 .71、.72 消失，平台两个条目永远连不上。
实现
- **设备名称必填且唯一**：加入设备必须填名称；库层面用部分唯一索引兜底
（ux_device_name / ux_device_fingerprint，WHERE 非空 → 兼容历史空值），管理页可改名
- **设备指纹**（ro.serialno）：网络设备添加/确认/扫描/采集型号时自动读取；
身份三层拆分——名称（人可读，稳定）、指纹（机器识别，稳定）、serial（当前地址，可变）
- **自动认领**：添加或确认设备时指纹命中池中已有记录 → 迁移原记录到新地址
（名称/型号/备注/启用状态/添加时间全保留），不新增条目
- **人工认领** `POST /api/devices/pool/relocate`：旧地址已断联、指纹没采过时的兜底——
人工指认"这条就是那台，现在在 X"，迁移并同步引用
- **引用同步**：认领/迁址时把 device_group.serials 与 task_job.target.serial 的旧地址
换成新地址。⚠️ 必须同时改**内存**：分组/任务在 TaskManager 里另有内存副本且调度用内存对象，
只改库不重启不生效 → device_pool 迁址后回调 TaskManager.sync_device_serial
（装配层用 set_move_hook 注册；device_pool 不能反向 import task_manager，会循环依赖）
- **前端**：待连接池新增「识别」列（指纹命中时提示"≈ 名称（原 IP）"，按钮变「认领为 X」）；
设备池新增「名称/指纹」列与「改名/换地址」操作；断联设备表也加「换地址」（用户看到断联就在这里）
- 添加设备接口改用 adb_connect_light（单次短超时），避免不可达 IP 让请求卡 30s+；重名校验提前到 adb 之前
文档：API.md §6 重写（设备身份/认领/新接口）、DATA_MODEL.md（新列 + v5/v6 迁移 + 唯一索引）、
ARCHITECTURE.md §4.0（身份三层与引用同步的内存坑）、DEPLOY.md 排查表加"断联但没关机"条目
自测（全通过）：名称必填/唯一/改名/重名拒绝（含库层面约束）；指纹采集（真实读到 .71 的
gy7lskwkkvj7c6b6）；自动认领（指纹命中→迁址+保留名称+带指纹+未命中不误判）；人工认领
（迁址+名称保留+分组与任务引用同步）；浏览器验证设备池/断联表/待连接池三个界面
405 lines
15 KiB
Python
405 lines
15 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
|
||
# 设备换地址（认领/迁址）后的回调：(old_serial, new_serial) -> None
|
||
# 由 web_server 装配时注册为 TaskManager.sync_device_serial——
|
||
# 分组/任务的内存副本在 TaskManager 里，只改库不改内存不生效（见 _move_device_row）
|
||
_move_hook = None
|
||
|
||
|
||
def set_move_hook(fn):
|
||
"""注册"设备换地址"回调（web_server 创建完 TaskManager 后调用）。"""
|
||
global _move_hook
|
||
_move_hook = fn
|
||
|
||
|
||
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]
|
||
|
||
|
||
# ================== 设备指纹（识别"同一台物理设备"） ==================
|
||
def read_fingerprint(serial, timeout=6):
|
||
"""读取设备指纹（ro.serialno）——设备换 IP 后据此认领回原记录。
|
||
|
||
只对网络设备（serial 含 ":"）读取：USB 设备的 serial 本身就是稳定序列号，
|
||
不存在"换地址"问题。采集失败返回空串（静默，不影响主流程）。
|
||
"""
|
||
serial = (serial or "").strip()
|
||
if not serial or ":" not in serial:
|
||
return ""
|
||
# 部分机型 ro.serialno 为空，退回 ro.boot.serialno
|
||
for prop in ("ro.serialno", "ro.boot.serialno"):
|
||
try:
|
||
r = subprocess.run([ADB_PATH, "-s", serial, "shell", "getprop", prop],
|
||
capture_output=True, timeout=timeout)
|
||
fp = (r.stdout or b"").decode("utf-8", errors="replace").strip()
|
||
except Exception:
|
||
continue
|
||
if fp and fp.isprintable():
|
||
return fp
|
||
return ""
|
||
|
||
|
||
def find_by_fingerprint(fingerprint, exclude_serial=""):
|
||
"""按指纹查池中设备（返回 dict 或 None），可排除指定 serial。"""
|
||
fingerprint = (fingerprint or "").strip()
|
||
if not fingerprint:
|
||
return None
|
||
with _ctx():
|
||
row = Device.query.filter(Device.fingerprint == fingerprint).first()
|
||
if not row or row.serial == exclude_serial:
|
||
return None
|
||
return row.to_dict()
|
||
|
||
|
||
def name_taken(name, exclude_serial=""):
|
||
"""名称是否已被其它设备占用（唯一约束的应用层检查，给友好提示用）。"""
|
||
name = (name or "").strip()
|
||
if not name:
|
||
return False
|
||
with _ctx():
|
||
q = Device.query.filter(Device.name == name)
|
||
if exclude_serial:
|
||
q = q.filter(Device.serial != exclude_serial)
|
||
return q.first() is not None
|
||
|
||
|
||
def _move_device_row(old, new_serial, fingerprint=""):
|
||
"""把 old 记录迁到 new_serial（调用方持 app context）。
|
||
|
||
只动 device 表本身；**分组/任务引用的同步交给上层**（见 set_move_hook）：
|
||
分组与任务在 TaskManager 里还有一份内存副本，调度用的是内存对象——
|
||
只改库不改内存，不重启不生效。而 device_pool 不能反向依赖 task_manager
|
||
（会造成循环 import），所以用回调把这件事交给装配层。
|
||
"""
|
||
old_serial, name = old.serial, (old.name or "")
|
||
# 新地址上若已有记录（重复添加等），以"被认领的旧记录"为准，删掉它
|
||
dup = Device.query.get(new_serial)
|
||
if dup is not None and dup.serial != old.serial:
|
||
db.session.delete(dup)
|
||
old.serial = new_serial
|
||
if fingerprint:
|
||
old.fingerprint = fingerprint
|
||
db.session.commit()
|
||
if _move_hook is not None:
|
||
try:
|
||
_move_hook(old_serial, new_serial)
|
||
except Exception as e:
|
||
_log.warning(f"设备迁址后同步分组/任务引用失败: {e}")
|
||
return old_serial, name
|
||
|
||
|
||
def claim_device(new_serial, fingerprint):
|
||
"""认领：同一台物理设备换了地址，把池中旧记录迁到新 serial 并同步所有引用。
|
||
|
||
做三件事（一个事务内）：
|
||
1. 把旧记录的 serial 改成新地址（名称/型号/备注/启用状态/添加时间全部保留）
|
||
2. 同步 device_group.serials 里的旧 serial → 新 serial（否则分组吊着死 IP）
|
||
3. 同步 task_job.target.serial（指定设备模式的任务目标）
|
||
|
||
返回 (old_serial, name)；没有匹配到旧记录时返回 (None, "")。
|
||
"""
|
||
fingerprint = (fingerprint or "").strip()
|
||
new_serial = (new_serial or "").strip()
|
||
if not fingerprint or not new_serial:
|
||
return None, ""
|
||
with _ctx():
|
||
old = Device.query.filter(Device.fingerprint == fingerprint,
|
||
Device.serial != new_serial).first()
|
||
if not old:
|
||
return None, ""
|
||
old_serial, name = _move_device_row(old, new_serial)
|
||
_log.info(f"设备认领: 指纹 {fingerprint} 的『{name or old_serial}』"
|
||
f"由 {old_serial} 迁到 {new_serial}")
|
||
return old_serial, name
|
||
|
||
|
||
def relocate_device(old_serial, new_serial, fingerprint=""):
|
||
"""人工认领：把池中 old_serial 的记录改到 new_serial（换地址的手工兜底）。
|
||
|
||
用在"设备已经断联、读不到指纹"的场景：设备换了 IP 后旧地址连不上，
|
||
指纹也没采过，自动认领无从匹配——此时由人工指认"这条就是那台，
|
||
现在在 X"，本函数负责迁移并同步分组/任务引用。
|
||
|
||
返回 (ok, name, msg)。
|
||
"""
|
||
old_serial = (old_serial or "").strip()
|
||
new_serial = (new_serial or "").strip()
|
||
if not old_serial or not new_serial:
|
||
return False, "", "缺少参数"
|
||
if old_serial == new_serial:
|
||
return False, "", "新旧地址相同，无需迁移"
|
||
with _ctx():
|
||
old = Device.query.get(old_serial)
|
||
if not old:
|
||
return False, "", f"设备池中没有 {old_serial}"
|
||
if Device.query.get(new_serial) is not None:
|
||
return False, "", f"{new_serial} 已在设备池中，请先处理那条记录"
|
||
_, name = _move_device_row(old, new_serial, fingerprint)
|
||
_log.info(f"设备人工认领: 『{name or old_serial}』{old_serial} → {new_serial}")
|
||
return True, name, f"『{name or old_serial}』已迁到 {new_serial}，分组/任务引用已同步"
|
||
|
||
|
||
# ================== 管理（CRUD） ==================
|
||
def add_device(serial, name="", note="", enabled=True, fingerprint=""):
|
||
"""添加/更新设备（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
|
||
if fingerprint:
|
||
d.fingerprint = fingerprint
|
||
db.session.commit()
|
||
return False
|
||
db.session.add(Device(serial=serial, name=name or "", note=note or "",
|
||
enabled=enabled, fingerprint=fingerprint or "",
|
||
created_at=time.strftime("%Y-%m-%d %H:%M")))
|
||
db.session.commit()
|
||
_log.info(f"设备池新增设备: {serial}")
|
||
return True
|
||
|
||
|
||
def rename_device(serial, name):
|
||
"""重命名设备（唯一性由调用方先校验）。返回是否成功。"""
|
||
name = (name or "").strip()
|
||
with _ctx():
|
||
d = Device.query.get(serial)
|
||
if not d or not name or (d.name or "") == name:
|
||
return False
|
||
old = d.name or ""
|
||
d.name = name
|
||
db.session.commit()
|
||
_log.info(f"设备池重命名: {old or serial} → {name}（{serial}）")
|
||
return True
|
||
|
||
|
||
def set_fingerprint(serial, fingerprint):
|
||
"""补写设备指纹（认领需要；采集是后来才做的）。"""
|
||
fingerprint = (fingerprint or "").strip()
|
||
if not fingerprint:
|
||
return False
|
||
with _ctx():
|
||
d = Device.query.get(serial)
|
||
if not d or d.fingerprint == fingerprint:
|
||
return False
|
||
d.fingerprint = fingerprint
|
||
db.session.commit()
|
||
_log.info(f"设备池采集指纹: {serial} -> {fingerprint}")
|
||
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_fingerprint(serial, timeout=6):
|
||
"""读取并写回设备指纹，返回指纹（失败返回空串）。"""
|
||
fp = read_fingerprint(serial, timeout=timeout)
|
||
if fp:
|
||
set_fingerprint(serial, fp)
|
||
return fp
|
||
|
||
|
||
def refresh_info(serial):
|
||
"""采集一台设备的型号 + 指纹（指纹缺失时才读，避免每次启动都白跑一次 adb）。
|
||
|
||
返回 (model, fingerprint)。任一失败都静默——设备池展示用，不影响主流程。
|
||
"""
|
||
model = refresh_model(serial)
|
||
fp = ""
|
||
with _ctx():
|
||
row = Device.query.get(serial)
|
||
fp = (row.fingerprint or "") if row else ""
|
||
if not fp:
|
||
fp = refresh_fingerprint(serial)
|
||
return model, fp
|
||
|
||
|
||
def refresh_all_models():
|
||
"""批量采集池内在线设备的型号（顺带补齐缺失的设备指纹），并发 10，后台线程调用。
|
||
|
||
返回成功数。指纹补齐很重要：老库里的设备没有指纹，补上之后换 IP 才能被认领。
|
||
"""
|
||
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_info, s): s for s in serials}
|
||
for fut in as_completed(futures):
|
||
try:
|
||
model, _fp = fut.result()
|
||
if model:
|
||
ok += 1
|
||
except Exception:
|
||
pass
|
||
_log.info(f"设备池批量采集型号完成: {ok}/{len(serials)} 台")
|
||
return ok
|