341 lines
13 KiB
Python
341 lines
13 KiB
Python
"""设备自动发现：扫描网段中开放 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():
|
||
"""待连接列表：只返回当前在线的设备（离线候选不可确认，不展示）。
|
||
|
||
后端直接过滤（2026-09-04）：不依赖前端 JS 版本，任何客户端都拿不到
|
||
离线条目；设备恢复在线后扫描自动更新 last_seen 并重新出现在列表。
|
||
"""
|
||
online = _parse_adb_devices(_adb("devices"))
|
||
with _ctx():
|
||
rows = [p.to_dict() for p in PendingDevice.query.order_by(
|
||
PendingDevice.first_seen.desc()).all()]
|
||
return [dict(r, online=True) for r in rows if r["serial"] in online]
|
||
|
||
|
||
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, "已忽略"
|