修正检测投屏端口逻辑

This commit is contained in:
2025-10-31 19:41:44 +08:00
parent bdc6414f69
commit 929061e4b5
4 changed files with 27 additions and 16 deletions

View File

@@ -119,31 +119,42 @@ class DeviceInfo:
threading.Thread(target=self.check_iproxy_ports).start() threading.Thread(target=self.check_iproxy_ports).start()
# =============== 核心端口连通性检测HTTP 方式) =================
# =============== 核心:端口连通性检测 ================= def _is_local_port_open(self, port: int, udid: str, timeout: float = 5) -> bool:
def _is_local_port_open(self, port: int,udid:str, timeout: float = 5) -> bool: print("开始HTTP方式检测端口")
print("开始监听剪口")
""" """
Windows: 尝试连接 127.0.0.1:port能连上即认为端口可用iproxy 正在监听/转发) 使用 HTTP 方式检测:向 http://127.0.0.1:port/ 发送一次 HEAD 请求
不抛异常,返回 True/False 只要建立连接并收到合法的 HTTP 响应(任意 1xx~5xx 状态码),即认为 HTTP 可达
遇到连接失败、超时、协议不对等异常,视为不可用。
""" """
if not isinstance(port, int) or port <= 0 or port > 65535: if not isinstance(port, int) or port <= 0 or port > 65535:
LogManager.error("端口不可用",udid=udid) LogManager.error("端口不可用(非法端口号)", udid=udid)
return False return False
try: try:
print("尝试监听") # HEAD 更轻;若后端对 HEAD 不友好,可改为 "GET", "/"
# create_connection 会在连接成功时立即返回 socket conn = http.client.HTTPConnection("127.0.0.1", int(port), timeout=timeout)
with socket.create_connection(("127.0.0.1", int(port)), timeout=timeout): conn.request("HEAD", "/")
print("端口正常") resp = conn.getresponse()
status = resp.status
# 读到响应即可关闭
conn.close()
# 任何合法 HTTP 状态码都说明“HTTP 服务在监听且可交互”,包括 404/401/403/5xx
if 100 <= status <= 599:
print(f"HTTP端口正常status={status}")
return True return True
except OSError: else:
LogManager.error("端口不可用",udid=udid) LogManager.error(f"HTTP状态码异常: {status}", udid=udid)
return False
except Exception as e:
# 连接被拒绝、超时、不是HTTP协议正确响应比如返回了非HTTP的字节流都会到这里
LogManager.error(f"HTTP检测失败{e}", udid=udid)
return False return False
# =============== 一轮检查:发现不通就移除 ================= # =============== 一轮检查:发现不通就移除 =================
def check_iproxy_ports(self, connect_timeout: float = 3) -> None: def check_iproxy_ports(self, connect_timeout: float = 3) -> None:
time.sleep(60) time.sleep(20)
print("开始监听投屏端口") print("开始监听投屏端口")
while True: while True:
snapshot = list(self._models.items()) # [(deviceId, DeviceModel), ...] snapshot = list(self._models.items()) # [(deviceId, DeviceModel), ...]

View File

@@ -50,7 +50,7 @@ if __name__ == "__main__":
# 清空日志 # 清空日志
LogManager.clearLogs() LogManager.clearLogs()
# main(sys.argv) main(sys.argv)
# 添加iOS开发包到电脑上 # 添加iOS开发包到电脑上
deployer = DevDiskImageDeployer(verbose=True) deployer = DevDiskImageDeployer(verbose=True)

View File

@@ -26,7 +26,7 @@ def _force_utf8_everywhere():
except Exception: except Exception:
pass pass
# _force_utf8_everywhere() _force_utf8_everywhere()
class LogManager: class LogManager:
""" """