zw/iOSAI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复打包问题

Browse Source
This commit is contained in:
milk
2025-10-23 18:53:22 +08:00
parent 2310333a60
commit 6cf4d9cb03
2188 changed files with 889511 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
Module/DeviceInfo.py
View File

@@ -122,9 +122,7 @@ class DeviceInfo:
args=(udid,)
).start()
else:
print("准备启动wda")
dev.app_start(WdaAppBundleId)
print("启动wda完成")
print("启动wda成功")
time.sleep(3)
return True
@@ -138,12 +136,12 @@ class DeviceInfo:
c.home()
size = c.window_size()
scale = c.scale
print("已获取到屏幕大小信息")
return int(size.width), int(size.height), float(scale)
except Exception as e:
print("获取设备信息遇到错误：", e)
return 0, 0, 0
...
# ---------------- 原来代码不变，只替换下面一个函数 ----------------
def _start_iproxy(self, udid: str, port: int) -> Optional[subprocess.Popen]:
try:

7
Module/FlaskService.py
View File

@@ -197,6 +197,9 @@ def tapAction():
body = request.get_json()
udid = body.get("udid")
client = wda.USBClient(udid, wdaFunctionPort)
print("-----------------------")
print(client)
print("-----------------------")
session = client.session()
session.appium_settings({"snapshotMaxDepth": 0})
x = body.get("x")
@@ -243,7 +246,7 @@ def longPressAction():
def growAccount():
body = request.get_json()
udid = body.get("udid")
Variables.commentList = body.get("comment")
# Variables.commentList = body.get("comment")
manager = ScriptManager()
event = threading.Event()
@@ -289,7 +292,7 @@ def passAnchorData():
# 主播列表
acList = data.get("anchorList", [])
Variables.commentList = data.get("comment")
# Variables.commentList = data.get("comment")
LogManager.info(f"[INFO] 获取数据: {idList} {acList}")

338
Module/IOSActivator.py
View File

