添加设备数量限制。修复误杀iproxy逻辑。

This commit is contained in:
2025-11-04 20:33:09 +08:00
parent 81761a576b
commit 486c84efb6
9 changed files with 134 additions and 53 deletions

View File

@@ -129,6 +129,8 @@ class DeviceInfo:
self._last_seen: Dict[str, float] = {}
self._manager = FlaskSubprocessManager.get_instance()
self._iproxy_path = self._find_iproxy()
self._check_fail: Dict[str, int] = {}
self.MAX_DEVICES = 6
LogManager.info("DeviceInfo 初始化完成", udid="system")
print("[Init] DeviceInfo 初始化完成")
@@ -137,64 +139,81 @@ class DeviceInfo:
# =============== 核心端口连通性检测HTTP 方式) =================
def _is_local_port_open(self, port: int, udid: str, timeout: float = 5) -> bool:
"""
使用 HTTP 方式检测: http://127.0.0.1:port/ 发送一次 HEAD 请求
只要建立连接并收到合法的 HTTP 响应(任意 1xx~5xx 状态码),即认为 HTTP 可达。
遇到连接失败、超时、协议不对等异常,视为不可用
使用 HTTP 方式检测: http://127.0.0.1:port/status 发送 GET
1xx~5xx 任意状态码都视作“HTTP 可达WDA 常返回 200/404/401
✅ 超时改为默认 5 秒,更抗抖
"""
if not isinstance(port, int) or port <= 0 or port > 65535:
LogManager.error("端口不可用(非法端口号)", udid=udid)
return False
try:
# HEAD 更轻;若后端对 HEAD 不友好,可改为 "GET", "/"
conn = http.client.HTTPConnection("127.0.0.1", int(port), timeout=timeout)
conn.request("HEAD", "/")
conn.request("GET", "/status")
resp = conn.getresponse()
_ = resp.read(128)
status = resp.status
# 读到响应即可关闭
conn.close()
# 任何合法 HTTP 状态码都说明“HTTP 服务在监听且可交互”,包括 404/401/403/5xx
if 100 <= status <= 599:
return True
else:
LogManager.error(f"HTTP状态码异常: {status}", udid=udid)
return False
except Exception as e:
# 连接被拒绝、超时、不是HTTP协议正确响应比如返回了非HTTP的字节流都会到这里
LogManager.error(f"HTTP检测失败{e}", udid=udid)
return False
# =============== 一轮检查:发现不通就移除 =================
def check_iproxy_ports(self, connect_timeout: float = 3) -> None:
"""
周期性健康检查 iproxy -> WDA HTTP 可达性。
✅ 改为“连续失败 3 次才移除”,大幅降低抖动下的误删。
"""
# 给系统和 WDA 一点缓冲时间
time.sleep(20)
FAIL_THRESHOLD = 3 # 连续失败 N 次才视为离线
INTERVAL_SEC = 10 # 巡检间隔
while True:
snapshot = list(self._models.items()) # [(deviceId, DeviceModel), ...]
for device_id, model in snapshot:
try:
# 只处理在线且端口合法的设备
# 只处理在线的 iOStype==1
if model.type != 1:
continue
port = int(model.screenPort)
if port <= 0 or port > 65535:
continue
ok = self._is_local_port_open(port, timeout=connect_timeout, udid=device_id)
if not ok:
print(f"[iproxy-check] 端口不可连,移除设备 deviceId={device_id} port={port}")
ok = self._is_local_port_open(port, udid=device_id, timeout=connect_timeout)
if ok:
# 成功即清零失败计数
try:
self._remove_device(device_id) # 这里面可安全地改 self._models
self._check_fail[device_id] = 0
except Exception:
pass
# 可选:打印心跳日志过于刷屏,这里省略
continue
# 记录失败计数
cnt = self._check_fail.get(device_id, 0) + 1
self._check_fail[device_id] = cnt
print(f"[iproxy-check] FAIL #{cnt} deviceId={device_id} port={port}")
if cnt >= FAIL_THRESHOLD:
print(f"[iproxy-check] 连续失败{cnt}次,移除设备 deviceId={device_id} port={port}")
# 清掉计数并移除
self._check_fail.pop(device_id, None)
try:
self._remove_device(device_id)
except Exception as e:
print(f"[iproxy-check] _remove_device 异常 deviceId={device_id}: {e}")
else:
# 心跳日志按需开启,避免刷屏
# print(f"[iproxy-check] OK deviceId={device_id} port={port}")
pass
except Exception as e:
print(f"[iproxy-check] 单设备检查异常: {e}")
# 8秒间隔
time.sleep(10)
time.sleep(INTERVAL_SEC)
def listen(self):
LogManager.method_info("进入主循环", "listen", udid="system")
@@ -216,16 +235,27 @@ class DeviceInfo:
with self._lock:
known = set(self._models.keys())
current_online_count = sum(1 for m in self._models.values() if getattr(m, "type", 2) == 1)
for udid in online - known:
if (now - self._first_seen.get(udid, now)) >= self.ADD_STABLE_SEC:
print(f"[Add] 检测到新设备: {udid}")
try:
self._add_device(udid)
except Exception as e:
LogManager.method_error(f"新增失败:{e}", "listen", udid=udid)
print(f"[Add] 新增失败 {udid}: {e}")
# ==== 限流新增(最多 6 台)====
candidates = list(online - known)
stable_candidates = [udid for udid in candidates
if (now - self._first_seen.get(udid, now)) >= self.ADD_STABLE_SEC]
# 谁先出现谁优先
stable_candidates.sort(key=lambda u: self._first_seen.get(u, now))
capacity = max(0, self.MAX_DEVICES - current_online_count)
to_add = stable_candidates[:capacity] if capacity > 0 else []
for udid in to_add:
print(f"[Add] 检测到新设备: {udid}")
try:
self._add_device(udid)
except Exception as e:
LogManager.method_error(f"新增失败:{e}", "listen", udid=udid)
print(f"[Add] 新增失败 {udid}: {e}")
# ==== 处理离线 ====
for udid in list(known):
if udid in online:
continue
@@ -265,10 +295,19 @@ class DeviceInfo:
print(f"[WDA] /status@{local_port} 等待超时 {udid}")
return False
def _add_device(self, udid: str):
print(f"[Add] 开始新增设备 {udid}")
# ====== 上限保护:并发安全的首次检查 ======
with self._lock:
if udid in self._models:
print(f"[Add] 设备已存在,跳过 {udid}")
return
current_online_count = sum(1 for m in self._models.values() if getattr(m, "type", 2) == 1)
if current_online_count >= self.MAX_DEVICES:
print(f"[Add] 已达设备上限 {self.MAX_DEVICES},忽略新增 {udid}")
return
if not self._trusted(udid):
print(f"[Add] 未信任设备 {udid}, 跳过")
return
@@ -294,22 +333,25 @@ class DeviceInfo:
return
print(f"[WDA] WDA 就绪,准备获取屏幕信息 {udid}")
# 给 WDA 一点稳定时间,避免刚 ready 就查询卡住
time.sleep(0.5)
# 带超时的屏幕信息获取,避免卡死在 USBClient 调用里
w, h, s = self._screen_info_with_timeout(udid, timeout=3.5)
if not (w and h and s):
# 再做几次快速重试(带超时)
for i in range(4):
print(f"[Screen] 第{i + 1}次获取失败, 重试中... {udid}")
time.sleep(0.6)
w, h, s = self._screen_info_with_timeout(udid, timeout=3.5)
if w and h and s:
break
if not (w and h and s):
print(f"[Screen] 屏幕信息仍为空,继续添加 {udid}")
# ====== 上限保护:在分配端口前做二次检查(防并发越界)======
with self._lock:
current_online_count = sum(1 for m in self._models.values() if getattr(m, "type", 2) == 1)
if current_online_count >= self.MAX_DEVICES:
print(f"[Add](二次检查)已达设备上限 {self.MAX_DEVICES},忽略新增 {udid}")
return
port = self._alloc_port()
print(f"[iproxy] 准备启动 iproxy 映射 {port}->{wdaScreenPort}")
proc = self._start_iproxy(udid, local_port=port)
@@ -331,43 +373,51 @@ class DeviceInfo:
def _remove_device(self, udid: str):
"""
移除设备及其转发,通知上层。
幂等:重复调用不会出错
移除设备及其转发,通知上层(幂等)
✅ 同时释放 self._reserved_ports 中可能残留的保留端口
✅ 同时清理 _iproxy / _port_by_udid。
"""
print(f"[Remove] 正在移除设备 {udid}")
# --- 1. 锁内执行所有轻量字典操作 ---
# --- 1. 锁内取出并清空字典 ---
with self._lock:
model = self._models.pop(udid, None)
proc = self._iproxy.pop(udid, None)
self._port_by_udid.pop(udid, None)
self._first_seen.pop(udid, None)
self._last_seen.pop(udid, None)
port = self._port_by_udid.pop(udid, None)
# --- 2. 锁外执行重操作 ---
# 杀进程
# --- 2. 杀进程 ---
try:
self._kill(proc)
except Exception as e:
print(f"[Remove] 杀进程异常 {udid}: {e}")
# 准备下线模型model 可能为 None
if model is None:
model = DeviceModel(
deviceId=udid, screenPort=-1, width=0, height=0, scale=0.0, type=2
)
# --- 3. 释放“保留端口”(如果还在集合里)---
if isinstance(port, int) and port > 0:
try:
self._release_port(port)
except Exception as e:
print(f"[Remove] 释放保留端口异常 {udid}: {e}")
# --- 4. 构造下线模型并通知 ---
if model is None:
model = DeviceModel(deviceId=udid, screenPort=-1, width=0, height=0, scale=0.0, type=2)
# 标记状态为离线
model.type = 2
model.ready = False
model.screenPort = -1
# 通知上层
try:
self._manager_send(model)
except Exception as e:
print(f"[Remove] 通知上层异常 {udid}: {e}")
# --- 5. 清理失败计数(健康检查用)---
try:
if hasattr(self, "_check_fail"):
self._check_fail.pop(udid, None)
except Exception:
pass
print(f"[Remove] 设备移除完成 {udid}")
def _trusted(self, udid: str) -> bool:
@@ -478,6 +528,10 @@ class DeviceInfo:
return False
def _spawn_iproxy(self, udid: str, local_port: int, remote_port: int) -> Optional[subprocess.Popen]:
"""
启动 iproxy 子进程。
✅ 将 stdout/stderr 写入 log/iproxy/{udid}_{port}.log便于追查“端口被占用/被拦截/崩溃”等原因。
"""
creationflags = 0
startupinfo = None
if os.name == "nt":
@@ -487,29 +541,52 @@ class DeviceInfo:
si.dwFlags |= subprocess.STARTF_USESHOWWINDOW
si.wShowWindow = 0
startupinfo = si
cmd = [self._iproxy_path, "-u", udid, str(local_port), str(remote_port)]
# 日志文件
log_dir = Path("log/iproxy")
try:
print(f"[iproxy] 启动进程 {cmd}")
log_dir.mkdir(parents=True, exist_ok=True)
except Exception:
pass
log_path = log_dir / f"{udid}_{local_port}.log"
try:
print(f"[iproxy] 启动进程 {cmd} (log={log_path})")
logfile = open(log_path, "ab", buffering=0)
return subprocess.Popen(
cmd,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
stdout=logfile,
stderr=logfile,
creationflags=creationflags,
startupinfo=startupinfo,
)
except Exception as e:
print(f"[iproxy] 创建进程失败: {e}")
return None
LogManager.error(f"[iproxy] 创建进程失败: {e}", "system")
def _start_iproxy(self, udid: str, local_port: int) -> Optional[subprocess.Popen]:
"""
启动 iproxy 并等待本地端口监听成功。
✅ 监听成功后,立刻释放 self._reserved_ports 对应的“保留”,避免保留池越攒越多。
"""
proc = self._spawn_iproxy(udid, local_port=local_port, remote_port=wdaScreenPort)
if not proc:
print(f"[iproxy] 启动失败 {udid}")
return None
if not self._wait_until_listening(local_port, 3.0):
self._kill(proc)
print(f"[iproxy] 未监听, 已杀死 {udid}")
return None
# ✅ 监听成功,释放“保留端口”
try:
self._release_port(local_port)
except Exception as e:
print(f"[iproxy] 释放保留端口异常: {e}")
print(f"[iproxy] 启动成功 port={local_port} {udid}")
return proc

View File

@@ -667,7 +667,6 @@ def delete_last_message():
def stopAllTask():
idList = request.get_json()
code, msg = ThreadManager.batch_stop(idList)
time.sleep(2)
return ResultData(code, [], msg).toJson()
# 切换账号

View File

@@ -224,6 +224,11 @@ class ScriptManager():
self.comment_flow(filePath, session, udid, recomend_cx, recomend_cy)
event.wait(timeout=2)
home = AiUtils.findHomeButton(session)
if not home:
raise Exception("没有找到首页按钮,重置")
videoTime = random.randint(15, 30)
for _ in range(videoTime):
if event.is_set():