阶段3: 摘除 STF——删除 stf_client/stf_device_mgmt 与全部引用(STF 容器重启/设备管理/强制释放占用/agent 按钮/占用列),调度与设备管理完全基于本地设备池+adb;错误类迁入 device_worker,create_worker 去 stf 参数

This commit is contained in:
2026-08-18 08:08:32 +08:00
parent 98244d8161
commit a5ce57b6fd
17 changed files with 71 additions and 956 deletions
+9 -236
View File
@@ -33,7 +33,6 @@ from markupsafe import escape as _esc
from flask_login import (LoginManager, login_user, logout_user,
login_required, current_user)
from core.stf_client import STFClient
from core.task_manager import TaskManager
from core.device_worker import clear_worker_error, clear_all_worker_errors, get_all_worker_status
from core.logger import get_logger, _LOG_DIR, _MODULE_FILES
@@ -41,11 +40,9 @@ from core.models import db, init_db, User, DeviceGroup, TaskJob, CustomAction
from core.apk_manager import ApkManager
from core.adb_helper import (screenshot, list_installed_apps, adb_connect_light,
adb_connect, _ADB_LOCK, _adb)
from core import uiauto_helper, tailscale_client, stf_device_mgmt, ssh_client
from core import uiauto_helper, tailscale_client
from core.tailscale_client import TailscaleError
from core.stf_device_mgmt import StfDevError
from core.ssh_client import SSHError
from config import ADB_PATH, STF_DOCKER_CONTAINER
from config import ADB_PATH
from tasks import list_task_types, get_task_class
_log = get_logger("web")
@@ -67,13 +64,12 @@ login_manager.login_view = "login"
# 先初始化数据库(含旧 JSON 迁移),再创建 TaskManager(需要 app context 读写 DB)
init_db(app)
stf = STFClient()
# 设备池(本地清单 + adb 在线状态):阶段 0 起作为 STF 的替代数据源,
# 首次启动自动从 STF 导入现有设备;阶段 3 摘除 STF 后此模块独立运行
from core import device_pool
device_pool.init_app(app)
mgr = TaskManager(stf, app=app)
apk_mgr = ApkManager(stf, app=app)
mgr = TaskManager(app=app)
apk_mgr = ApkManager(app=app)
@login_manager.user_loader
@@ -232,7 +228,6 @@ def api_health():
if d.get("worker_status") in ("running", "connecting")),
"device_error": sum(1 for d in devs
if d.get("worker_status") in ("error", "failed")),
"occupied": sum(1 for d in devs if d.get("stf_occupied")),
"jobs": len(mgr.jobs),
})
except Exception as e:
@@ -738,18 +733,6 @@ def api_stop_all():
return jsonify({"ok": True, "stopped": stopped})
@app.route("/api/release", methods=["POST"])
@perm_required(PERM_DEVICES)
def api_release():
# 阶段 1:调度已不占 STF,此处仅清理迁移期遗留的 STF 占用记录
# (阶段 3 摘除 STF 后此接口改为清理本实例 _running/worker 状态)
try:
released, failed = stf.release_all_mine()
except Exception:
released, failed = [], []
return jsonify({"ok": True, "released": released, "failed": failed})
@app.route("/api/device/clear_error", methods=["POST"])
@perm_required(PERM_DEVICES)
def api_device_clear_error():
@@ -899,7 +882,7 @@ def api_steps_test():
"""测试单个步骤:在指定设备上试执行,验证选择器是否命中(步骤编辑器"测试此步骤")。
请求: {"serial": "100.100.10.11:5555", "step": {"type": "click", "params": {...}}}
只读连接(adb connect + u2),不占用/释放 STF,与运行中任务互不干扰。
只读连接(adb connect + u2),与运行中任务互不干扰。
返回 result: "命中" / "未找到" / "已执行"。
"""
data = request.json or {}
@@ -1208,23 +1191,15 @@ def _merged_device_list():
states[serial] = "pool"
except Exception:
pass
# 迁移期:STF 里 adb 可达但 agent 未就绪的设备仍列出,供维护终端修复 agent
try:
for d in stf.list_all_devices():
serial = d.get("serial", "")
if serial and d.get("present") and not d.get("ready") and serial not in states:
states[serial] = "stf_not_ready"
except Exception:
pass
return [{"serial": s, "state": st} for s, st in sorted(states.items())]
@app.route("/api/adb/devices")
@admin_required
def api_adb_devices():
"""维护终端设备列表(仅管理员):本地 adb 已连接 + STF 在线设备池
(含 agent 未就绪的待修复设备,state=stf_not_ready)。
"""维护终端设备列表(仅管理员):本地 adb 已连接 + 设备池(SQLite)。
状态:device/offline=本机 adb 实际状态;pool=设备池已配置但本机未连接。
供终端设备选择器使用——选中后自动附加 `-s <serial>`,
离线/未连接的设备也可选,配合"重连设备"按钮恢复。
"""
@@ -1272,7 +1247,7 @@ def api_tools_clipboard_set():
"""工具-剪贴板注入:把指定文字写入一台或多台设备的剪贴板。
请求: {"serials": ["100.100.10.11:5555", ...], "text": "要注入的文字"}
设备来源与维护终端一致:本地 adb 已连接设备(含 USB)+ STF 在线池。
设备来源与维护终端一致:本地 adb 已连接设备(含 USB)+ 设备池。
实现:u2 jsonrpc.setClipboard(实测 cmd clipboard 在 MIUI 上不存在)。
"""
data = request.json or {}
@@ -1319,152 +1294,6 @@ def _set_device_clipboard(serial, text):
return False, f"{type(e).__name__}: {str(e)[:120]}"
@app.route("/api/stf/devmgmt/status")
@admin_required
def api_stf_devmgmt_status():
"""STF 设备管理:脚本配置的 IP + 220 adb 实际连接状态。"""
try:
st = stf_device_mgmt.status()
return jsonify({"ok": True, **st})
except StfDevError as e:
return jsonify({"ok": False, "error": str(e)}), 502
@app.route("/api/stf/devmgmt/add", methods=["POST"])
@admin_required
def api_stf_devmgmt_add():
"""STF 设备管理:添加设备(写入脚本 + 立即 connect)。
迁移期桥接:同步加入本地设备池(SQLite),保证新设备可被调度。
阶段 3 摘除 STF 后此接口随设备池管理面板退役。
"""
ip = (request.json or {}).get("ip", "").strip()
if not ip:
return jsonify({"ok": False, "error": "请输入设备 IP"}), 400
try:
msgs = stf_device_mgmt.add_device(ip)
except StfDevError as e:
return jsonify({"ok": False, "error": str(e)}), 502
# 桥接:同步本地设备池(serial 统一 IP:5555)
try:
device_pool.add_device(f"{ip}:5555" if ":" not in ip else ip)
except Exception as e:
_log.warning(f"STF 添加设备桥接设备池失败: {e}")
return jsonify({"ok": True, "msgs": msgs})
@app.route("/api/stf/devmgmt/remove", methods=["POST"])
@admin_required
def api_stf_devmgmt_remove():
"""STF 设备管理:移除设备(脚本删除 + disconnect)。
迁移期桥接:同步从本地设备池删除。阶段 3 此接口退役。
"""
ip = (request.json or {}).get("ip", "").strip()
if not ip:
return jsonify({"ok": False, "error": "请输入设备 IP"}), 400
try:
msgs = stf_device_mgmt.remove_device(ip)
except StfDevError as e:
return jsonify({"ok": False, "error": str(e)}), 502
try:
device_pool.remove_device(f"{ip}:5555" if ":" not in ip else ip)
except Exception:
pass
return jsonify({"ok": True, "msgs": msgs})
def _app_ver_on_device(serial, pkg):
"""单设备包版本查询:pm path 检查安装 → dumpsys 取 versionName/versionCode。
IP:port 设备先 adb connect(已连接自动跳过;绝不 disconnect,红线)。
查询是只读 shell 命令,无需 adb server 锁(锁只保护 connect/kill-server 类操作)。
"""
if ":" in serial:
try:
adb_connect(serial)
except Exception as e:
return {"installed": False, "error": f"adb 连接失败: {e}"}
try:
r = subprocess.run([ADB_PATH, "-s", serial, "shell", "pm", "path", pkg],
capture_output=True, timeout=15)
if b"package:" not in (r.stdout or b""):
return {"installed": False, "version_name": "", "version_code": ""}
r2 = subprocess.run([ADB_PATH, "-s", serial, "shell", "dumpsys", "package", pkg],
capture_output=True, timeout=20)
out2 = (r2.stdout or b"").decode("utf-8", errors="replace")
vm = re.search(r"versionName=(\S+)", out2)
vc = re.search(r"versionCode=(\d+)", out2)
return {"installed": True,
"version_name": vm.group(1) if vm else "",
"version_code": vc.group(1) if vc else ""}
except subprocess.TimeoutExpired:
return {"installed": False, "error": "查询超时"}
except Exception as e:
return {"installed": False, "error": str(e)}
@app.route("/api/tools/appver", methods=["POST"])
@admin_required
def api_tools_appver():
"""工具-应用版本管理:查询所有设备上指定包名的安装情况与版本号。
请求: {"pkg": "com.ss.android.ugc.aweme"}
设备来源:本地 adb(含 USB)+ STF 在线池,并发查询(最多 10 台同时)。
"""
pkg = (request.json or {}).get("pkg", "").strip()
if not re.match(r"^[A-Za-z0-9_.]+$", pkg or ""):
return jsonify({"ok": False, "error": "包名格式不正确(仅字母/数字/._)"}), 400
devices = _merged_device_list()
if not devices:
return jsonify({"ok": False, "error": "无可用设备"}), 404
results = {}
with ThreadPoolExecutor(max_workers=min(10, len(devices))) as pool:
futures = {pool.submit(_app_ver_on_device, d["serial"], pkg): d for d in devices}
for fut in as_completed(futures, timeout=90):
d = futures[fut]
try:
results[d["serial"]] = fut.result()
except Exception as e:
results[d["serial"]] = {"installed": False, "error": str(e)}
fail = sum(1 for v in results.values() if v.get("error"))
return jsonify({"ok": True, "results": results, "fail": fail,
"total": len(results)})
@app.route("/api/stf/devmgmt/delete", methods=["POST"])
@admin_required
def api_stf_devmgmt_delete():
"""STF 设备管理:彻底删除(脚本移除 + disconnect + 清 STF 池记录)。
迁移期桥接:同步从本地设备池删除。阶段 3 此接口退役。
"""
ip = (request.json or {}).get("ip", "").strip()
if not ip:
return jsonify({"ok": False, "error": "请输入设备 IP"}), 400
try:
msgs = stf_device_mgmt.delete_device(ip, stf=stf)
except StfDevError as e:
return jsonify({"ok": False, "error": str(e)}), 502
try:
device_pool.remove_device(f"{ip}:5555" if ":" not in ip else ip)
except Exception:
pass
return jsonify({"ok": True, "msgs": msgs})
@app.route("/api/stf/devmgmt/reconnect", methods=["POST"])
@admin_required
def api_stf_devmgmt_reconnect():
"""STF 设备管理:一键重连(手动运行 220 的 connect_devices.sh)。"""
try:
ok, summary = stf_device_mgmt.reconnect_all()
return jsonify({"ok": True, "output": summary})
except StfDevError as e:
return jsonify({"ok": False, "error": str(e)}), 502
# ================== API:设备池管理(本地 SQLite 清单,新增设备的规范化入口) ==================
@app.route("/api/devices/pool", methods=["GET"])
@perm_required(PERM_DEVICES)
def api_devices_pool_list():
@@ -1553,46 +1382,6 @@ def api_devices_pool_reconnect():
return jsonify({"ok": True, "msg": "重连已启动(后台并发,约 10-20 秒)"})
@app.route("/api/stf/restart", methods=["POST"])
@admin_required
def api_stf_restart():
"""一键重启 STF Docker 容器(仅管理员)。
通过 SSH(config.STF_SSH_TARGET,默认 [email protected])执行
`docker restart <容器>`(config.STF_DOCKER_CONTAINER,默认 stf),
随后轮询 STF API 确认恢复。
影响:STF 约 30s 不可用,期间占用状态重置、任务中断——有任务在跑时拒绝执行。
"""
running = mgr.get_running()
if running:
return jsonify({"ok": False,
"error": f"有 {len(running)} 台设备正在运行任务,请先停止再维护"}), 409
try:
code, out, err = ssh_client.run(
f"docker restart {STF_DOCKER_CONTAINER} && "
f"docker ps --filter name={STF_DOCKER_CONTAINER} --format '{{{{.Names}}}}: {{{{.Status}}}}'",
timeout=60)
except SSHError as e:
return jsonify({"ok": False, "error": str(e)}), 502
if code != 0:
_log.error(f"STF 容器重启失败: {err or out}")
return jsonify({"ok": False,
"error": f"SSH 执行失败:{err or out}(请确认 SSH 配置(STF_SSH_TARGET / STF_SSH_PASSWORD)且容器名 {STF_DOCKER_CONTAINER} 正确)"}), 502
# 轮询 STF API 确认恢复(最长 30s)
alive = False
for _ in range(30):
try:
stf.list_all_devices()
alive = True
break
except Exception:
time.sleep(1)
_log.info(f"STF 容器 {STF_DOCKER_CONTAINER} 已重启,API 恢复={alive}")
return jsonify({"ok": True,
"msg": "STF 容器已重启" + ("" if alive else "(STF API 尚未恢复,请稍后刷新)"),
"detail": out.strip(), "api_alive": alive})
# ================== API:Tailscale 管理(仅管理员) ==================
# 通过 Tailscale 官方 API v2 管理 tailnet 设备(列表/改名/授权/密钥不过期/删除/生成 auth key)。
# 设备 IP 由 tailnet 自动分配,API 无法修改,列表只读展示。
@@ -1812,25 +1601,9 @@ def _run_server(host, port):
if __name__ == "__main__":
from config import WEB_HOST, WEB_PORT, AUTO_RELEASE_STALE_OCCUPY
from config import WEB_HOST, WEB_PORT
# 自动启动 uiautodev 服务(元素抓取功能依赖,端口 20242)
_ensure_uiauto_running()
# 启动时处理可能残留的 STF 占用(上次异常退出/强杀遗留)。
# AUTO_RELEASE_STALE_OCCUPY=True(单实例)时自动释放;否则仅提示。
# 多实例共用一个 STF 账户时不要开启,误放会中断另一实例的任务。
try:
_mine = stf.list_my_devices()
if _mine:
_serials = ", ".join(d["serial"] for d in _mine)
if AUTO_RELEASE_STALE_OCCUPY:
_log.warning(f"启动时发现 {len(_mine)} 台设备残留占用,自动释放(AUTO_RELEASE_STALE_OCCUPY=True):{_serials}")
_released, _failed = stf.release_all_mine()
_log.info(f"自动释放完成:成功 {len(_released)} 台,失败 {len(_failed)} 台")
else:
_log.warning(f"启动时发现 {len(_mine)} 台设备仍被本账户占用(可能异常退出残留):{_serials}。"
f"单实例无人值守可设 AUTO_RELEASE_STALE_OCCUPY=True 自动清理")
except Exception:
pass
_log.info(f"管理后台: http://localhost:{WEB_PORT}/ (admin/admin123)")
try:
_run_server(WEB_HOST, WEB_PORT)