@@ -1,9 +1,7 @@
import os
import random
import re
import socket
import sys
import socket
import subprocess
from typing import Optional
@@ -12,59 +10,180 @@ from Entity.Variables import WdaAppBundleId
class IOSActivator:
"""
轻量 iOS 激活器（仅代码调用）：
1) 启动 `pymobiledevice3 remote tunneld`（基于传入 UDID）
2) 自动挂载 Developer Disk Image
3) 设备隧道就绪后启动 WDA：
- 优先使用 `--rsd <host> <port>` 直连（支持 IPv6）
- 失败再用 `PYMOBILEDEVICE3_TUNNEL=127.0.0.1:<port>` 作为退路（仅 IPv4）
轻量 iOS 激活器（仅代码调用） - 打包安全版
1) 启动 `pymobiledevice3 remote tunneld`（子进程常驻）
2) 自动挂载 Developer Disk Image（进程内调用 CLI）
3) 设备隧道就绪后启动 WDA（进程内调用 CLI）：
- 优先使用 `--rsd <host> <port>`（支持 IPv6）
- 失败再用 `PYMOBILEDEVICE3_TUNNEL=127.0.0.1:<port>` 回退（仅 IPv4）
"""
def __init__(self, python_executable: Optional[str] = None):
self.python = python_executable or sys.executable
# 仅用于旧逻辑兼容；本版本已不依赖 self.python 去 -m 调 CLI
self.python = python_executable or None
# --------------------------
# 内部工具
# --------------------------
def _auto_mount_developer_disk(self, udid: str, retries: int = 3, backoff_seconds: float = 2.0) -> None:
"""使用 `pymobiledevice3 mounter auto-mount` 为指定 UDID 挂载开发者镜像（带重试）。"""
env = os.environ.copy()
# =========================================================================
# 内部工具：进程内调用 pymobiledevice3 CLI
# =========================================================================
def _pmd3_run(self, args: list[str], udid: str, extra_env: Optional[dict] = None) -> str:
import subprocess
launcher, env = self._resolve_pmd3_argv_and_env()
env["PYMOBILEDEVICE3_UDID"] = udid
env["PYTHONUNBUFFERED"] = "1"
env.setdefault("PYTHONIOENCODING", "utf-8")
if extra_env:
for k, v in extra_env.items():
if v is None:
env.pop(k, None)
else:
env[k] = str(v)
cmd = [*launcher, *args]
print("[pmd3]", " ".join(map(str, cmd)))
try:
return subprocess.check_output(cmd, text=True, stderr=subprocess.STDOUT, env=env) or ""
except subprocess.CalledProcessError as exc:
raise RuntimeError(exc.output or f"pymobiledevice3 执行失败，退出码 {exc.returncode}")
max_attempts = max(1, int(retries))
last_err = None
for i in range(max_attempts):
def _pmd3_run_subprocess(self, full_args: list[str], extra_env: Optional[dict] = None) -> str:
"""
兜底：通过子进程执行 pymobiledevice3（使用 _resolve_pmd3_argv_and_env()）。
"""
import subprocess
launcher, env = self._resolve_pmd3_argv_and_env()
if extra_env:
for k, v in extra_env.items():
if v is None:
env.pop(k, None)
else:
env[k] = str(v)
cmd = [*launcher, *full_args]
cmd = self._ensure_str_list(cmd)
print("[pmd3-subproc]", " ".join(cmd))
try:
out = subprocess.check_output(cmd, text=True, stderr=subprocess.STDOUT, env=env)
return out or ""
except subprocess.CalledProcessError as exc:
raise RuntimeError(exc.output or f"pymobiledevice3 子进程执行失败，代码 {exc.returncode}")
# =========================================================================
# 旧版环境依赖的替代方案：仅用于启动 tunneld 的子进程（需要常驻）
# =========================================================================
def _resolve_pmd3_argv_and_env(self):
import os, sys, shutil, subprocess
from pathlib import Path
env = os.environ.copy()
env["PYTHONUNBUFFERED"] = "1"
env.setdefault("PYTHONIOENCODING", "utf-8")
# 1) 明确使用 IOSAI_PYTHON（如果上层已设置）
prefer_py = env.get("IOSAI_PYTHON")
# 2) 构造一批候选路径（先根目录，再 Scripts）
base = Path(sys.argv[0]).resolve()
base_dir = base.parent if base.is_file() else base
py_name = "python.exe" if os.name == "nt" else "python"
sidecar_candidates = [
base_dir / "python-rt" / py_name, # ✅ 嵌入式/你现在的摆放方式
base_dir / "python-rt" / "Scripts" / py_name, # 兼容虚拟环境式
base_dir.parent / "python-rt" / py_name,
base_dir.parent / "python-rt" / "Scripts" / py_name,
]
# 若外层已通过 IOSAI_PYTHON 指了路径，也放进候选
if prefer_py:
sidecar_candidates.insert(0, Path(prefer_py))
# 打印探测
for cand in sidecar_candidates:
print(f"[IOSAI] 🔎 probing sidecar at: {cand}")
if cand.is_file():
try:
out = subprocess.check_output(
[str(cand), "-c", "import pymobiledevice3;print('ok')"],
text=True, stderr=subprocess.STDOUT, env=env, timeout=6
)
if "ok" in out:
print(f"[IOSAI] ✅ sidecar selected: {cand}")
return ([str(cand), "-u", "-m", "pymobiledevice3"], env)
except Exception:
# 候选解释器存在但没装 pmd3，就继续找下一个
pass
# 3) 回退：系统 PATH 里直接找 pymobiledevice3 可执行
exe = shutil.which("pymobiledevice3")
if exe:
print(f"[IOSAI] ✅ use PATH executable: {exe}")
return ([exe], env)
# 4) 兜底：从系统 Python 里找装了 pmd3 的解释器
py_candidates = []
base_exec = getattr(sys, "_base_executable", None)
if base_exec and os.path.isfile(base_exec):
py_candidates.append(base_exec)
for name in ("python3.exe", "python.exe", "py.exe", "python3", "python"):
p = shutil.which(name)
if p and p not in py_candidates:
py_candidates.append(p)
for py in py_candidates:
print(f"[IOSAI] 🔎 probing system python: {py}")
try:
out = subprocess.check_output(
[self.python, "-m", "pymobiledevice3", "mounter", "auto-mount"],
text=True,
stderr=subprocess.STDOUT,
env=env,
[py, "-c", "import pymobiledevice3;print('ok')"],
text=True, stderr=subprocess.STDOUT, env=env, timeout=6
)
if "ok" in out:
print(f"[IOSAI] ✅ system python selected: {py}")
return ([py, "-u", "-m", "pymobiledevice3"], env)
except Exception:
continue
raise RuntimeError("未检测到可用的 pymobiledevice3（建议携带 python-rt 或安装系统 Python+pmd3）。")
def _ensure_str_list(self, seq):
return [str(x) for x in seq]
# =========================================================================
# 功能函数-
# =========================================================================
def _auto_mount_developer_disk(self, udid: str, retries: int = 3, backoff_seconds: float = 2.0) -> None:
print("[mounter] Developer disk image mounted.")
"""
使用进程内 CLI：pymobiledevice3 mounter auto-mount（带重试）
"""
import time
last_err_text = ""
for i in range(max(1, retries)):
try:
out = self._pmd3_run(["mounter", "auto-mount"], udid)
if out:
for line in out.splitlines():
print(f"[mounter] {line}")
print("[mounter] Developer disk image mounted.")
return
except subprocess.CalledProcessError as exc:
lowered = (exc.output or "").lower()
if "already mounted" in lowered:
if "already mounted" in (out or "").lower():
print("[mounter] Developer disk image already mounted.")
return
last_err = exc
if i < max_attempts - 1:
print(f"[mounter] attempt {i+1}/{max_attempts} failed, retrying in {backoff_seconds}s ...")
else:
print("[mounter] Developer disk image mounted.")
return
except Exception as e:
last_err_text = str(e)
if i < retries - 1:
print(f"[mounter] attempt {i+1}/{retries} failed, retrying in {backoff_seconds}s ...")
try:
import time as _t
_t.sleep(backoff_seconds)
time.sleep(backoff_seconds)
except Exception:
pass
msg = last_err.output if isinstance(last_err, subprocess.CalledProcessError) else str(last_err)
raise RuntimeError(f"Auto-mount failed after {max_attempts} attempts: {msg}")
else:
raise RuntimeError(f"Auto-mount failed after {retries} attempts.\n{last_err_text}")
def _is_ipv4_host(self, host: str) -> bool:
return bool(re.match(r"^\d{1,3}(\.\d{1,3}){3}$", host))
import re as _re
return bool(_re.match(r"^\d{1,3}(\.\d{1,3}){3}$", host))
def _wait_for_rsd_ready(self, rsd_host: str, rsd_port: str, retries: int = 5, delay: float = 3.0) -> bool:
"""
@@ -86,23 +205,13 @@ class IOSActivator:
def _launch_wda_via_rsd(self, bundle_id: str, rsd_host: str, rsd_port: str, udid: str) -> None:
"""
使用 `--rsd <host> <port>` 直连设备隧道来启动 WDA（推荐路径，IPv4/IPv6 都 OK）。
不设置 PYMOBILEDEVICE3_TUNNEL，避免 IPv6 解析问题。
使用进程内 CLI：pymobiledevice3 developer dvt launch <bundle> --rsd <host> <port>
"""
print(f"[wda] Launch via RSD {rsd_host}:{rsd_port}, bundle: {bundle_id}")
env = os.environ.copy()
env["PYMOBILEDEVICE3_UDID"] = udid
args = [
self.python, "-m", "pymobiledevice3",
"developer", "dvt", "launch", bundle_id,
"--rsd", rsd_host, rsd_port,
]
try:
out = subprocess.check_output(args, text=True, stderr=subprocess.STDOUT, env=env)
except subprocess.CalledProcessError as exc:
raise RuntimeError(f"WDA launch via RSD failed: {exc.output}") from exc
out = self._pmd3_run(
["developer", "dvt", "launch", bundle_id, "--rsd", rsd_host, rsd_port],
udid=udid,
)
if out:
for line in out.splitlines():
print(f"[wda] {line}")
@@ -111,22 +220,19 @@ class IOSActivator:
def _launch_wda_via_http_tunnel(self, bundle_id: str, http_host: str, http_port: str, udid: str) -> None:
"""
退路：通过 HTTP 网关端口设置 PYMOBILEDEVICE3_TUNNEL（仅 IPv4）。
使用进程内 CLI 启动 WDA。
"""
if not self._is_ipv4_host(http_host):
raise RuntimeError(f"HTTP tunnel host must be IPv4, got {http_host}")
tunnel_endpoint = f"{http_host}:{http_port}"
print(f"[wda] Launch via HTTP tunnel {tunnel_endpoint}, bundle: {bundle_id}")
env = os.environ.copy()
env["PYMOBILEDEVICE3_TUNNEL"] = tunnel_endpoint
env["PYMOBILEDEVICE3_UDID"] = udid
args = [self.python, "-m", "pymobiledevice3", "developer", "dvt", "launch", bundle_id]
try:
out = subprocess.check_output(args, text=True, stderr=subprocess.STDOUT, env=env)
except subprocess.CalledProcessError as exc:
raise RuntimeError(f"WDA launch via HTTP tunnel failed: {exc.output}") from exc
out = self._pmd3_run(
["developer", "dvt", "launch", bundle_id],
udid=udid,
extra_env={"PYMOBILEDEVICE3_TUNNEL": tunnel_endpoint},
)
if out:
for line in out.splitlines():
print(f"[wda] {line}")
@@ -141,41 +247,69 @@ class IOSActivator:
return port
raise RuntimeError("No free port found for tunneld")
# --------------------------
# =========================================================================
# 对外方法
# --------------------------
# =========================================================================
def activate(
self,
udid: str,
wda_bundle_id: Optional[str] = WdaAppBundleId,
ready_timeout_sec: float = 120.0,
mount_retries: int = 3,
backoff_seconds: float = 2.0,
rsd_probe_retries: int = 5,
rsd_probe_delay_sec: float = 3.0,
self,
udid: str,
wda_bundle_id: Optional[str] = WdaAppBundleId,
ready_timeout_sec: float = 120.0,
mount_retries: int = 3,
backoff_seconds: float = 2.0,
rsd_probe_retries: int = 5,
rsd_probe_delay_sec: float = 3.0,
pre_mount_first: bool = True,
) -> str:
"""
执行：开隧道 -> （等待 RSD 就绪）-> 挂载镜像 -> 启动 WDA
- 优先用 `--rsd` 启动（先做 dvt list 探测）
- 失败再用 HTTP 端口作为退路
执行：挂镜像(可选) -> 开隧道 -> （等待 RSD 就绪）-> 启动 WDA
- 优先用 `--rsd` 启动
- 失败再用 HTTP 网关作为退路
"""
if not udid or not isinstance(udid, str):
raise ValueError("udid is required and must be a non-empty string")
print(f"[activate] UDID = {udid}")
env = os.environ.copy()
env["PYMOBILEDEVICE3_UDID"] = udid
# 1) 开隧道（子进程常驻）
# ⚠️ 检查管理员权限
if os.name == "nt":
import ctypes
try:
is_admin = ctypes.windll.shell32.IsUserAnAdmin() != 0
except Exception:
is_admin = False
if not is_admin:
print("[⚠️] 当前进程未以管理员身份运行，tunneld 可能无法创建 USB 隧道。")
import time as _t
start_ts = _t.time()
# 1) (可选) 先挂载，避免 tidevice 并发 DDI 竞态
if pre_mount_first:
try:
self._auto_mount_developer_disk(udid, retries=mount_retries, backoff_seconds=backoff_seconds)
_t.sleep(2) # 稳定 DDI
except Exception as e:
print(f"[activate] 预挂载失败（继续尝试开隧道后再挂载一次）：{e}")
# 2) 启动 tunneld 子进程
port = self._pick_available_port()
cmd = [self.python, "-m", "pymobiledevice3", "remote", "tunneld", "--port", str(port)]
launcher, env2 = self._resolve_pmd3_argv_and_env()
env2["PYTHONUNBUFFERED"] = "1"
env2.setdefault("PYTHONIOENCODING", "utf-8")
env2["PYMOBILEDEVICE3_UDID"] = udid
cmd = self._ensure_str_list([*launcher, "remote", "tunneld", "--port", str(port)])
print("[activate] 使用命令启动隧道:", " ".join(cmd))
proc = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
env=env,
bufsize=1,
universal_newlines=True,
env=env2,
)
captured: list[str] = []
@@ -185,13 +319,11 @@ class IOSActivator:
rsd_port: Optional[str] = None
device_tunnel_ready = False
wda_started = False
mount_done = False
import time as _t
start_ts = _t.time()
mount_done = pre_mount_first
HTTP_RE = re.compile(r"http://([0-9.]+):(\d+)")
RSD_CREATED_RE = re.compile(r"Created tunnel\s+--rsd\s+([^\s]+)\s+(\d+)")
RSD_FALLBACK_RE = re.compile(r"--rsd\s+(\S+?)[\s:](\d+)")
try:
assert proc.stdout is not None
@@ -209,23 +341,26 @@ class IOSActivator:
http_host, http_port = m.group(1), m.group(2)
print(f"[tunneld] Tunnel API: {http_host}:{http_port}")
# 捕获设备 RSD（可能 IPv6）
m = RSD_CREATED_RE.search(line)
if m:
rsd_host, rsd_port = m.group(1), m.group(2)
device_tunnel_ready = True
print(f"[tunneld] Device-level tunnel ready (RSD {rsd_host}:{rsd_port}).")
# ✅ 捕获 RSD（仅识别当前 UDID 的行）
if self._line_is_for_udid(line, udid):
m = RSD_CREATED_RE.search(line) or RSD_FALLBACK_RE.search(line)
if m and not device_tunnel_ready:
rsd_host, rsd_port = m.group(1), m.group(2)
device_tunnel_ready = True
print(f"[tunneld] Device-level tunnel ready (RSD {rsd_host}:{rsd_port}).")
else:
continue # 其它设备的隧道日志忽略
# 条件满足后推进
# 当设备隧道准备好后启动 WDA
if (not wda_started) and wda_bundle_id and device_tunnel_ready:
try:
if not mount_done:
self._auto_mount_developer_disk(
udid, retries=mount_retries, backoff_seconds=backoff_seconds
)
_t.sleep(2)
mount_done = True
# 先做 RSD 就绪探测，再走 RSD 启动
rsd_ok = False
if rsd_host and rsd_port:
rsd_ok = self._wait_for_rsd_ready(
@@ -233,15 +368,15 @@ class IOSActivator:
retries=rsd_probe_retries,
delay=rsd_probe_delay_sec,
)
if rsd_ok:
self._launch_wda_via_rsd(
bundle_id=wda_bundle_id,
rsd_host=rsd_host, # type: ignore[arg-type]
rsd_port=rsd_port, # type: ignore[arg-type]
rsd_host=rsd_host,
rsd_port=rsd_port,
udid=udid,
)
else:
# RSD 不就绪或失败，回退到 HTTP 网关（必须是 IPv4）
if http_host and http_port:
self._launch_wda_via_http_tunnel(
bundle_id=wda_bundle_id,
@@ -263,7 +398,7 @@ class IOSActivator:
proc.kill()
raise
# 超时保护（仍未启动 WDA）
# 超时保护
if (not wda_started) and ready_timeout_sec > 0 and (_t.time() - start_ts > ready_timeout_sec):
print(f"[tunneld] Timeout waiting for device tunnel ({ready_timeout_sec}s). Aborting.")
proc.terminate()
@@ -273,7 +408,7 @@ class IOSActivator:
proc.kill()
break
# 等待子进程结束（若已结束）
# 结束清理
try:
return_code = proc.wait(timeout=5)
except subprocess.TimeoutExpired:
@@ -298,3 +433,12 @@ class IOSActivator:
except Exception:
pass
raise
def _line_is_for_udid(self, line: str, udid: str) -> bool:
"""日志行是否属于目标 UDID。"""
try:
return udid.lower() in (line or "").lower()
except Exception:
return False

13
Module/Main.py
View File

@@ -1,7 +1,20 @@
# ===== Main.py 顶部放置（所有 import 之前）=====
import os
import sys
from pathlib import Path
if "IOSAI_PYTHON" not in os.environ:
base_path = Path(sys.argv[0]).resolve()
base_dir = base_path.parent if base_path.is_file() else base_path
sidecar = base_dir / "python-rt" / ("python.exe" if os.name == "nt" else "python")
if sidecar.exists():
os.environ["IOSAI_PYTHON"] = str(sidecar)
# ==============================================
import sys
from pathlib import Path
from Module.DeviceInfo import DeviceInfo
from Module.FlaskSubprocessManager import FlaskSubprocessManager
from Utils.DevDiskImageDeployer import DevDiskImageDeployer

BIN
Module/__pycache__/DeviceInfo.cpython-312.pyc
View File

Binary file not shown.

BIN
Module/__pycache__/FlaskService.cpython-312.pyc
View File

Binary file not shown.

BIN
Module/__pycache__/Main.cpython-312.pyc
View File

Binary file not shown.