butubb
8fdde978ea
用户反馈：① 元素抓取的选择设备列表显示的是 IP；② 所有 webhook 通知里都是 IP；
要都显示设备名。
- 通知链路统一补名字（调用点不用动）：`core/notifier.py`
· 新增 `set_device_name_resolver()` + `fill_device_names()`，在 `notify()` 与
`build_message()`（预览/测试发送也走它）里把 `serial` 补成 `device_name`、
把 `serials`/`devices` 列表逐项换成名字。
· 补一处就全带名字了——标题主体、字段表、聚合样本认的都是 `device_name`。
· **做成"纯内存回调"是刻意的**：notify() 的硬红线是零 DB，不能为了取个名字去查库
（那等于在业务线程里加一次阻塞查询）。
· 降级规则：调用点自己传了 device_name 就用它的；查不到名字（没命名/不在池里）
保留原地址；解析器缺失或抛异常都只是降级，绝不影响发送。
- `core/device_pool.py`：维护 `serial → 名称` 内存快照——`init_app` 同步刷一次、
增删改名/迁址后各刷一次（改名立刻生效）、`device-names` 线程每 60s 兜底刷一次
（覆盖整库恢复这类进程外改动）。`name_of()` 只读内存，可在通知路径上安全调用。
- `web_server.py`：装配层接上 `notifier.set_device_name_resolver(device_pool.name_of)`。
- 元素抓取/测试此步骤的设备列表：`GET /api/uiauto/devices` 的 `name` 改用**平台名**，
前端 `editor.js` 新增共用的 `_devCard()`——名字做主标题（粗体），
`型号 · 地址` 作副标题。
· 池外设备**退回地址而不是 uiautodev 的 name**：实测那份 name 是设备 codename
（一柜子机器全叫 "earth"），拿它认设备等于没名字，地址至少唯一。
· 「测试此步骤」的设备列表原来连状态角标都没有，一并统一成同一个卡片。
- 顺带修一个**既有 bug（不是本次需求）**：`/locate` 设备端定位页从 ce47a5b 那次
web 蓝图拆分起就一直 500——拆分时漏掉了 `render_template_string` 与
`markupsafe.escape as _esc` 两个 import。后者不只是缺个名字：定位页把 query 里的
serial 拼进 HTML，而 `render_template_string` 的模板名是 "<template>"、
**不会自动转义**，所以那还是个反射 XSS。已显式转义（已用 `<img onerror>` 验证）。
自测：device_pool 快照/改名即时生效、fill_device_names 全分支（含列表、
未命名保留地址、不在池保留地址、解析器缺失/抛异常降级）、真消息渲染断言
**通篇不含 IP**；GET 路由冒烟 54 个 0 个 500；前端语法 + 卡片渲染截图核对。
456 lines
17 KiB
Python
456 lines
17 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
|
||
refresh_names() # 同步刷一次名称快照，让通知从第一条起就有名字
|
||
try:
|
||
t = threading.Thread(target=_refresh_models_bg, daemon=True)
|
||
t.start()
|
||
t2 = threading.Thread(target=_names_loop, name="device-names", daemon=True)
|
||
t2.start()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# ================== 名称内存快照（给通知用） ==================
|
||
# 为什么单独存一份：webhook 通知里要显示"是哪台设备"（A08）而不是地址（IP），
|
||
# 但 `notifier.notify()` 有一条硬红线——**零 DB 访问**（不能为了取个名字去查库，
|
||
# 更不能在业务线程里阻塞）。所以由这里维护一份 serial→名称 的内存快照：
|
||
# 启动刷一次、池子有变动时刷一次、再兜底每 60s 刷一次（覆盖整库恢复等外部改动）。
|
||
_names = {}
|
||
_names_lock = threading.Lock()
|
||
_NAMES_INTERVAL = 60
|
||
|
||
|
||
def refresh_names():
|
||
"""把设备池的 serial→名称刷进内存快照（只查库，不碰 adb）。返回条数。"""
|
||
try:
|
||
rows = list_devices()
|
||
except Exception as e:
|
||
_log.debug(f"刷新设备名快照失败: {e}")
|
||
return 0
|
||
m = {}
|
||
for d in rows:
|
||
s = d.get("serial") or ""
|
||
if s:
|
||
m[s] = d.get("name") or ""
|
||
with _names_lock:
|
||
_names.clear()
|
||
_names.update(m)
|
||
return len(m)
|
||
|
||
|
||
def name_of(serial):
|
||
"""serial 对应的设备名称（**纯内存，可在通知路径上调用**）。没有则返回 ""。"""
|
||
if not serial:
|
||
return ""
|
||
with _names_lock:
|
||
return _names.get(str(serial), "")
|
||
|
||
|
||
def _names_loop():
|
||
"""兜底定时刷新名称快照（覆盖整库恢复这类进程外改动）。"""
|
||
while True:
|
||
time.sleep(_NAMES_INTERVAL)
|
||
refresh_names()
|
||
|
||
|
||
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()
|
||
refresh_names() # 键变了（serial → 名称的映射也跟着变）
|
||
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()
|
||
refresh_names() # 名字可能改了：通知里要立刻用新名字
|
||
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()
|
||
refresh_names()
|
||
_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()
|
||
refresh_names()
|
||
_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()
|
||
refresh_names()
|
||
_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
|