feat: 设备自动发现——定时扫描局域网+Tailscale 网段开放 adb 5555 的设备进入待连接池，用户确认后才加入正式设备池（不自动连接）；socket 并发探测+adb 短超时验证（state=device 过滤 unauthorized），配置存 app_meta 可前端调整；配套：adb_connect_light 验证短超时、DISABLE_SCHEDULER 回归开关
This commit is contained in:
@@ -0,0 +1,338 @@
|
||||
"""设备自动发现：扫描网段中开放 adb 5555 的设备 → 待连接池（pending）。
|
||||
|
||||
流程：socket 并发探测 5555（标准库，不碰 adb）→ 剔除已在设备池的 serial →
|
||||
串行 adb 验证（adb_connect_light，内部全局锁）→ 用 `adb devices` 的
|
||||
state=="device" 过滤（排除 unauthorized/offline，connect 输出不可信）→
|
||||
写入 pending_device 表。**扫描只验证、不自动连接入池**——用户在前端确认后
|
||||
才调 device_pool.add_device + adb_connect（见 web/devices_api.py）。
|
||||
|
||||
安全红线：绝不 kill-server / 绝不 disconnect（与全项目一致）。
|
||||
|
||||
配置存 app_meta KV 表（discovery_enabled/subnets/interval/port），定时线程
|
||||
每轮重读，开关/周期/网段改动即时生效无需重启。
|
||||
"""
|
||||
import ipaddress
|
||||
import json
|
||||
import socket
|
||||
import threading
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from datetime import datetime
|
||||
|
||||
from config import (USB_ADB_HOST,
|
||||
DISCOVERY_PORT, DISCOVERY_SUBNETS, DISCOVERY_INTERVAL)
|
||||
from core.adb_helper import _adb, adb_connect_light
|
||||
from core.logger import get_logger
|
||||
from core.models import db, PendingDevice
|
||||
|
||||
_log = get_logger("core.disc")
|
||||
|
||||
# 扫描配置键（app_meta）
|
||||
_K_ENABLED = "discovery_enabled"
|
||||
_K_SUBNETS = "discovery_subnets"
|
||||
_K_INTERVAL = "discovery_interval"
|
||||
_K_PORT = "discovery_port"
|
||||
|
||||
# 单网段主机数上限（防误配 /8 之类），超过截断并告警
|
||||
_MAX_HOSTS_PER_SUBNET = 1024
|
||||
# socket 探测并发数与单 IP 超时
|
||||
_PROBE_WORKERS = 64
|
||||
_PROBE_TIMEOUT = 0.4
|
||||
|
||||
_app = None
|
||||
_scan_lock = threading.Lock() # 定时/手动扫描互斥
|
||||
_stop_event = threading.Event() # shutdown 用
|
||||
_scanning = False # 状态快照（API/前端）
|
||||
_last_scan = None # (时间串, 开放数, 可连数, 新增数)
|
||||
_last_error = ""
|
||||
|
||||
|
||||
def _ctx():
|
||||
"""后台线程访问 db 时自行推 app context（仿 device_pool._ctx）。"""
|
||||
return _app.app_context() if _app else None
|
||||
|
||||
|
||||
def _fmt(ts=None):
|
||||
return (ts or datetime.now()).strftime("%Y-%m-%d %H:%M")
|
||||
|
||||
|
||||
# ================== 配置读写 ==================
|
||||
def get_settings():
|
||||
"""读发现配置（app_meta，缺键用 config 默认值补齐）。"""
|
||||
with _ctx():
|
||||
def _get(key, default):
|
||||
v = db.session.execute(
|
||||
db.text("SELECT value FROM app_meta WHERE key=:k"), {"k": key}).scalar()
|
||||
return v if v is not None else default
|
||||
try:
|
||||
subnets = json.loads(_get(_K_SUBNETS, "[]")) or DISCOVERY_SUBNETS
|
||||
except Exception:
|
||||
subnets = DISCOVERY_SUBNETS
|
||||
return {
|
||||
"enabled": _get(_K_ENABLED, "1") == "1", # 默认开启
|
||||
"subnets": subnets,
|
||||
"interval": int(_get(_K_INTERVAL, str(DISCOVERY_INTERVAL)) or DISCOVERY_INTERVAL),
|
||||
"port": int(_get(_K_PORT, str(DISCOVERY_PORT)) or DISCOVERY_PORT),
|
||||
}
|
||||
|
||||
|
||||
def save_settings(enabled=None, subnets=None, interval=None, port=None):
|
||||
"""写发现配置（部分字段更新）。subnets 逐项校验 CIDR，非法返回 (False, 原因)。"""
|
||||
if subnets is not None:
|
||||
clean = []
|
||||
for s in subnets:
|
||||
s = (s or "").strip()
|
||||
if not s:
|
||||
continue
|
||||
try:
|
||||
ipaddress.ip_network(s, strict=False)
|
||||
except ValueError:
|
||||
return False, f"网段格式错误: {s}"
|
||||
clean.append(s)
|
||||
subnets = clean
|
||||
try:
|
||||
if interval is not None and (int(interval) < 10 or int(interval) > 3600):
|
||||
return False, "扫描周期需在 10-3600 秒之间"
|
||||
if port is not None and (int(port) < 1 or int(port) > 65535):
|
||||
return False, "端口不合法"
|
||||
except (TypeError, ValueError):
|
||||
return False, "参数不合法"
|
||||
with _ctx():
|
||||
def _put(key, value):
|
||||
db.session.execute(
|
||||
db.text("INSERT OR REPLACE INTO app_meta(key,value) VALUES(:k,:v)"),
|
||||
{"k": key, "v": str(value)})
|
||||
if enabled is not None:
|
||||
_put(_K_ENABLED, "1" if enabled else "0")
|
||||
if subnets is not None:
|
||||
_put(_K_SUBNETS, json.dumps(subnets))
|
||||
if interval is not None:
|
||||
_put(_K_INTERVAL, str(int(interval)))
|
||||
if port is not None:
|
||||
_put(_K_PORT, str(int(port)))
|
||||
db.session.commit()
|
||||
return True, "已保存"
|
||||
|
||||
|
||||
# ================== 网段展开与端口探测 ==================
|
||||
def _expand_subnets(subnets, max_hosts=_MAX_HOSTS_PER_SUBNET):
|
||||
"""CIDR 列表 → IP 列表。排除 220 自身（USB_ADB_HOST）与本机 IP；
|
||||
非法网段跳过记日志；单网段超过 max_hosts 截断并告警。"""
|
||||
self_ips = {USB_ADB_HOST}
|
||||
try: # 本机自身 IP（接口枚举，尽力而为；无 hostname -I 的平台跳过）
|
||||
import subprocess
|
||||
out = subprocess.run(["hostname", "-I"], capture_output=True, text=True,
|
||||
timeout=3).stdout or ""
|
||||
self_ips.update(p for p in out.split() if p)
|
||||
except Exception:
|
||||
pass
|
||||
ips = []
|
||||
for cidr in subnets:
|
||||
try:
|
||||
net = ipaddress.ip_network((cidr or "").strip(), strict=False)
|
||||
except ValueError:
|
||||
_log.warning(f"发现: 跳过非法网段 {cidr}")
|
||||
continue
|
||||
hosts = [str(h) for h in net.hosts()
|
||||
if str(h) not in self_ips]
|
||||
if len(hosts) > max_hosts:
|
||||
_log.warning(f"发现: 网段 {cidr} 主机数 {len(hosts)} 超过上限，截断前 {max_hosts}")
|
||||
hosts = hosts[:max_hosts]
|
||||
ips.extend(hosts)
|
||||
return ips
|
||||
|
||||
|
||||
def _probe_port(ip, port, timeout=_PROBE_TIMEOUT):
|
||||
"""单 IP 端口探测（标准库 socket，超时/拒绝/不可达一律 False）。"""
|
||||
try:
|
||||
with socket.create_connection((ip, port), timeout=timeout):
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def _probe_open(ips, port, workers=_PROBE_WORKERS):
|
||||
"""并发探测，返回开放端口的主机列表。"""
|
||||
if not ips:
|
||||
return []
|
||||
with ThreadPoolExecutor(max_workers=workers) as ex:
|
||||
results = list(ex.map(lambda ip: (ip, _probe_port(ip, port)), ips))
|
||||
return [ip for ip, ok in results if ok]
|
||||
|
||||
|
||||
# ================== adb devices 解析 ==================
|
||||
def _parse_adb_devices(out):
|
||||
"""解析 `adb devices` 输出 → state=="device" 的 serial 集合。
|
||||
|
||||
只看 connect 输出不可靠（未授权设备也返回 connected），必须用 state 过滤。
|
||||
"""
|
||||
devices = set()
|
||||
for line in (out or "").splitlines()[1:]:
|
||||
parts = line.split()
|
||||
if len(parts) >= 2 and parts[0] and not parts[0].startswith("*"):
|
||||
if parts[1] == "device":
|
||||
devices.add(parts[0])
|
||||
return devices
|
||||
|
||||
|
||||
# ================== 扫描 ==================
|
||||
def scan_once(manual=False):
|
||||
"""执行一轮扫描。返回 (ok, result_dict)；后台线程/API 调用。"""
|
||||
global _scanning, _last_scan, _last_error
|
||||
if not _scan_lock.acquire(blocking=False):
|
||||
return False, {"error": "扫描进行中"}
|
||||
_scanning = True
|
||||
try:
|
||||
settings = get_settings()
|
||||
port = settings["port"]
|
||||
subnets = settings["subnets"]
|
||||
if not subnets:
|
||||
_last_error = "未配置扫描网段"
|
||||
return True, {"found": 0, "verified": 0, "new": 0, "error": _last_error}
|
||||
|
||||
# 1. socket 并发探测（不碰 adb，把几百台缩小到开放端口量级）
|
||||
open_ips = _probe_open(_expand_subnets(subnets), port)
|
||||
candidates = {f"{ip}:{port}" for ip in open_ips}
|
||||
|
||||
with _ctx():
|
||||
# 2. 剔除已在正式设备池的 serial
|
||||
configured = set(device_pool_list_configured())
|
||||
candidates -= configured
|
||||
|
||||
# 3. 串行 adb 验证：已连接的直接跳过（零成本），其余 adb_connect_light
|
||||
already = _parse_adb_devices(_adb("devices"))
|
||||
for serial in candidates - already:
|
||||
adb_connect_light(serial) # 内部全局锁串行；失败无妨，下面按 state 过滤
|
||||
|
||||
# 4. state==device 过滤（排除 unauthorized/offline）
|
||||
verified = _parse_adb_devices(_adb("devices")) & candidates
|
||||
|
||||
# 5. 写 pending：新 → 插入；已有 → 更新 last_seen
|
||||
now = _fmt()
|
||||
existing = {p.serial for p in PendingDevice.query.all()}
|
||||
added = 0
|
||||
for serial in verified:
|
||||
source = "tailscale" if serial.split(":")[0].startswith("100.") else "lan"
|
||||
if serial in existing:
|
||||
PendingDevice.query.filter_by(serial=serial).update(
|
||||
{"last_seen": now})
|
||||
else:
|
||||
db.session.add(PendingDevice(serial=serial, source=source,
|
||||
first_seen=now, last_seen=now))
|
||||
added += 1
|
||||
db.session.commit()
|
||||
_log.info(f"发现: 探测开放 {len(open_ips)} 台，可连 {len(verified)} 台，"
|
||||
f"新增待连接 {added} 台")
|
||||
_last_scan = (now, len(open_ips), len(verified), added)
|
||||
_last_error = ""
|
||||
return True, {"found": len(open_ips), "verified": len(verified), "new": added}
|
||||
except Exception as e:
|
||||
_log.warning(f"发现扫描异常: {e}")
|
||||
_last_error = str(e)[:200]
|
||||
return False, {"error": _last_error}
|
||||
finally:
|
||||
_scanning = False
|
||||
_scan_lock.release()
|
||||
|
||||
|
||||
def device_pool_list_configured():
|
||||
"""设备池已配置 serial（延迟 import 避免循环依赖）。"""
|
||||
from core import device_pool
|
||||
return device_pool.list_configured()
|
||||
|
||||
|
||||
# ================== 定时扫描线程 ==================
|
||||
def _discovery_loop():
|
||||
"""定时扫描 daemon 线程。每轮重读配置（开关/周期即时生效）。
|
||||
|
||||
启动先 sleep 15s 避让 web_server 预连接线程（两者都抢 adb 全局锁）。
|
||||
等待用 30s 切片（_stop_event.wait），关停/改周期 ≤30s 生效。
|
||||
"""
|
||||
try:
|
||||
_stop_event.wait(15)
|
||||
while not _stop_event.is_set():
|
||||
try:
|
||||
settings = get_settings()
|
||||
if settings["enabled"]:
|
||||
scan_once()
|
||||
except Exception as e:
|
||||
_log.warning(f"发现定时扫描异常: {e}")
|
||||
wait_s = max(10, settings.get("interval", 60))
|
||||
waited = 0
|
||||
while waited < wait_s and not _stop_event.is_set():
|
||||
_stop_event.wait(min(30, wait_s - waited))
|
||||
waited += min(30, wait_s - waited)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def init_app(app):
|
||||
"""web_server 启动时调用：绑 app + 起定时扫描 daemon 线程。"""
|
||||
global _app
|
||||
_app = app
|
||||
_stop_event.clear()
|
||||
t = threading.Thread(target=_discovery_loop, name="device-discovery", daemon=True)
|
||||
t.start()
|
||||
_log.info("设备自动发现线程已启动（默认 60s 扫描一次）")
|
||||
|
||||
|
||||
def shutdown():
|
||||
"""优雅退出：置 stop_event，等待中的循环在切片边界退出。"""
|
||||
_stop_event.set()
|
||||
|
||||
|
||||
# ================== 对外查询与确认 ==================
|
||||
def get_status():
|
||||
"""API 状态快照：配置 + 扫描状态 + 最近一轮结果 + 待连接数量。"""
|
||||
settings = get_settings()
|
||||
with _ctx():
|
||||
pending = PendingDevice.query.count()
|
||||
return {
|
||||
"enabled": settings["enabled"],
|
||||
"subnets": settings["subnets"],
|
||||
"interval": settings["interval"],
|
||||
"port": settings["port"],
|
||||
"scanning": _scanning,
|
||||
"last_scan": _last_scan[0] if _last_scan else "",
|
||||
"last_result": ({"found": _last_scan[1], "verified": _last_scan[2],
|
||||
"new": _last_scan[3]} if _last_scan else None),
|
||||
"last_error": _last_error,
|
||||
"pending_count": pending,
|
||||
}
|
||||
|
||||
|
||||
def list_pending():
|
||||
"""待连接列表（含当前在线状态：本机 adb 一次解析）。"""
|
||||
online = _parse_adb_devices(_adb("devices"))
|
||||
with _ctx():
|
||||
rows = [p.to_dict() for p in PendingDevice.query.order_by(
|
||||
PendingDevice.first_seen.desc()).all()]
|
||||
for r in rows:
|
||||
r["online"] = r["serial"] in online
|
||||
return rows
|
||||
|
||||
|
||||
def confirm_pending(serial, name=""):
|
||||
"""确认连接：pending 行 → 正式设备池（add_device upsert）→ 删 pending。
|
||||
|
||||
返回 (ok, msg, is_new)。adb_connect + 采型号由 API 层后台线程做。
|
||||
"""
|
||||
with _ctx():
|
||||
row = PendingDevice.query.get(serial)
|
||||
if not row:
|
||||
return False, "设备不在待连接列表", False
|
||||
from core import device_pool
|
||||
is_new = device_pool.add_device(serial, name=name or "")
|
||||
PendingDevice.query.filter_by(serial=serial).delete()
|
||||
db.session.commit()
|
||||
return True, "已加入设备池" + ("" if is_new else "（已存在，信息已更新）"), is_new
|
||||
|
||||
|
||||
def ignore_pending(serial):
|
||||
"""忽略：删除 pending 行（下轮扫描可能再次发现）。"""
|
||||
with _ctx():
|
||||
row = PendingDevice.query.get(serial)
|
||||
if not row:
|
||||
return False, "设备不在待连接列表"
|
||||
PendingDevice.query.filter_by(serial=serial).delete()
|
||||
db.session.commit()
|
||||
return True, "已忽略"
|
||||
Reference in New Issue
Block a user