一、项目概述
WebAppForIOS 是一个功能完整的 iOS 设备 Web 调试平台，基于 Python + Flask实现跨平台 Web 调试功能，支持 Windows / macOS / Linux 平台，通过浏览器即可实现对 iPhone 设备的全面管理与调试。
项目已兼容 iOS 26 版本，并支持 iOS 17+ Tunnel 模式，确保与最新 iOS 系统的适配。通过Web界面即可完成 设备操作、调试与诊断，图像化界面化操作提升 iOS 研发与测试的效率。
二、使用指南
设备链接步骤
打开 开发者模式
使用USB-C链接线链接 iOS 设备
在手机上选择 “信任此电脑”
点击 “刷新设备” 按钮
从下拉列表中选择目标设备
三、功能特性
1、设备管理
自动检测 iOS 设备
获取型号、系统版本、序列号
远程重启设备
检测开发者模式状态
实时监控连接状态
2、应用管理
获取已安装的第三方应用列表
按名称或 Bundle ID 搜索应用
启动 / 停止应用
上传并安装 IPA 文件
查看运行中应用进程
3、截图与录屏
单次截图
实时录屏（MJPEG / MP4）
开始/停止录制，文件自动保存
下载录制文件
4、系统日志
实时日志流 (SSE)
按关键字 / 级别过滤日志
导出日志文件
支持原始日志 / 解析后日志
5、崩溃日志管理
获取崩溃日志列表
支持搜索与批量导出
删除指定日志
6、系统功能
获取电池信息
查看磁盘空间
管理 AssistiveTouch / VoiceOver / Zoom
管理设备配置文件
7、调试功能
Debug 模式（三连击Copyright文字触发）
导出运行日志
清理遗留 ios.exe 进程
启动/退出时自动清理资源
8、运行环境与常见问题：
运行环境：
Python 3.8+
运行平台：
Windows/macOS/Linux
USB 驱动：
iPhone 需信任电脑
iOS 17+：
如需 Tunnel 功能
设备连接失败
检查 USB 线确认已“信任此电脑”重启设备
Tunnel 启动失败
Windows 确认 wintun.dll 已安装Linux/macOS 用 sudo 运行检查防火墙
截图/录屏失败
确认开发者模式已启用检查开发者镜像挂载情况
ios.exe 进程累积
使用 Debug 清理进程重启应用
四、项目框架代码及代码内容：
1、项目框架
### 前端架构 HTML5 + CSS3 + JavaScript + Bootstrap + jQuery
web_function/
├── app.py # Flask 主应用，API 路由定义
├── templates/
│ └── index.html # 主页面模板
└── static/
├── wcss/ # CSS 样式文件
│ ├── bootstrap.min.css
│ ├── style.css
│ └── sweetalert2.min.css
├── wjss/ # JavaScript 文件
│ ├── app_ios.js
│ ├── bootstrap.min.js
│ ├── jquery.min.js
│ └── sweetalert2.all.min.js
└── wresource/
├── background.jpg
├── favicon.ico
└── wicon.ico
### 后端架构 (Python + Flask)
backend_function/
├── config.py # 配置文件
├── goios_wrapper.py # go-ios 封装
├── tunnel_manager.py # Tunnel 管理
├── ios_prechecker.py # 设备检查器
└── common_utils.py # 工具函数
IOSPrechecker/
├── executable/ # go-ios 可执行文件
│ └── win/
│ ├── ios.exe
│ └── selfIdentity.plist
├── utils/ # go-ios 压缩包
├── devimages/ # 开发者镜像
└── wintun/ # Windows Tunnel 驱动
ios_downloads/ # 下载文件目录
└── crashes/ # 崩溃日志
ios_logs/ # 日志目录
requirements.txt # Python 依赖2、app代码
import os
import sys
import time
import uuid
import json
import signal
import atexit
import logging
import webbrowser
import threading
import subprocess
from logging.handlers import RotatingFileHandler
from typing import Any, Dict
from urllib.parse import urlsplit, urlunsplit
from flask import (
Flask, render_template, request, jsonify,
send_from_directory, abort, Response, stream_with_context
)
from werkzeug.utils import secure_filename
from backend_function.common_utils import (
get_local_ip,
normalize_ios_info,
build_ordered_ios_info,
normalize_ios_list_output,
parse_ios_apps,
to_int,
extract_goios_opts,
terminate_process,
get_device_model,
now_timestamp_str,
create_required_directories,
cleanup_old_tunnel_processes,
cleanup_all_ios_processes,
create_signed_download_token,
consume_signed_download_token,
parse_ps_apps_raw,
parse_crash_ls_items,
crash_export_collect,
crash_zip_dir,
crash_remove_many, listen_event_stream, syslog_start_session,
run_with_quick_check_and_escalate,
stream_syslog_sse,
)
from backend_function.config import Config
from backend_function.goios_wrapper import GoIOSManager
from backend_function.ios_prechecker import IOSPrechecker
from backend_function.tunnel_manager import TunnelManager
from backend_function.common_utils import start_mjpeg_to_mp4, stop_recorder
app = Flask(__name__)
# 简单签名/令牌存储（内存，短时有效）
_SIGNED_DOWNLOADS: Dict[str, Dict[str, Any]] = {}
# ===== 配置与日志 =====
app.config.from_object(Config)
app.config['ENV'] = 'development'
app.config['DEBUG'] = True
os.makedirs(app.config["LOG_DIR"], exist_ok=True)
log_path = os.path.join(app.config["LOG_DIR"], "ios_app.log")
handler = RotatingFileHandler(log_path, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8")
fmt = logging.Formatter("[%(asctime)s] %(levelname)s %(name)s - %(message)s")
handler.setFormatter(fmt)
app.logger.setLevel(logging.INFO)
app.logger.propagate = True
root_logger = logging.getLogger()
root_logger.setLevel(logging.INFO)
# 防重复：仅将文件句柄挂到根记录器
if not any(isinstance(h, RotatingFileHandler) and getattr(h, 'baseFilename', None) == log_path for h in root_logger.handlers):
root_logger.addHandler(handler)
# ===== 初始化 go-ios 管理器与检查器 =====
goios = GoIOSManager(
goios_root=app.config["GOIOS_DIR"],
bin_dir=app.config["GOIOS_EXECUTABLE_DIR"],
bin_path_override=app.config["GOIOS_BIN_PATH"],
)
tunnel = TunnelManager(goios)
prechecker = IOSPrechecker(goios, tunnel)
# === 创建必要目录 ===
create_required_directories(app.config, app.logger)
# === Windows tunnel 依赖检查（一次） ===
if os.name == 'nt' and hasattr(tunnel, 'check_windows_wintun'):
wintun_ok, wintun_msg = tunnel.check_windows_wintun()
if wintun_ok:
app.logger.debug("Windows tunnel 依赖检查: %s", wintun_msg)
else:
app.logger.warning("Windows tunnel 依赖检查失败：%s", wintun_msg)
# === 清理旧的 tunnel 进程 ===
cleanup_old_tunnel_processes(tunnel, app.logger)
# === 清理遗留的 ios.exe 进程 ===
cleanup_all_ios_processes(app.logger)
# 目录已在上面的 create_directories() 中创建
FORWARDS: Dict[str, Dict[str, Dict[str, Any]]] = {}
STREAMS: Dict[str, Dict[str, Dict[str, Any]]] = {}
SYSLOGS: Dict[str, Dict[str, Dict[str, Any]]] = {}
# ===== 应用停止时清理逻辑 =====
def cleanup_tunnel():
"""应用停止时清理隧道连接和遗留进程"""
try:
app.logger.debug("应用停止，清理隧道连接...")
tunnel.stop()
app.logger.debug("隧道清理完成")
except (AttributeError, OSError) as exc:
app.logger.warning("隧道清理失败: %s", exc)
try:
app.logger.debug("应用停止，清理遗留的ios.exe进程...")
cleanup_all_ios_processes(app.logger)
app.logger.debug("ios.exe进程清理完成")
except Exception as exc:
app.logger.warning("ios.exe进程清理失败: %s", exc)
def signal_handler(signum, _frame):
"""信号处理器"""
app.logger.debug("接收到停止信号 %s，开始清理...", signum)
cleanup_tunnel()
sys.exit(0)
# 注册清理函数
atexit.register(cleanup_tunnel)
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
# ===== 工具函数 =====
# 已迁移到 backend_function.common_utils
# ===== 路由 =====
@app.route("/")
def index():
success, out = goios.list_devices(details=False)
devices = []
if success and out:
for line in out.splitlines():
val = line.strip()
if val and len(val) >= 16:
devices.append(val)
return render_template("index.html", devices=devices)
@app.route("/favicon.ico")
def favicon():
return send_from_directory(os.path.join(app.root_path, 'static', 'wresource'), 'favicon.ico')
@app.route("/guide")
def guide():
"""引导文档页面"""
return render_template("ios_introduce_guide_index.html")
# ---------- 设备管理 ----------
@app.route("/api/devices")
def api_devices():
# 设备列表不需要tunnel
details = request.args.get("details", "0") == "1"
opts = extract_goios_opts(request.args)
success, out_text = goios.list_devices(details=details, **opts)
if not success:
return jsonify({"ok": False, "msg": "获取设备列表失败", "raw": out_text}), 500
devices = normalize_ios_list_output(out_text or "")
return jsonify({"ok": True, "devices": devices, "raw": out_text})
@app.route("/api/device_info")
def api_device_info():
udid = request.args.get("udid", "").strip()
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
# 使用快速检查而不是完整检查
is_ready, check_msg = prechecker.quick_check(udid)
if not is_ready:
return jsonify({"ok": False, "msg": check_msg}), 400
opts = extract_goios_opts(request.args)
got_info, detail = goios.device_info(udid, **opts)
info_list = []
if got_info and detail:
try:
parsed = json.loads(detail) if isinstance(detail, str) else detail
info_map = normalize_ios_info(parsed)
info_list = build_ordered_ios_info(info_map)
except json.JSONDecodeError as exc:
app.logger.warning("device_info JSON 解析失败: %s", exc)
except (KeyError, TypeError, ValueError) as exc:
app.logger.warning("device_info 数据结构异常: %s", exc)
except Exception as exc:
app.logger.exception("device_info 未知异常: %s", exc)
return jsonify({"ok": got_info, "info": info_list, "raw": detail})
# 新增：单次截屏（返回图片二进制）
@app.route("/api/screenshot")
def api_screenshot():
udid = request.args.get("udid", "").strip()
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
# 尝试快速路径：不启动 tunnel，直接截屏；失败再完整检查后重试
is_ready, check_msg = prechecker.quick_check(udid)
if not is_ready:
return jsonify({"ok": False, "msg": check_msg}), 400
model = get_device_model(udid, goios)
ts = now_timestamp_str()
display_name = f"{model}_{udid}_screenshot_{ts}.png"
import time
fname = f"{udid}_screenshot_{int(time.time())}.png" # 实际存储名
save_path = os.path.join(app.config["UPLOAD_FOLDER"], fname)
# 第一次尝试
first_ok, first_out = goios.screenshot(udid, save_path)
if not first_ok:
app.logger.warning("screenshot 首次失败，将执行完整检查后重试: %s", first_out)
# 第二次尝试：强制挂载开发者镜像
try:
# 先确保tunnel运行
tunnel_ok, tunnel_msg = tunnel.status()
if not tunnel_ok:
return jsonify({"ok": False, "msg": f"Tunnel启动失败: {tunnel_msg}"}), 500
# 智能挂载开发者镜像
extra_opts = tunnel.get_goios_opts(udid) or {}
# 先尝试自动下载和挂载（go-ios会自动检测设备版本并下载对应镜像）
app.logger.info("尝试自动下载和挂载开发者镜像...")
image_ok, image_out = goios.image_auto(udid, basedir=app.config["DEVIMAGES_DIR"], extra_env={"ENABLE_GO_IOS_AGENT": "user"}, **extra_opts)
# 验证镜像是否真正挂载成功
if image_ok:
app.logger.info("自动挂载命令成功，验证镜像状态...")
# 检查镜像列表，确认是否真正挂载
list_ok, list_out = goios.image_list(udid, **extra_opts)
if list_ok and "none" not in list_out.lower():
app.logger.info("镜像验证成功: %s", list_out)
image_ok = True
else:
app.logger.warning("镜像验证失败，实际未挂载: %s", list_out)
image_ok = False
if not image_ok:
app.logger.warning("自动挂载开发者镜像失败: %s", image_out)
# 如果自动挂载失败，尝试手动挂载现有镜像
app.logger.info("尝试手动挂载现有开发者镜像...")
# 查找可用的开发者镜像
devimages_dir = app.config["DEVIMAGES_DIR"]
available_images = []
# 扫描所有子目录中的DeveloperDiskImage.dmg文件
for root, dirs, files in os.walk(devimages_dir):
for file in files:
if file == "DeveloperDiskImage.dmg":
available_images.append(os.path.join(root, file))
if available_images:
# 尝试挂载第一个可用的镜像
image_path = available_images[0]
app.logger.info("尝试挂载镜像: %s", image_path)
# 尝试多种挂载方式
mount_attempts = [
# 方式1: 标准挂载
lambda: goios.image_mount(udid, path=image_path, extra_env={"ENABLE_GO_IOS_AGENT": "user"}, **extra_opts),
# 方式2: 跳过签名验证（如果支持）
lambda: goios.image_mount(udid, path=image_path, extra_env={"ENABLE_GO_IOS_AGENT": "user", "SKIP_SIGNATURE_VERIFICATION": "1"}, **extra_opts),
# 方式3: 使用不同的环境变量
lambda: goios.image_mount(udid, path=image_path, extra_env={"ENABLE_GO_IOS_AGENT": "user", "GO_IOS_SKIP_SIGNATURE": "1"}, **extra_opts)
]
image_ok = False
for i, mount_attempt in enumerate(mount_attempts):
try:
app.logger.info("尝试挂载方式 %d", i + 1)
image_ok, image_out = mount_attempt()
if image_ok:
app.logger.info("挂载方式 %d 命令成功，验证镜像状态...", i + 1)
# 验证镜像是否真正挂载成功
list_ok, list_out = goios.image_list(udid, **extra_opts)
if list_ok and "none" not in list_out.lower():
app.logger.info("挂载方式 %d 验证成功: %s", i + 1, list_out)
image_ok = True
break
else:
app.logger.warning("挂载方式 %d 验证失败，实际未挂载: %s", i + 1, list_out)
image_ok = False
else:
app.logger.warning("挂载方式 %d 失败: %s", i + 1, image_out)
except Exception as e:
app.logger.warning("挂载方式 %d 异常: %s", i + 1, e)
if not image_ok:
app.logger.warning("所有挂载方式都失败，尝试直接截图")
# 即使挂载失败，也尝试直接截图（作为兜底）
retry_ok, retry_out = goios.screenshot(udid, save_path, extra_env={"ENABLE_GO_IOS_AGENT": "user"}, **extra_opts)
if not retry_ok:
error_msg = f"""开发者镜像挂载失败，截图也失败: {retry_out}
可能的解决方案:
1. 确保设备已开启开发者模式
2. 重新连接设备
3. 重启设备
4. 检查设备是否信任此电脑
5. 如果问题持续，请尝试在设备上手动安装开发者镜像
错误详情: {retry_out}"""
return jsonify({"ok": False, "msg": error_msg}), 500
else:
# 直接截图成功
first_ok, first_out = retry_ok, retry_out
else:
app.logger.error("未找到可用的开发者镜像文件，尝试直接截图")
# 没有镜像文件，尝试直接截图
retry_ok, retry_out = goios.screenshot(udid, save_path, extra_env={"ENABLE_GO_IOS_AGENT": "user"}, **extra_opts)
if not retry_ok:
error_msg = f"""未找到可用的开发者镜像文件，截图也失败: {retry_out}
可能的解决方案:
1. 确保设备已开启开发者模式
2. 重新连接设备
3. 重启设备
4. 检查设备是否信任此电脑
5. 如果问题持续，请尝试在设备上手动安装开发者镜像
错误详情: {retry_out}"""
return jsonify({"ok": False, "msg": error_msg}), 500
else:
# 直接截图成功
first_ok, first_out = retry_ok, retry_out
else:
app.logger.info("开发者镜像挂载成功")
# 如果镜像挂载成功，重试截图
if image_ok:
app.logger.info("开发者镜像挂载成功，重试截图")
retry_ok, retry_out = goios.screenshot(udid, save_path, extra_env={"ENABLE_GO_IOS_AGENT": "user"}, **extra_opts)
if not retry_ok:
# 兜底尝试：直接截图
app.logger.warning("带隧道参数截图失败，直接尝试: %s", retry_out)
retry_ok, retry_out = goios.screenshot(udid, save_path, extra_env={"ENABLE_GO_IOS_AGENT": "user"})
if not retry_ok:
error_msg = f"""开发者镜像已挂载，但截图失败: {retry_out}
可能的解决方案:
1. 确保设备已开启开发者模式
2. 重新连接设备
3. 重启设备
4. 检查设备是否信任此电脑
5. 如果问题持续，请尝试在设备上手动安装开发者镜像
错误详情: {retry_out}"""
return jsonify({"ok": False, "msg": error_msg}), 500
else:
# 将成功结果赋回
first_ok, first_out = retry_ok, retry_out
except Exception as exc:
app.logger.exception("截图重试过程中异常: %s", exc)
return jsonify({"ok": False, "msg": f"截图重试失败: {exc}"}), 500
try:
return send_from_directory(
app.config["UPLOAD_FOLDER"],
os.path.basename(first_out),
mimetype="image/png",
as_attachment=True,
download_name=display_name,
)
except Exception as exc:
app.logger.exception("发送截图文件失败: %s", exc)
return jsonify({"ok": False, "msg": f"发送文件失败: {exc}"}), 500
# ---------- 应用管理 ----------
@app.route("/api/apps")
def api_apps():
udid = request.args.get("udid", "").strip()
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
only_list = request.args.get("list", "0") == "1"
opts = extract_goios_opts(request.args)
apps_ok, apps_raw = run_with_quick_check_and_escalate(
prechecker,
udid,
lambda: goios.apps_list(udid, only_list=only_list, **opts)
)
apps = parse_ios_apps(apps_raw or "", third_party_only=True)
return jsonify({"ok": apps_ok, "apps": apps, "raw": apps_raw})
@app.route("/api/install", methods=["POST"])
def api_install():
udid = (request.form.get("udid") or "").strip()
file = request.files.get("file")
if not udid or not file:
return jsonify({"ok": False, "msg": "缺少 udid 或 IPA 文件"}), 400
install_ready, install_msg = prechecker.check_all(udid)
if not install_ready:
return jsonify({"ok": False, "msg": install_msg}), 400
fname_lower = (file.filename or "").lower()
if not fname_lower.endswith(".ipa"):
return jsonify({"ok": False, "msg": "仅支持 .ipa 文件"}), 400
filename = secure_filename(file.filename)
unique_name = f"{uuid.uuid4().hex}_{filename}"
save_path = os.path.join(app.config["UPLOAD_FOLDER"], unique_name)
file.save(save_path)
opts = extract_goios_opts(request.args or request.form or {})
install_ok, install_raw = goios.install_ipa(udid, save_path, **opts)
return jsonify({"ok": install_ok, "raw": install_raw, "filename": unique_name})
@app.route("/api/launch", methods=["POST"])
def api_launch():
data = request.get_json(force=True, silent=True) or {}
udid = (data.get("udid") or "").strip()
bundle_id = (data.get("bundle_id") or "").strip()
wait = bool(data.get("wait", False))
if not udid or not bundle_id:
return jsonify({"ok": False, "msg": "缺少 udid 或 bundle_id"}), 400
opts = extract_goios_opts(data)
launch_ok, launch_raw = run_with_quick_check_and_escalate(
prechecker,
udid,
lambda: goios.launch_app(udid, bundle_id, wait=wait, **opts)
)
return jsonify({"ok": launch_ok, "raw": launch_raw})
@app.route("/api/kill", methods=["POST"])
def api_kill():
data = request.get_json(silent=True) or {}
udid = (data.get("udid") or "").strip()
bundle_id = (data.get("bundle_id") or "").strip()
if not udid or not bundle_id:
return jsonify({"ok": False, "msg": "缺少 udid 或 bundle_id"}), 400
opts = extract_goios_opts(data)
kill_ok, kill_msg2 = run_with_quick_check_and_escalate(
prechecker,
udid,
lambda: goios.kill_app(udid=udid, bundle_id=bundle_id, **opts)
)
return jsonify({"ok": kill_ok, "msg": kill_msg2})
# ---------- 设备重启 ----------
@app.route("/api/reboot", methods=["POST"])
def api_reboot():
data = request.get_json(silent=True) or {}
udid = (data.get("udid") or "").strip()
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
opts = extract_goios_opts(data)
reboot_ok, reboot_raw = run_with_quick_check_and_escalate(
prechecker,
udid,
lambda: goios.reboot(udid, **opts)
)
return jsonify({"ok": reboot_ok, "raw": reboot_raw})
# ---------- 电量 ----------
@app.route("/api/battery")
def api_battery():
udid = request.args.get("udid", "").strip()
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
opts = extract_goios_opts(request.args)
battery_ok, battery_raw = run_with_quick_check_and_escalate(
prechecker,
udid,
lambda: goios.battery_info(udid, **opts)
)
return jsonify({"ok": battery_ok, "raw": battery_raw})
@app.route("/api/battery/detail")
def api_battery_detail():
udid = request.args.get("udid", "").strip()
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
opts = extract_goios_opts(request.args)
b_ok, b_raw = run_with_quick_check_and_escalate(prechecker, udid, lambda: goios.battery_info(udid, **opts))
r_ok, r_raw = run_with_quick_check_and_escalate(prechecker, udid, lambda: goios.battery_registry(udid, **opts))
detail = {"batterycheck": None, "batteryregistry": None}
try:
detail["batterycheck"] = json.loads(b_raw) if b_ok and b_raw else None
except json.JSONDecodeError:
detail["batterycheck"] = None
try:
detail["batteryregistry"] = json.loads(r_raw) if r_ok and r_raw else None
except json.JSONDecodeError:
detail["batteryregistry"] = None
return jsonify({"ok": True, "detail": detail, "raw": {"batterycheck": b_raw, "batteryregistry": r_raw}})
@app.route("/api/diskspace")
def api_diskspace():
udid = request.args.get("udid", "").strip()
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
opts = extract_goios_opts(request.args)
disk_ok, disk_raw = run_with_quick_check_and_escalate(
prechecker,
udid,
lambda: goios.diskspace(udid, **opts)
)
return jsonify({"ok": disk_ok, "raw": disk_raw})
@app.route("/api/diskspace/detail")
def api_diskspace_detail():
udid = request.args.get("udid", "").strip()
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
opts = extract_goios_opts(request.args)
disk_ok, disk_raw = run_with_quick_check_and_escalate(prechecker, udid, lambda: goios.diskspace(udid, **opts))
return jsonify({"ok": disk_ok, "raw": disk_raw})
# ---------- 开发者镜像 ----------
@app.route("/api/image/auto", methods=["POST"])
def api_image_auto():
data = request.get_json(silent=True) or {}
udid = (data.get("udid") or "").strip()
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
basedir = (data.get("basedir") or "").strip() or None
opts = extract_goios_opts(data)
image_ok, image_raw = run_with_quick_check_and_escalate(
prechecker,
udid,
lambda: goios.image_auto(udid, basedir, **opts)
)
return jsonify({"ok": image_ok, "raw": image_raw})
# ---------- 设备状态 ----------
@app.route("/api/devicestate/list")
def api_devicestate_list():
udid = request.args.get("udid", "").strip() or None
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
opts = extract_goios_opts(request.args)
state_ok, state_raw = run_with_quick_check_and_escalate(prechecker, udid, lambda: goios.devicestate_list(udid=udid, **opts))
return jsonify({"ok": state_ok, "raw": state_raw})
@app.route("/api/devicestate/enable", methods=["POST"])
def api_devicestate_enable():
data = request.get_json(force=True, silent=True) or {}
udid = (data.get("udid") or "").strip()
t = (data.get("profile_type_id") or "").strip()
p = (data.get("profile_id") or "").strip()
if not (udid and t and p):
return jsonify({"ok": False, "msg": "缺少 udid / profile_type_id / profile_id"}), 400
opts = extract_goios_opts(data)
state_enable_ok, state_enable_raw = run_with_quick_check_and_escalate(prechecker, udid, lambda: goios.devicestate_enable(udid, t, p, **opts))
return jsonify({"ok": state_enable_ok, "raw": state_enable_raw})
# ---------- 模拟位置 ----------
@app.route("/api/setlocation", methods=["POST"])
def api_setlocation():
data = request.get_json(force=True, silent=True) or {}
udid = (data.get("udid") or "").strip()
lat = data.get("lat"); lon = data.get("lon")
if not udid or lat is None or lon is None:
return jsonify({"ok": False, "msg": "缺少 udid/lat/lon"}), 400
opts = extract_goios_opts(data)
loc_ok, loc_raw = run_with_quick_check_and_escalate(prechecker, udid, lambda: goios.set_location(udid, float(lat), float(lon), **opts))
return jsonify({"ok": loc_ok, "raw": loc_raw})
# ---------- 截屏流 ----------
@app.route("/api/screenshot/stream/start", methods=["POST"])
def api_ss_start():
from backend_function.common_utils import check_port_available, find_free_port, wait_for_mjpeg_stream
payload = request.get_json(silent=True) or {}
udid = (payload.get("udid") or "").strip()
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
# 为提高成功率：在启动截图流前执行完整检查（包含隧道与开发者镜像挂载）
ready_ok, ready_msg = prechecker.check_all(udid)
if not ready_ok:
return jsonify({"ok": False, "msg": ready_msg}), 400
port = to_int(payload.get("port"), 3333) or 3333
# 检查端口是否已被占用
if not check_port_available(port):
new_port = find_free_port(start_port=port + 1, max_tries=10)
if new_port:
app.logger.warning("端口 %d 已占用，使用端口 %d", port, new_port)
port = new_port
else:
return jsonify({"ok": False, "msg": f"端口 {port} 已占用且无可用端口"}), 500
opts = extract_goios_opts(payload)
# 添加 tunnel 参数以提升连接稳定性
extra_opts = tunnel.get_goios_opts(udid) if hasattr(tunnel, 'get_goios_opts') else {}
opts.update(extra_opts)
# 注入环境变量提升兼容性
opts["extra_env"] = {"ENABLE_GO_IOS_AGENT": "user"}
app.logger.debug("启动 MJPEG 流: port=%d, opts=%s", port, opts)
p = goios.screenshot_stream_popen(udid=udid, port=port, **opts)
if p is None:
return jsonify({"ok": False, "msg": "启动 MJPEG 流失败"}), 500
# 等待 MJPEG 服务器启动
# 使用实际的服务器 IP 地址，而不是 localhost
server_ip = get_local_ip() or "127.0.0.1"
mjpeg_url = f"http://{server_ip}:{port}"
if not wait_for_mjpeg_stream(mjpeg_url, p, max_wait_seconds=15, logger=app.logger):
# 如果使用服务器 IP 失败，尝试 localhost 作为回退
if server_ip != "127.0.0.1":
app.logger.warning("使用服务器 IP %s 连接失败，尝试 localhost 回退", server_ip)
mjpeg_url_fallback = f"http://127.0.0.1:{port}"
if wait_for_mjpeg_stream(mjpeg_url_fallback, p, max_wait_seconds=10, logger=app.logger):
mjpeg_url = mjpeg_url_fallback
else:
# 两种方式都失败，终止进程
try:
p.terminate()
p.wait(timeout=5)
except (OSError, subprocess.TimeoutExpired):
pass
return jsonify({"ok": False, "msg": "MJPEG 流超时未就绪 (多种地址尝试失败)"}), 500
else:
# 超时，终止进程
try:
p.terminate()
p.wait(timeout=5)
except (OSError, subprocess.TimeoutExpired):
pass
return jsonify({"ok": False, "msg": "MJPEG 流超时未就绪 (15秒)"}), 500
sid = uuid.uuid4().hex
STREAMS.setdefault(udid, {})[sid] = {"p": p, "port": port}
parts = urlsplit(request.host_url)
url = urlunsplit((parts.scheme, f"{parts.hostname}:{port}", "", "", ""))
# 开启服务端录制
model = get_device_model(udid, goios)
ts = now_timestamp_str()
base = f"{model}_{udid}_record_{ts}"
# 启动进程监控线程
def monitor_process():
import time
while True:
# 从 STREAMS 中获取当前进程引用，避免闭包问题
if udid in STREAMS and sid in STREAMS[udid]:
current_p = STREAMS[udid][sid].get("p")
if current_p and current_p.poll() is not None:
app.logger.error("MJPEG 流进程意外退出 (退出码: %s)", current_p.returncode)
# 检查会话是否仍然有效（可能已被停止）
if udid in STREAMS and sid in STREAMS[udid]:
# 尝试重启进程
app.logger.debug("尝试重启 MJPEG 流...")
new_p = goios.screenshot_stream_popen(udid=udid, port=port, **opts)
if new_p:
STREAMS[udid][sid]["p"] = new_p
app.logger.debug("MJPEG 流已重启")
else:
app.logger.error("MJPEG 流重启失败")
break
else:
# 会话已被停止，退出监控
app.logger.debug("检测到会话已停止，退出监控线程")
break
else:
# 流会话已被清理，退出监控
app.logger.debug("流会话已被清理，退出监控线程")
break
time.sleep(5) # 每5秒检查一次
monitor_thread = threading.Thread(target=monitor_process, daemon=True, name=f"Monitor-{sid}")
monitor_thread.start()
# 延迟启动录制，确保 MJPEG 流完全稳定
import time
time.sleep(2.0) # 增加等待时间
rec_ctx = start_mjpeg_to_mp4(mjpeg_url=mjpeg_url, out_dir=app.config["UPLOAD_FOLDER"], basename=base, logger=app.logger)
STREAMS[udid][sid].update({
"rec_ctx": rec_ctx,
"rec_path": rec_ctx.get("path"),
"rec_name": rec_ctx.get("name"),
"monitor_thread": monitor_thread
})
return jsonify({
"ok": True,
"id": sid, # 前端期望 id 字段
"url": url,
"port": port,
"msg": f"MJPEG 流已启动 (端口: {port})"
})
# 安全下载令牌生成
# 已迁移至 common_utils.create_signed_download_token/consume_signed_download_token
@app.route("/api/screenshot/stream/stop", methods=["POST"])
def api_ss_stop():
payload = request.get_json(silent=True) or {}
app.logger.debug("停止录屏请求: %s", payload)
udid = (payload.get("udid") or "").strip()
stream_id = (payload.get("id") or "").strip() # 前端传递的是 id
app.logger.debug("解析参数: udid=%s, stream_id=%s", udid, stream_id)
app.logger.debug("当前活动流: %s", STREAMS)
if not udid or not stream_id:
app.logger.warning("缺少必要参数: udid=%s, stream_id=%s", udid, stream_id)
return jsonify({"ok": False, "msg": "缺少 udid/id"}), 400
# 先查找流信息
udid_streams = STREAMS.get(udid, {})
if stream_id not in udid_streams:
app.logger.warning("未找到流会话: udid=%s, stream_id=%s, 可用流: %s", udid, stream_id, list(udid_streams.keys()))
return jsonify({"ok": False, "msg": "无此会话"}), 404
# 获取流信息
info = udid_streams[stream_id]
if not isinstance(info, dict) or "p" not in info:
app.logger.warning("流信息格式错误: %s", info)
return jsonify({"ok": False, "msg": "会话信息无效"}), 404
# 停止进程
terminate_process(info["p"])
# 停止录制
rec_ctx = info.get("rec_ctx")
if rec_ctx:
stop_recorder(rec_ctx)
# 清理监控线程（如果存在）
monitor_thread = info.get("monitor_thread")
if monitor_thread and monitor_thread.is_alive():
app.logger.debug("清理监控线程")
# 监控线程是 daemon 线程，会在主线程退出时自动结束
# 这里主要是为了日志记录
# 删除流信息
del udid_streams[stream_id]
if not udid_streams: # 如果该设备没有其他流，删除设备条目
STREAMS.pop(udid, None)
rec_path = (rec_ctx or {}).get("path")
rec_name = (rec_ctx or {}).get("name")
# 生成签名下载链接
download_url = None
if rec_name and os.path.exists(rec_path or ""):
token = create_signed_download_token(rec_name, ttl_seconds=600)
download_url = f"/api/download_secure/{token}"
app.logger.debug("停止录屏成功: file=%s, download_url=%s", rec_name, download_url)
return jsonify({"ok": True, "file": rec_name, "download_url": download_url})
# ---------- 端口转发 ----------
@app.route("/api/forward/start", methods=["POST"])
def api_forward_start():
data = request.get_json(force=True, silent=True) or {}
udid = (data.get("udid") or "").strip()
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
fwd_ready, fwd_msg = prechecker.check_all(udid)
if not fwd_ready:
return jsonify({"ok": False, "msg": fwd_msg}), 400
host_port = to_int(data.get("host_port"), 0) or 0
target_port = to_int(data.get("target_port"), 0) or 0
if not host_port or not target_port:
return jsonify({"ok": False, "msg": "缺少 host_port/target_port"}), 400
opts = extract_goios_opts(data)
p = goios.forward_popen(udid=udid, host_port=host_port, target_port=target_port, **opts)
if p is None:
return jsonify({"ok": False, "msg": "启动失败"}), 500
fid = uuid.uuid4().hex
FORWARDS.setdefault(udid, {})[fid] = {"p": p, "host_port": host_port, "target_port": target_port}
return jsonify({"ok": True, "id": fid, "udid": udid})
@app.route("/api/forward/stop", methods=["POST"])
def api_forward_stop():
payload = request.get_json(silent=True) or {}
udid = (payload.get("udid") or "").strip()
fid = (payload.get("id") or "").strip()
if not udid or not fid:
return jsonify({"ok": False, "msg": "缺少 udid/id"}), 400
info = (FORWARDS.get(udid) or {}).pop(fid, None)
if not isinstance(info, dict) or "p" not in info:
return jsonify({"ok": False, "msg": "无此转发会话"}), 404
terminate_process(info["p"])
return jsonify({"ok": True})
# ---------- 文件下载 ----------
@app.route("/api/download/<path:fname>")
def api_download_file(fname: str):
# 仅允许从上传目录下载，并作为附件返回
safe_name = os.path.basename(fname)
full_path = os.path.join(app.config["UPLOAD_FOLDER"], safe_name)
if not os.path.exists(full_path):
return jsonify({"ok": False, "msg": "文件不存在"}), 404
return send_from_directory(app.config["UPLOAD_FOLDER"], safe_name, as_attachment=True)
# 新增：带签名的安全下载，短时有效
@app.route("/api/download_secure/<token>")
def api_download_secure(token: str):
try:
fname = consume_signed_download_token(token)
except KeyError:
return abort(403)
# 仅允许相对路径并限制在 UPLOAD_FOLDER 内
safe_rel = fname.replace("\", "/").lstrip("/")
full_path = os.path.normpath(os.path.join(app.config["UPLOAD_FOLDER"], safe_rel))
if not full_path.startswith(os.path.normpath(app.config["UPLOAD_FOLDER"])):
return abort(403)
if not os.path.exists(full_path):
return jsonify({"ok": False, "msg": "文件不存在"}), 404
directory = os.path.dirname(os.path.relpath(full_path, app.config["UPLOAD_FOLDER"]))
filename = os.path.basename(full_path)
base_dir = str(app.config["UPLOAD_FOLDER"]) # ensure str
dir_part = str(directory)
send_dir = base_dir if dir_part == "." or dir_part == "" else os.path.join(base_dir, dir_part)
resp = send_from_directory(send_dir, str(filename), as_attachment=True)
# 防缓存
resp.headers["Cache-Control"] = "no-store"
return resp
# ---------- 运行中应用（基于 ps --apps） ----------
@app.route("/api/apps/running")
def api_apps_running():
udid = request.args.get("udid", "").strip()
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
opts = extract_goios_opts(request.args)
ps_ok, ps_raw = run_with_quick_check_and_escalate(prechecker, udid, lambda: goios.ps_apps(udid, **opts))
# 容错解析
processes = parse_ps_apps_raw(ps_raw or "") if ps_ok and ps_raw else []
return jsonify({"ok": ps_ok, "list": processes, "raw": ps_raw})
# ---------- Crash 日志 ----------
@app.route("/api/crash/ls")
def api_crash_ls():
import os
udid = request.args.get("udid", "").strip()
pattern_input = (request.args.get("pattern") or "").strip()
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
ok_pre, pre_msg = prechecker.quick_check(udid)
if not ok_pre:
return jsonify({"ok": False, "msg": pre_msg}), 400
# 模式：含通配符(*, ?)走 go-ios 侧过滤；否则后端全量+不区分大小写包含匹配
has_glob = any(ch in pattern_input for ch in ("*", "?")) if pattern_input else False
go_pattern = pattern_input if has_glob else None
ok_list, raw = run_with_quick_check_and_escalate(
prechecker, udid, lambda: goios.crash_ls(udid, go_pattern)
)
items = parse_crash_ls_items(raw) if ok_list and raw else []
if ok_list and items and pattern_input and not has_glob:
kw = pattern_input.lower()
items = [x for x in items if kw in x.lower()]
# ✅ 只保留带扩展名的文件（过滤掉目录、无后缀）
items = [x for x in items if "." in os.path.basename(x)]
return jsonify({"ok": ok_list, "items": items, "raw": raw})
# ---------- Crash 日志 ----------
@app.route("/api/crash/cp", methods=["POST"])
def api_crash_cp():
data = request.get_json(silent=True) or {}
udid = (data.get("udid") or "").strip()
pattern = (data.get("pattern") or "*").strip() or "*"
patterns = data.get("patterns") if isinstance(data.get("patterns"), list) else None
# 仅允许复制到服务端的 UPLOAD_FOLDER/crashes/<uuid>/
crash_root = os.path.join(app.config["UPLOAD_FOLDER"], "crashes")
os.makedirs(crash_root, exist_ok=True)
# batch_dir = os.path.join(crash_root, uuid.uuid4().hex)
# os.makedirs(batch_dir, exist_ok=True)
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
# 收集导出文件（执行前 quick_check，失败升级）
def _do_export():
res = crash_export_collect(
goios, udid,
patterns if patterns else [pattern],
crash_root,
logger=app.logger
)
# 用 ok/raw 表示整体结果
return bool(res.get("ok")), res
ok_flag, result = run_with_quick_check_and_escalate(prechecker, udid, _do_export)
files = (result or {}).get("files") or []
app.logger.info("Crash 导出: udid=%s, ok=%s, files=%s", udid, ok_flag, files)
download_url = None
zip_name = ""
if ok_flag:
count = len(files)
if count == 1:
fp = files[0]
if os.path.exists(fp):
rel = str(os.path.relpath(fp, app.config["UPLOAD_FOLDER"]).replace("\", "/"))
token = create_signed_download_token(rel, ttl_seconds=1800)
download_url = f"/api/download_secure/{token}"
elif count >= 2:
# 2+ 文件统一打包
zip_path = crash_zip_dir((result or {}).get("batch_dir"), crash_root, logger=app.logger)
if zip_path and os.path.exists(zip_path):
zip_name = os.path.basename(zip_path)
rel = str(os.path.relpath(zip_path, app.config["UPLOAD_FOLDER"]).replace("\", "/"))
token = create_signed_download_token(rel, ttl_seconds=1800)
download_url = f"/api/download_secure/{token}"
return jsonify({
"ok": ok_flag,
"zip": zip_name,
"download_url": download_url,
"download_urls": None, # 预留字段，未来支持多链接
"raw": (result or {}).get("raw")
})
@app.route("/api/crash/rm", methods=["POST"])
def api_crash_rm():
import os
data = request.get_json(silent=True) or {}
udid = (data.get("udid") or "").strip()
cwd = (data.get("cwd") or ".").strip() or "."
pattern = (data.get("pattern") or "*").strip() or "*"
patterns = data.get("patterns") if isinstance(data.get("patterns"), list) else None
recursive = bool(data.get("recursive", True))
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
def _do_rm():
res = crash_remove_many(
goios, udid,
patterns if patterns else [pattern],
cwd=cwd,
recursive=recursive
)
return bool(res.get("ok")), res
ok_rm, res_rm = run_with_quick_check_and_escalate(prechecker, udid, _do_rm)
raw_output = (res_rm or {}).get("raw", "")
success = bool((res_rm or {}).get("ok"))
# === 提取删除的文件名（只保留带扩展名的文件） ===
deleted_files = []
try:
for line in raw_output.strip().splitlines():
if line.strip().startswith("{"):
obj = json.loads(line)
if obj.get("msg") == "delete" and "path" in obj:
path = obj["path"]
if "." in os.path.basename(path): # ✅ 只要带后缀的文件
deleted_files.append(path)
except Exception as e:
app.logger.warning("Crash 删除日志解析失败: %s", e)
# === 日志输出 ===
if deleted_files:
app.logger.info("Crash 删除: udid=%s, success=%s, deleted_files=%s",
udid, success, deleted_files)
else:
snippet = raw_output[:200] + "..." if len(raw_output) > 200 else raw_output
app.logger.info("Crash 删除: udid=%s, success=%s, raw_snippet=%s",
udid, success, snippet)
# === 错误返回 ===
if not success and raw_output:
if "permission denied" in raw_output.lower() or "access denied" in raw_output.lower():
error_msg = "权限不足，无法删除某些 crash 文件。可能需要管理员权限或设备解锁。"
elif "no such file" in raw_output.lower() or "file not found" in raw_output.lower():
error_msg = "部分文件不存在或已被删除。"
elif "usage:" in raw_output.lower() or "help" in raw_output.lower():
error_msg = "删除命令参数不正确，请检查 pattern 和 cwd 参数。"
else:
error_msg = f"删除失败: {raw_output}"
return jsonify({
"ok": False,
"msg": error_msg,
"raw": raw_output,
"diagnostic": {
"udid": udid,
"cwd": cwd,
"pattern": pattern,
"patterns": patterns,
"recursive": recursive,
"suggestion": "尝试使用更具体的 pattern 或检查文件权限"
}
})
return jsonify({
"ok": success,
"msg": "删除操作完成" if success else "删除操作失败",
"raw": raw_output,
"deleted_files": deleted_files
})
# ---------- 配置文件管理 ----------
@app.route("/api/profile/list")
def api_profile_list():
udid = request.args.get("udid", "").strip()
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
ok_l, raw_l = run_with_quick_check_and_escalate(prechecker, udid, lambda: goios.profile_list(udid))
items = []
if ok_l and raw_l:
try:
data = json.loads(raw_l)
if isinstance(data, list):
items = data # 保持原始结构（对象/字符串），交由前端格式化展示
except json.JSONDecodeError:
items = [s.strip() for s in str(raw_l).splitlines() if s.strip()]
return jsonify({"ok": ok_l, "items": items, "raw": raw_l})
@app.route("/api/profile/remove", methods=["POST"])
def api_profile_remove():
data = request.get_json(silent=True) or {}
udid = (data.get("udid") or "").strip()
names = data.get("names") or []
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
if not isinstance(names, list) or not names:
return jsonify({"ok": False, "msg": "缺少要移除的配置文件名"}), 400
results = []
all_ok = True
for profile_name in names:
ok_r, raw_r = run_with_quick_check_and_escalate(prechecker, udid, lambda pn=profile_name: goios.profile_remove(udid, str(pn)))
results.append({"name": profile_name, "ok": ok_r, "raw": raw_r})
all_ok = all_ok and ok_r
return jsonify({"ok": all_ok, "results": results})
# ---------- 开发者模式 ----------
@app.route("/api/devmode/get")
def api_devmode_get():
udid = request.args.get("udid", "").strip()
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
opts = extract_goios_opts(request.args)
devmode_ok, devmode_raw = run_with_quick_check_and_escalate(
prechecker,
udid,
lambda: goios.devmode_get(udid, **opts)
)
return jsonify({"ok": devmode_ok, "raw": devmode_raw})
@app.route("/api/devmode/check")
def api_devmode_check():
"""
开发者模式检测，确保tunnel状态正常
"""
udid = request.args.get("udid", "").strip()
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
try:
# 使用prechecker的check_all方法，它会自动处理tunnel状态
ok_pre, pre_msg = prechecker.check_all(udid)
if not ok_pre:
return jsonify({"ok": False, "raw": f"设备检查失败: {pre_msg}"}), 500
# 获取tunnel参数
tunnel_opts = tunnel.get_goios_opts(udid) if hasattr(tunnel, 'get_goios_opts') else {}
# 调用go-ios命令检测开发者模式
devmode_ok, devmode_raw = goios.devmode_get(udid, **tunnel_opts)
return jsonify({"ok": devmode_ok, "raw": devmode_raw})
except Exception as e:
app.logger.exception("开发者模式检测异常: %s", e)
return jsonify({"ok": False, "raw": str(e)}), 500
@app.route("/api/devmode/enable", methods=["POST"])
def api_devmode_enable():
data = request.get_json(silent=True) or {}
udid = (data.get("udid") or "").strip()
enable_post_restart = bool(data.get("enable_post_restart", True))
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
opts = extract_goios_opts(data)
devmode_ok, devmode_raw = run_with_quick_check_and_escalate(
prechecker,
udid,
lambda: goios.devmode_enable(udid, enable_post_restart=enable_post_restart, **opts)
)
return jsonify({"ok": devmode_ok, "raw": devmode_raw})
# ---------- 辅助功能 ----------
@app.route("/api/assistive/<feature>/<action>", methods=["POST"])
def api_assistive(feature: str, action: str):
data = request.get_json(silent=True) or {}
udid = (data.get("udid") or "").strip()
force = bool(data.get("force", False))
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
feature = feature.strip().lower()
action = action.strip().lower()
if feature not in ("assistivetouch", "voiceover", "zoom"):
return jsonify({"ok": False, "msg": "不支持的功能"}), 400
if action not in ("enable", "disable", "toggle", "get"):
return jsonify({"ok": False, "msg": "不支持的操作"}), 400
ok_a, raw_a = run_with_quick_check_and_escalate(prechecker, udid, lambda: goios.assistive(udid, feature, action, force=force))
return jsonify({"ok": ok_a, "raw": raw_a})
# ---------- 设备事件（SSE） ----------
@app.route("/api/devices/events")
def api_devices_events():
def _gen():
for chunk in listen_event_stream(goios):
yield chunk
return Response(stream_with_context(_gen()), mimetype='text/event-stream')
# ---------- 系统日志（启动/停止，保存为文件供下载） ----------
@app.route("/api/syslog/start", methods=["POST"])
def api_syslog_start():
data = request.get_json(silent=True) or {}
udid = (data.get("udid") or "").strip()
parse = bool(data.get("parse", True))
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
ok_pre, pre_msg = prechecker.quick_check(udid)
if not ok_pre:
return jsonify({"ok": False, "msg": pre_msg}), 400
sid = uuid.uuid4().hex
# 使用通用工具启动写文件会话
ctx = syslog_start_session(goios, udid, app.config["UPLOAD_FOLDER"], parse=parse, logger=app.logger)
if not ctx.get("ok"):
return jsonify({"ok": False, "msg": ctx.get("msg") or "启动失败"}), 500
SYSLOGS.setdefault(udid, {})[sid] = {"p": ctx.get("p"), "path": ctx.get("path"), "name": ctx.get("name")}
return jsonify({"ok": True, "id": sid, "file": ctx.get("name")})
@app.route("/api/syslog/stream")
def api_syslog_stream():
udid = (request.args.get("udid") or "").strip()
parse = request.args.get("parse", "1") != "0"
keywords = [s.strip() for s in (request.args.get("kw") or "").split(",") if s.strip()]
levels = [s.strip().lower() for s in (request.args.get("lv") or "").split(",") if s.strip()]
if not udid:
return jsonify({"ok": False, "msg": "缺少 udid"}), 400
# 使用 quick_check，必要时前端根据报错提示
ok_pre, pre_msg = prechecker.quick_check(udid)
if not ok_pre:
return jsonify({"ok": False, "msg": pre_msg}), 400
def _gen():
# 查找现有的 syslog 进程
existing_process = None
for session_info in SYSLOGS.get(udid, {}).values():
if session_info.get("p"):
existing_process = session_info["p"]
break
for chunk in stream_syslog_sse(goios, udid, parse=parse, logger=app.logger,
keywords=keywords, levels=levels,
existing_process=existing_process):
yield chunk
resp = Response(stream_with_context(_gen()), mimetype='text/event-stream')
resp.headers["Cache-Control"] = "no-cache"
resp.headers["X-Accel-Buffering"] = "no" # 兼容反代
resp.headers["Connection"] = "keep-alive"
return resp
@app.route("/api/syslog/stop", methods=["POST"])
def api_syslog_stop():
data = request.get_json(silent=True) or {}
udid = (data.get("udid") or "").strip()
sid = (data.get("id") or "").strip()
if not udid or not sid:
return jsonify({"ok": False, "msg": "缺少 udid/id"}), 400
info = (SYSLOGS.get(udid) or {}).pop(sid, None)
if not info or "p" not in info:
return jsonify({"ok": False, "msg": "无此会话"}), 404
terminate_process(info["p"])
log_name = info.get("name")
syslog_file_path = info.get("path")
download_url = None
if log_name and syslog_file_path and os.path.exists(syslog_file_path):
rel = str(os.path.relpath(syslog_file_path, app.config["UPLOAD_FOLDER"]).replace("\", "/"))
token = create_signed_download_token(rel, ttl_seconds=1800)
download_url = f"/api/download_secure/{token}"
return jsonify({"ok": True, "download_url": download_url})
# ========== 健康检查 ==========
@app.route("/health")
def health():
return jsonify({"ok": True}), 200
# ========== Debug功能 ==========
@app.route("/api/debug/cleanup-processes", methods=["POST"])
def api_debug_cleanup_processes():
"""手动清理所有ios.exe进程"""
try:
cleanup_all_ios_processes(app.logger)
return jsonify({"ok": True, "msg": "ios.exe进程清理完成"})
except Exception as e:
app.logger.exception("手动清理ios.exe进程异常: %s", e)
return jsonify({"ok": False, "msg": f"清理失败: {str(e)}"}), 500
@app.route("/api/debug/export-logs", methods=["POST"])
def api_debug_export_logs():
"""导出应用日志文件用于调试分析"""
try:
# 源日志文件路径
source_log_path = os.path.join(app.config["LOG_DIR"], "ios_app.log")
# 检查源文件是否存在
if not os.path.exists(source_log_path):
return jsonify({"ok": False, "msg": "日志文件不存在"}), 404
# 生成副本文件名（带时间戳）
timestamp = now_timestamp_str()
copy_filename = f"debug_logs_{timestamp}.log"
copy_path = os.path.join(app.config["UPLOAD_FOLDER"], copy_filename)
# 复制文件内容（流式读取，避免大文件内存问题）
try:
with open(source_log_path, 'r', encoding='utf-8') as src_file:
with open(copy_path, 'w', encoding='utf-8') as dst_file:
# 分块读取和写入，避免内存占用过大
chunk_size = 8192 # 8KB chunks
while True:
chunk = src_file.read(chunk_size)
if not chunk:
break
dst_file.write(chunk)
# 生成下载链接
rel = str(os.path.relpath(copy_path, app.config["UPLOAD_FOLDER"]).replace("\", "/"))
token = create_signed_download_token(rel, ttl_seconds=3600) # 1小时有效期
download_url = f"/api/download_secure/{token}"
app.logger.info("Debug日志导出成功: %s", copy_filename)
return jsonify({
"ok": True,
"filename": copy_filename,
"download_url": download_url,
"msg": "日志导出成功"
})
except (OSError, IOError) as e:
app.logger.error("复制日志文件失败: %s", e)
return jsonify({"ok": False, "msg": f"复制文件失败: {str(e)}"}), 500
except Exception as e:
app.logger.exception("Debug日志导出异常: %s", e)
return jsonify({"ok": False, "msg": f"导出失败: {str(e)}"}), 500
# ===== 自启动浏览器 =====
use_local_ip = get_local_ip()
def open_browser():
time.sleep(1)
cert = app.config.get("SSL_CERT_FILE")
key = app.config.get("SSL_KEY_FILE")
scheme = "https" if (cert and key and os.path.exists(cert) and os.path.exists(key)) else "http"
webbrowser.open_new_tab(f'{scheme}://{use_locals_ip}:5001')
if __name__ == '__main__':
if not hasattr(app, 'browser_opened') or not app.browser_opened:
threading.Thread(target=open_browser, daemon=True).start()
app.browser_opened = True
app.logger.info("Web 控制台启动中： http://%s:%s", use_local_ip, 5001)
ssl_cert = app.config.get("SSL_CERT_FILE")
ssl_key = app.config.get("SSL_KEY_FILE")
ssl_ctx = None
if ssl_cert and ssl_key and os.path.exists(ssl_cert) and os.path.exists(ssl_key):
ssl_ctx = (ssl_cert, ssl_key)
app.logger.info("以 HTTPS 启动: https://%s:%s", use_local_ip, 5001)
app.run(host=use_local_ip, port=5001, debug=True, use_reloader=False, ssl_context=ssl_ctx)3、HTML代码
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Web App For IOS Debugger</title>
<!-- 本地 Bootstrap -->
<link rel="stylesheet" href="/static/wcss/bootstrap.min.css">
<!-- 本地 FontAwesome -->
<link rel="stylesheet" href="/static/wcss/all.min.css">
<!-- 自定义样式（外部 CSS 文件） -->
<link rel="stylesheet" href="/static/wcss/style.css">
<!-- SweetAlert -->
<link rel="stylesheet" href="/static/wcss/sweetalert2.min.css">
</head>
<body class="bg-light d-flex flex-column min-vh-100">
<!-- Header -->
<header class="container my-4">
<div class="d-flex justify-content-between align-items-center">
<div class="flex-grow-1 text-center">
<h1>Web App For IOS Debugger</h1>
</div>
<div class="guide-entry-wrapper">
<button id="guide-entry-btn" class="btn btn-outline-primary guide-entry-btn" title="使用指南">
📖 使用指南
</button>
</div>
</div>
</header>
<main class="container">
<div class="row g-3">
<!-- 设备管理 -->
<div class="col-12 col-md-6 col-lg-3">
<div class="card shadow-sm">
<div class="card-header text-center">
<h2 class="card-title">设备管理</h2>
</div>
<div class="card-body">
<select id="device-select" class="form-select mb-2 text-center-select">
<option value="">加载中...</option>
</select>
<div class="row g-2 mb-3">
<div class="col-6"><button id="refresh-devices-button" class="btn btn-primary w-100">刷新设备</button></div>
<div class="col-6"><button id="dm-reboot-button" class="btn btn-warning w-100">重启设备</button></div>
<div id="devmode-status-tip" class="small text-muted mt-2"></div>
</div>
<div class="d-grid gap-2 mb-3">
<button id="get-device-info-button" class="btn btn-success">获取设备信息</button>
<div id="device-info" class="border p-2 small info-list"></div>
</div>
<div class="border rounded p-2">
<div class="small fw-bold mb-2">辅助功能</div>
<div class="row g-2">
<div class="col-12">
<select id="assistive-feature-select" class="form-select">
<option value="assistivetouch">AssistiveTouch</option>
<option value="voiceover">VoiceOver</option>
<option value="zoom">Zoom</option>
</select>
</div>
<div class="col-12">
<div class="row g-2">
<div class="col-6"><button id="assistive-enable-button" class="btn btn-success w-100">开启</button></div>
<div class="col-6"><button id="assistive-disable-button" class="btn btn-danger w-100">关闭</button></div>
</div>
</div>
</div>
<div id="assistive-tip" class="form-text mt-2"></div>
</div>
</div>
</div>
</div>
<!-- 应用管理 -->
<div class="col-12 col-md-6 col-lg-3">
<div class="card shadow-sm">
<div class="card-header text-center">
<h2 class="card-title">应用管理</h2>
</div>
<div class="card-body">
<div class="mb-2">
<button id="get-installed-apps-button" class="btn btn-warning w-100">
获取已安装的三方应用
</button>
</div>
<input type="text" id="app-search-input" class="form-control mb-2" placeholder="搜索名称或 Bundle ID">
<select id="app-select" class="form-select mb-2"></select>
<div class="row g-2 mb-3">
<div class="col-6"><button id="launch-app-button" class="btn btn-success w-100">启动应用</button></div>
<div class="col-6"><button id="stop-app-button" class="btn btn-danger w-100">停止应用运行</button></div>
</div>
<div class="d-grid gap-2 mb-3">
<button id="apps-running-button" class="btn btn-primary w-100">查看运行中应用</button>
<div id="apps-running-list" class="small border p-2 mb-3"></div>
</div>
<div class="mb-2">
<input type="file" id="ipa-file" accept=".ipa" class="form-control">
<div class="form-text text-muted">仅支持单个 .ipa 文件</div>
</div>
<button id="install-ipa-button" class="btn btn-success w-100 mb-3">安装 IPA</button>
<div id="install-result-tip" class="small text-muted mt-2"></div>
</div>
</div>
</div>
<!-- 截图与录屏 -->
<div class="col-12 col-md-6 col-lg-3">
<div class="card shadow-sm">
<div class="card-header text-center">
<h2 class="card-title">截图与录屏</h2>
</div>
<div class="card-body text-center">
<!-- 去掉 onclick，事件由 app_ios.js 统一绑定 -->
<div class="row g-2 mb-2">
<div class="col-6"><button id="take-screenshot-button" class="btn btn-warning w-100">设备截图</button></div>
<div class="col-6"><button id="record-screen-button" class="btn btn-primary w-100">设备录屏</button></div>
</div>
<div id="screenshot-preview"></div>
</div>
</div>
</div>
<!-- 日志与配置管理 -->
<div class="col-12 col-md-6 col-lg-3">
<div class="card shadow-sm">
<div class="card-header text-center">
<h2 class="card-title">日志与配置</h2>
</div>
<div class="card-body">
<div class="border rounded p-2 mb-3">
<div class="small fw-bold mb-2">Crash日志管理</div>
<input type="text" id="crash-search-input" class="form-control mb-2" placeholder="搜索关键字或通配 (*ips*)">
<div class="mb-2">
<button id="crash-list-button" class="btn btn-primary w-100">Crash日志获取</button>
</div>
<div class="custom-multiselect mb-2" id="crash-multi">
<button type="button" class="btn btn-light border trigger w-100" id="crash-multi-trigger" disabled>请选择日志（先点击👆获取）</button>
<div class="menu" id="crash-multi-menu"></div>
</div>
<select id="crash-select" class="form-select mb-2"></select>
<div class="row g-2">
<div class="col-6"><button id="crash-export-button" class="btn btn-success w-100">导出所选</button></div>
<div class="col-6"><button id="crash-delete-button" class="btn btn-danger w-100">删除所选</button></div>
</div>
<div id="crash-export-tip" class="form-text mt-2"></div>
</div>
<div class="border rounded p-2 mb-3">
<div class="small fw-bold mb-2">配置文件管理</div>
<div class="row g-2 mb-2">
<div class="col-6"><button id="profile-list-button" class="btn btn-success w-100">配置获取</button></div>
<div class="col-6"><button id="profile-remove-button" class="btn btn-danger w-100">移除所选</button></div>
</div>
<div id="profile-multi" class="mb-2">
<button id="profile-multi-trigger" class="btn btn-light border trigger w-100" type="button" disabled>请选择配置（先点击👆获取）</button>
<div id="profile-multi-menu" class="border rounded bg-white p-2 mt-1"></div>
</div>
<select id="profile-select" class="form-select mb-2" multiple size="5"></select>
<div id="profile-tip" class="form-text mt-2"></div>
</div>
<div class="border rounded p-2 mb-3">
<div class="small fw-bold mb-2">系统日志</div>
<div class="row g-2 mb-2">
<div class="col-6"><button id="syslog-start-button" class="btn btn-primary w-100">开始系统日志</button></div>
<div class="col-6"><button id="syslog-stop-button" class="btn btn-danger w-100">停止并下载</button></div>
</div>
<div class="mt-2 d-grid gap-2 gap-md-2">
<div class="row g-2 align-items-center">
<div class="col-12"><input id="syslog-kw" type="text" class="form-control form-control-sm" placeholder="关键字/进程/模块（逗号分隔）"></div>
<div class="col-12">
<select id="syslog-lv" class="form-select form-select-sm">
<option value="">全部级别</option>
<option value="info">info</option>
<option value="warning">warning</option>
<option value="error">error</option>
</select>
</div>
</div>
<div class="d-flex gap-2 mt-2">
<button type="button" class="btn btn-success w-100" id="syslog-apply-filter">应用筛选</button>
<button type="button" class="btn btn-danger w-100" id="syslog-clear-live">清空日志</button>
</div>
</div>
<div id="syslog-download-link" class="form-text mt-2"></div>
<div id="syslog-live" class="border rounded p-2">
<div id="syslog-placeholder">日志预览区域</div>
</div>
</div>
</div>
</div>
</div>
</div>
</main>
<!-- Loading 遮罩 -->
<div id="loading-overlay">
<div class="loading-spinner"></div>
<div class="loading-text">Loading...</div>
</div>
<!-- Footer -->
<footer class="footer mt-auto py-3 text-center border-top">
<div class="container">
<span class="text-muted" id="debug-trigger">© 2024–2025 Web App for iOS Debugger. All rights reserved.</span>
</div>
</footer>
<!-- JS -->
<script src="/static/wjss/jquery.min.js"></script>
<script src="/static/wjss/sweetalert2.all.min.js"></script>
<script src="/static/wjss/bootstrap.min.js"></script>
<script src="/static/wjss/app_ios.js"></script>
</body>
</html>4、JS代码
// -*- coding: utf-8 -*-
// /static/wjss/app_ios.js
// 依赖：jQuery、SweetAlert2
(function (global) {
"use strict";
const API = {
refreshDevices: "/api/devices", // GET -> { ok, devices:[] } 或 { deviceList:[] }
getDeviceInfo: "/api/device_info", // GET ?udid=...
screenshot: "/api/screenshot", // GET ?udid=... -> image blob
// 新增的三个后端端点（已有后端实现）：
apps: "/api/apps", // GET ?udid=...&list=0 -> {ok, raw}
kill: "/api/kill", // POST JSON {udid, bundle_id}
install: "/api/install", // POST FormData {udid, file}
// 录屏流（后端已提供）
streamStart: "/api/screenshot/stream/start",
streamStop: "/api/screenshot/stream/stop",
// 启动与重启
launch: "/api/launch",
reboot: "/api/reboot",
// 新增
appsRunning: "/api/apps/running",
crashLs: "/api/crash/ls",
crashCp: "/api/crash/cp",
crashRm: "/api/crash/rm",
devmodeGet: "/api/devmode/get",
devmodeCheck: "/api/devmode/check", // 轻量级检测，不依赖tunnel
devmodeEnable: "/api/devmode/enable",
profileList: "/api/profile/list",
profileRemove: "/api/profile/remove",
assistive: (feature, action) => `/api/assistive/${feature}/${action}`,
deviceEvents: "/api/devices/events",
syslogStart: "/api/syslog/start",
syslogStop: "/api/syslog/stop",
batteryDetail: "/api/battery/detail",
diskDetail: "/api/diskspace/detail",
debugExportLogs: "/api/debug/export-logs",
debugCleanupProcesses: "/api/debug/cleanup-processes",
};
// —— 小工具 ——
let _refreshingDevices = false;
let _allThirdPartyApps = []; // {bundleId, name?, version?}
let _streamSession = null; // { id, udid, url }
function showLoading(text) {
const el = document.getElementById("loading-overlay");
if (el) {
const t = el.querySelector(".loading-text");
if (t) t.textContent = text || "Loading...";
el.style.display = "flex";
}
}
function hideLoading() {
const el = document.getElementById("loading-overlay");
if (el) el.style.display = "none";
}
async function alertAndHide(opts) {
// 解决遮罩与提示冲突：弹窗前隐藏遮罩
hideLoading();
try {
if (opts && String(opts.icon) === "success") {
return Promise.resolve({ dismissed: true });
}
return Swal.fire(opts);
} catch (e) {
return Promise.resolve();
}
}
// 通用：fetch JSON（非200抛错），以及POST JSON封装
async function apiFetchJSON(url, options) {
const res = await fetch(url, options);
if (!res.ok) {
let msg = "HTTP " + res.status;
try { const j = await res.json(); msg = j.msg || j.raw || msg; } catch {}
throw new Error(msg);
}
return res.json();
}
function apiPostJSON(url, obj) {
return apiFetchJSON(url, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(obj || {})
});
}
// 统一通知封装
function notifySuccess(title, text, timer=1200) { return alertAndHide({ icon: "success", title, text, timer }); }
function notifyError(title, text, timer=3000) { return alertAndHide({ icon: "error", title, text, timer }); }
function notifyInfo(title, text, timer=1800) { return alertAndHide({ icon: "info", title, text, timer }); }
function notifyWarning(title, text, timer=2000) { return alertAndHide({ icon: "warning", title, text, timer }); }
function setBusy($btn, busy) {
if ($btn && $btn.length) $btn.prop("disabled", !!busy);
}
// 受控日志（默认静默；仅当 window.__IOSAPP_DEBUG__ 为真时输出）
function logDebug(){ try { if (window && window.__IOSAPP_DEBUG__) console.debug.apply(console, arguments); } catch(_){} }
function logWarn(){ try { if (window && window.__IOSAPP_DEBUG__) console.warn.apply(console, arguments); } catch(_){} }
function getUDID() {
return $("#device-select").val() || "";
}
function syncHiddenUdid() {
$("#selected-udid").val(getUDID());
}
// 补充通用函数（避免未定义）
async function withUdid($btn, fn) {
const udid = getUDID();
if (!udid) {
await alertAndHide({ icon: "warning", title: "请选择设备", timer: 2000 });
if ($btn && $btn.length) setBusy($btn, false);
return null;
}
return fn && fn(udid);
}
function bindDocNS(eventName, ns, selector, handler) {
$(document).off(`${eventName}.${ns}`).on(`${eventName}.${ns}`, selector, handler);
}
function bindDocGlobal(eventName, ns, handler) {
$(document).off(`${eventName}.${ns}`).on(`${eventName}.${ns}`, handler);
}
function fmtModel(model) {
if (!model) return "";
const s = String(model).trim();
return s.split(",")[0];
}
function sectionTitle(text, extraClass) {
const cls = extraClass ? ("info-section-title " + extraClass) : "info-section-title";
return `<div class="${cls}">${text}：</div>`;
}
function shouldHideBaseLabel(label) {
const s = String(label || "").trim();
// 兼容 Wi‑Fi 的不同连字符：普通连字符- 或非折行连字符‑
return /(Wi[-‑]?Fi地址|蓝牙地址|以太网地址)/i.test(s);
}
// =========================
// 设备列表
// =========================
async function refreshDevices() {
if (_refreshingDevices) return;
_refreshingDevices = true;
const $btn = $("#refresh-devices-button");
setBusy($btn, true);
try {
const res = await fetch(API.refreshDevices + "?details=1", { cache: "no-store" });
if (!res.ok) throw new Error("HTTP " + res.status);
const data = await res.json();
// 归一化
const raw = Array.isArray(data) ? data : (data.devices || data.deviceList || []);
const items = raw.map(d => {
if (typeof d === "string") return { name: d, udid: d, model: "" };
const udid = d.udid || d.UDID || d.Udid;
const model = d.model || d.ProductType || "";
const name = d.name || d.ProductName || model || udid;
return { name, udid, model };
});
// 去重（按 UDID）
const seen = new Set();
const uniqueItems = [];
for (const it of items) {
if (!it.udid || seen.has(it.udid)) continue;
seen.add(it.udid);
uniqueItems.push(it);
}
const $select = $("#device-select");
const $list = $("#ios-device-list");
$select.empty(); $list.empty();
if (uniqueItems.length === 0) {
await alertAndHide({ icon: "warning", title: "未检测到设备", text: "请检查连接。", timer: 2500 });
$select.html('<option value="">未检测到设备</option>');
$list.append('<li class="list-group-item text-muted">未检测到设备</li>');
$("#selected-udid").val("");
// 没有设备时显示提示
showDevModeStatus("no-device");
return;
}
// 标签：型号 · UDID；值仍然是 UDID
$select.html(uniqueItems.map(it => {
const label = (it.model ? fmtModel(it.model) + " · " : "") + it.udid;
return `<option value="${it.udid}">${label}</option>`;
}).join(""));
$list.html(uniqueItems.map(it => {
const label = (it.model ? fmtModel(it.model) + " · " : "") + it.udid;
const title = it.name || it.model || "iOS 设备";
return `<li class="list-group-item"><b>${title}</b><br>${label}</li>`;
}).join(""));
$("#selected-udid").val(uniqueItems[0].udid);
// 触发设备选择变化事件，启动开发者模式检测
setTimeout(() => {
$("#device-select").trigger("change");
}, 100);
} catch (e) {
await alertAndHide({ icon: "error", title: "刷新设备失败", text: e.message || "未知错误", timer: 3000 });
} finally {
setBusy($btn, false);
_refreshingDevices = false;
}
}
// =========================
// 设备信息
// =========================
async function getDeviceInfo() {
const $btn = $("#get-device-info-button");
setBusy($btn, true);
const udid = getUDID();
if (!udid) {
await alertAndHide({ icon: "warning", title: "请选择设备", timer: 2000 });
return setBusy($btn, false);
}
try {
// 并行请求：基础信息 + 电池合并 + 磁盘
const [resBase, resBat, resDisk] = await Promise.all([
fetch(`${API.getDeviceInfo}?udid=${encodeURIComponent(udid)}`, { cache: "no-store" }),
fetch(`${API.batteryDetail}?udid=${encodeURIComponent(udid)}`, { cache: "no-store" }),
fetch(`${API.diskDetail}?udid=${encodeURIComponent(udid)}`, { cache: "no-store" }),
]);
const base = await resBase.json().catch(()=>({ok:false}));
const bat = await resBat.json().catch(()=>({ok:false}));
const dsk = await resDisk.json().catch(()=>({ok:false}));
const baseList = Array.isArray(base.info) ? base.info : [];
// 过滤不展示项
const filtered = baseList.filter(it => !shouldHideBaseLabel(it.label));
const head = filtered.slice(0, 4);
const tail = filtered.slice(4);
const headHtml = head.map(it => {
const label = String(it.label || "");
let value = String(it.value ?? "");
if (label === "产品型号") value = fmtModel(value);
return `<div><span class="info-k">${label}：</span> ${value}</div>`;
}).join("") || '<div class="text-muted">暂无可显示的设备信息</div>';
const tailHtml = tail.map(it => {
const label = String(it.label || "");
let value = String(it.value ?? "");
if (label === "产品型号") value = fmtModel(value);
return `<div><span class="info-k">${label}：</span> ${value}</div>`;
}).join("");
let html = sectionTitle("常规设备信息") + headHtml;
if (tail.length) {
html += `
<div class="mt-1">
<a href="#" class="small">展开更多</a>
<div class="mt-2">${tailHtml}</div>
</div>`;
}
// 电池（默认折叠，统一纳入更多区）
const ck = (bat && bat.detail && bat.detail.batterycheck) || {};
const rg = (bat && bat.detail && bat.detail.batteryregistry) || {};
const bCap = (ck.BatteryCurrentCapacity != null) ? `${ck.BatteryCurrentCapacity}%` : "UNKNOWN";
const temp = (rg.Temperature != null) ? (rg.Temperature/100).toFixed(2) + "°C" : "UNKNOWN";
const volt = (rg.Voltage != null) ? (rg.Voltage/1000).toFixed(3) + " V" : "UNKNOWN";
const design = (rg.DesignCapacity != null) ? `${rg.DesignCapacity} mAh` : "UNKNOWN";
const nominal = (rg.NominalChargeCapacity != null) ? `${rg.NominalChargeCapacity} mAh` : "UNKNOWN";
const batteryInner = `
<div><span class="info-k">电池容量：</span> ${bCap}</div>
<div><span class="info-k">电池温度：</span> ${temp}</div>
<div><span class="info-k">电池电压：</span> ${volt}</div>
<div><span class="info-k">设计容量：</span> ${design}</div>
<div><span class="info-k">标称容量：</span> ${nominal}</div>
`;
// 磁盘（默认折叠，统一纳入更多区）
let bs = "UNKNOWN", free = "UNKNOWN", used = "UNKNOWN", total = "UNKNOWN";
if (dsk && dsk.ok && dsk.raw) {
let parsed = false;
try {
const disk = JSON.parse(dsk.raw);
if (disk) {
bs = disk.BlockSize != null ? `${disk.BlockSize}KB` : "UNKNOWN";
free = disk.FreeSpace || disk.free || "UNKNOWN";
used = disk.UsedSpace || disk.used || "UNKNOWN";
total = disk.TotalSpace || disk.total || "UNKNOWN";
parsed = true;
}
} catch {}
if (!parsed) {
const text = String(dsk.raw);
const mBlock = text.match(/BlockSize:s*([0-9.]+)/i);
const mFree = text.match(/FreeSpace:s*([^
]+)/i);
const mUsed = text.match(/UsedSpace:s*([^
]+)/i);
const mTotal = text.match(/TotalSpace:s*([^
]+)/i);
bs = mBlock ? `${mBlock[1]}KB` : bs;
free = mFree ? mFree[1].trim() : free;
used = mUsed ? mUsed[1].trim() : used;
total = mTotal ? mTotal[1].trim() : total;
}
}
const diskInner = `
<div><span class="info-k">存储块规格：</span> ${bs}</div>
<div><span class="info-k">闲置空间：</span> ${free}</div>
<div><span class="info-k">已使用空间：</span> ${used}</div>
<div><span class="info-k">总体空间：</span> ${total}</div>
`;
// 统一更多区内容（包含：常规尾部 + 电池 + 磁盘）稍后 append 进 #device-info-more
const extraInfoHtml = `
<div class="info-section-title mt-2">电池信息：</div>
<div class="mt-2">${batteryInner}</div>
<div class="info-section-title mt-2">磁盘信息：</div>
<div class="mt-2">${diskInner}</div>
`;
$("#device-info").html(html);
// 将电池与磁盘信息附加到统一折叠区
if (document.getElementById("device-info-more")) {
$("#device-info-more").append(extraInfoHtml).hide();
}
// 移除顶部的文字/图标折叠触发器（若存在）
if (document.getElementById("device-info-toggle")) {
$("#device-info-toggle").remove();
}
// 修正展开区容器多余外边距，避免"系统版本"与"CPU 架构"之间出现空行
(function(){
const $more = $("#device-info-more");
if ($more.length) {
$more.removeClass("mt-2");
const $wrap = $more.parent();
try {
if ($wrap && $wrap.length) {
$more.appendTo("#device-info");
$wrap.removeClass("mt-1");
if ($wrap.children().length === 0) $wrap.remove();
}
} catch(_) {}
}
})();
// 添加底部居中的折叠按钮（内联SVG双箭头）
if (!document.getElementById("device-info-toggle-btn")) {
$("#device-info").append(
'<div class="info-collapse-btn-wrapper">' +
'<button type="button" class="btn btn-light border info-collapse-btn" aria-expanded="false" title="展开/收起" aria-label="展开/收起">' +
'<svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round" width="14" height="14" aria-hidden="true">' +
'<polyline points="6 7 12 13 18 7"></polyline>' +
'<polyline points="6 13 12 19 18 13"></polyline>' +
'</svg>' +
'</button>' +
'</div>'
);
}
// 绑定按钮点击，平滑展开/收起，并保持按钮始终在底部
$("#device-info-toggle-btn").off("click").on("click", function(e){
e.preventDefault();
const $more = $("#device-info-more");
const isOpen = $more.is(":visible");
if (isOpen) {
$more.slideUp(200);
} else {
$more.slideDown(200, function(){
// 展开后将按钮保持在容器底部
const $btnWrap = $("#device-info-toggle-btn").closest('.info-collapse-btn-wrapper');
if ($btnWrap.length) {
$btnWrap.appendTo('#device-info');
}
});
}
$(this).attr('aria-expanded', String(!isOpen));
});
await alertAndHide({ icon: "success", title: "获取设备信息成功", timer: 1200 });
} catch (e) {
await alertAndHide({ icon: "error", title: "获取设备信息出错", text: e.message, timer: 3000 });
} finally {
setBusy($btn, false);
}
}
function renderInfoList(list) {
const arr = Array.isArray(list) ? list : [];
const html = arr.filter(it => !shouldHideBaseLabel(it.label)).map(it => {
const label = String(it.label || "");
let value = String(it.value ?? "");
if (label === "产品型号") value = fmtModel(value);
return `<div><span class="info-k">${label}：</span> ${value}</div>`;
}).join("")
|| '<div class="text-muted">暂无可显示的设备信息</div>';
$("#device-info").html(html);
}
async function renderBatteryAndDisk(part = "both") {
const udid = getUDID();
if (!udid) return;
try {
const needBattery = part === "both" || part === "battery";
const needDisk = part === "both" || part === "disk";
const tasks = [];
if (needBattery) tasks.push(fetch(`${API.batteryDetail}?udid=${encodeURIComponent(udid)}`)); else tasks.push(Promise.resolve(null));
if (needDisk) tasks.push(fetch(`${API.diskDetail}?udid=${encodeURIComponent(udid)}`)); else tasks.push(Promise.resolve(null));
const [bRes, dRes] = await Promise.all(tasks);
const bData = bRes ? await bRes.json() : null;
const dData = dRes ? await dRes.json() : null;
const root = $("#device-info");
let html = root.html() || "";
// 去除旧块（简单基于标题关键字）
if (needBattery) html = html.replace(/<div class="info-section-title mt-2">电池信息：</div>[sS]*?(?=(<div class="info-section-title mt-2">|$))/g, "");
if (needDisk) html = html.replace(/<div class="info-section-title mt-2">磁盘信息：</div>[sS]*?(?=(<div class="info-section-title mt-2">|$))/g, "");
const parts = [];
if (needBattery && bData && bData.ok) {
const ck = bData.detail && bData.detail.batterycheck || {};
const rg = bData.detail && bData.detail.batteryregistry || {};
if (ck || rg) {
parts.push("<div class="info-section-title mt-2">电池信息：</div>");
if (ck) {
parts.push(`<div><span class="info-k">电池容量：</span> ${ck.BatteryCurrentCapacity ?? "?"}%</div>`);
}
if (rg) {
const temp = rg.Temperature != null ? (rg.Temperature/100).toFixed(2) + "°C" : "?";
const volt = rg.Voltage != null ? (rg.Voltage/1000).toFixed(3) + " V" : "?";
parts.push(`<div><span class="info-k">电池温度：</span> ${temp}</div>`);
parts.push(`<div><span class="info-k">电池电压：</span> ${volt}</div>`);
parts.push(`<div><span class="info-k">设计容量：</span> ${rg.DesignCapacity ?? "?"} mAh</div>`);
parts.push(`<div><span class="info-k">标称容量：</span> ${rg.NominalChargeCapacity ?? "?"} mAh</div>`);
}
}
}
if (needDisk && dData && dData.ok && dData.raw) {
const lines = [];
let parsed = false;
try {
const disk = JSON.parse(dData.raw);
const bs = disk.BlockSize ? `${disk.BlockSize}KB` : undefined;
const free = disk.FreeSpace || disk.free || undefined;
const used = disk.UsedSpace || disk.used || undefined;
const total = disk.TotalSpace || disk.total || undefined;
if (bs) lines.push(`<div>存储块规格：${bs}</div>`);
if (free) lines.push(`<div>闲置空间：${free}</div>`);
if (used) lines.push(`<div>已使用空间：${used}</div>`);
if (total) lines.push(`<div>总体空间：${total}</div>`);
parsed = true;
} catch {}
if (!parsed) {
const text = String(dData.raw);
const mBlock = text.match(/BlockSize:s*([0-9.]+)/i);
const mFree = text.match(/FreeSpace:s*([^
]+)/i);
const mUsed = text.match(/UsedSpace:s*([^
]+)/i);
const mTotal = text.match(/TotalSpace:s*([^
]+)/i);
const bs = mBlock ? `${mBlock[1]}KB` : undefined;
const free = mFree ? mFree[1].trim() : undefined;
const used = mUsed ? mUsed[1].trim() : undefined;
const total = mTotal ? mTotal[1].trim() : undefined;
if (bs) lines.push(`<div>存储块规格：${bs}</div>`);
if (free) lines.push(`<div>闲置空间：${free}</div>`);
if (used) lines.push(`<div>已使用空间：${used}</div>`);
if (total) lines.push(`<div>总体空间：${total}</div>`);
}
if (lines.length) {
parts.push("<div class="info-section-title mt-2">磁盘信息：</div>");
parts.push(lines.join(""));
}
}
root.html(html + parts.join(""));
} catch {}
}
// =========================
// 获取三方应用 & 搜索 & 停止
// =========================
// 解析 /api/apps 的 raw 输出为统一数组
function parseAppsRaw(raw) {
const apps = [];
if (!raw) return apps;
// 优先 JSON
try {
const data = JSON.parse(raw);
const arr = Array.isArray(data) ? data
: (data.apps || data.Apps || data.applications || data) || [];
for (const obj of arr) {
if (!obj || typeof obj !== "object") continue;
const bundleId = obj.CFBundleIdentifier || obj.bundleID || obj.bundleId || obj.Bundle || obj.id;
const name = obj.CFBundleDisplayName || obj.CFBundleName || obj.BundleName || obj.name || obj.Name || "";
const version = obj.CFBundleShortVersionString || obj.ShortVersion || obj.version || obj.Version || "";
const appType = obj.ApplicationType || obj.appType || obj.Type || "";
// 过滤系统应用：优先看 ApplicationType，其次看包名前缀
const isSystem = /system/i.test(String(appType)) || (bundleId && bundleId.startsWith("com.apple."));
if (!bundleId || isSystem) continue;
apps.push({ bundleId, name, version });
}
} catch {
// 纯文本回退：尝试 "bundleId - Name - Version" 的格式
const lines = String(raw).split(/
?
/);
for (const line of lines) {
const m = line.match(/^s*([a-zA-Z0-9._-]+)s*(?:-s*([^-
]+))?(?:-s*([^
]+))?s*$/);
if (!m) continue;
const bundleId = (m[1] || "").trim();
if (!bundleId || bundleId.startsWith("com.apple.")) continue;
const name = (m[2] || "").trim();
const version = (m[3] || "").trim();
apps.push({ bundleId, name, version });
}
}
// 去重
const seen = new Set();
return apps.filter(a => {
if (!a.bundleId || seen.has(a.bundleId)) return false;
seen.add(a.bundleId);
return true;
});
}
async function getInstalledApps() {
const udid = getUDID();
if (!udid) return alertAndHide({ icon: "warning", title: "请选择设备", timer: 2000 });
const $btn = $("#get-installed-apps-button");
setBusy($btn, true);
try {
// 取完整 JSON（list=0），便于区分系统/三方
const url = `${API.apps}?udid=${encodeURIComponent(udid)}&list=0`;
const res = await fetch(url, { cache: "no-store" });
if (!res.ok) throw new Error("HTTP " + res.status);
const data = await res.json();
if (!data.ok) throw new Error(data.raw || "获取应用列表失败");
const apps = parseAppsRaw(data.raw);
_allThirdPartyApps = apps;
renderAppsList(_allThirdPartyApps);
if (apps.length === 0) {
await alertAndHide({ icon: "info", title: "三方应用为空", text: "未检索到三方应用。", timer: 2000 });
} else {
await alertAndHide({ icon: "success", title: "获取成功", text: `共 ${apps.length} 个三方应用`, timer: 1200 });
}
} catch (e) {
await alertAndHide({ icon: "error", title: "获取失败", text: e.message || "未知错误", timer: 3000 });
} finally {
setBusy($btn, false);
}
}
function renderAppsList(apps) {
const $sel = $("#app-select");
$sel.empty();
if (!apps || apps.length === 0) {
$sel.append('<option value="">（三方应用为空）</option>');
return;
}
const options = apps
.slice()
.sort((a, b) => (a.name || a.bundleId).localeCompare(b.name || b.bundleId))
.map(a => {
const label = (a.name ? a.name + " · " : "") + a.bundleId + (a.version ? ` · v${a.version}` : "");
return `<option value="${a.bundleId}">${label}</option>`;
})
.join("");
$sel.html(options);
}
// 输入框过滤
function filterAppsBySearch() {
const q = ($("#app-search-input").val() || "").trim().toLowerCase();
const filtered = !q
? _allThirdPartyApps
: _allThirdPartyApps.filter(a =>
(a.bundleId || "").toLowerCase().includes(q) ||
(a.name || "").toLowerCase().includes(q)
);
renderAppsList(filtered);
}
// Kill 停止应用
async function stopApp() {
const udid = getUDID();
if (!udid) {
return alertAndHide({ icon: "warning", title: "请选择设备", timer: 2000 });
}
const bundleId = $("#app-select").val();
if (!bundleId) {
return alertAndHide({ icon: "warning", title: "请选择要停止的应用", timer: 2000 });
}
const $btn = $("#stop-app-button");
setBusy($btn, true);
try {
const res = await fetch(API.kill, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ udid, bundle_id: bundleId })
});
if (!res.ok) throw new Error("HTTP " + res.status);
const data = await res.json();
if (!data.ok) {
const msg = data.msg || data.raw || "停止失败";
// 针对"未运行"提示单独处理
if (/未运行/.test(msg)) {
await alertAndHide({
icon: "warning",
title: "应用未运行",
text: msg,
timer: 2500
});
} else {
await alertAndHide({
icon: "error",
title: "停止失败",
text: msg,
timer: 3000
});
}
return; // 提前结束
}
// 正常 Kill 成功
await alertAndHide({
icon: "success",
title: "已停止",
text: bundleId,
timer: 1200
});
} catch (e) {
await alertAndHide({
icon: "error",
title: "停止失败",
text: e.message || "未知错误",
timer: 3000
});
} finally {
setBusy($btn, false);
}
}
// =========================
// IPA 安装（单文件、格式校验）
// =========================
function validateIpaInput() {
const input = document.getElementById("ipa-file");
if (!input) return { ok: false, msg: "未找到文件选择器" };
const files = input.files;
if (!files || files.length === 0) return { ok: false, msg: "请选择 .ipa 文件" };
if (files.length > 1) return { ok: false, msg: "仅支持单个 .ipa 文件" };
const file = files[0];
const name = (file.name || "").toLowerCase();
if (!name.endsWith(".ipa")) return { ok: false, msg: "仅支持 .ipa 格式" };
return { ok: true, file };
}
async function installIpa() {
const udid = getUDID();
if (!udid) return alertAndHide({ icon: "warning", title: "请选择设备", timer: 2000 });
const check = validateIpaInput();
if (!check.ok) return alertAndHide({ icon: "warning", title: "文件校验失败", text: check.msg, timer: 2500 });
const file = check.file;
const $btn = $("#install-ipa-button");
setBusy($btn, true);
try {
const fd = new FormData();
fd.append("udid", udid);
fd.append("file", file);
const res = await fetch(API.install, { method: "POST", body: fd });
if (!res.ok) {
let msg = "HTTP " + res.status;
try { const e = await res.json(); msg = e.msg || msg; } catch {}
throw new Error(msg);
}
const data = await res.json();
if (!data.ok) throw new Error(data.raw || "安装失败");
$("#install-result-tip").text(`安装任务已提交：${file.name}`);
await alertAndHide({ icon: "success", title: "安装成功", text: file.name, timer: 1500 });
// 清空选择
$("#ipa-file").val("");
} catch (e) {
await alertAndHide({ icon: "error", title: "安装失败", text: e.message || "未知错误", timer: 3000 });
} finally {
setBusy($btn, false);
}
}
// =========================
// 截图
// =========================
async function takeScreenshot() {
const $btn = $("#take-screenshot-button");
setBusy($btn, true);
const udid = getUDID();
if (!udid) {
await alertAndHide({ icon: "warning", title: "请选择设备", timer: 2000 });
return setBusy($btn, false);
}
showLoading("正在截屏...");
try {
// 后端将以附件形式返回，并带上规范化文件名
const res = await fetch(`${API.screenshot}?udid=${encodeURIComponent(udid)}`, { cache: "no-store" });
if (!res.ok) {
try { const err = await res.json(); throw new Error(err.msg || `HTTP ${res.status}`); } catch {
throw new Error(`HTTP ${res.status}`);
}
}
// 从响应头解析文件名
const cd = res.headers.get("Content-Disposition") || "";
let downloadName = "screenshot.png";
try {
const m = cd.match(/filename*=UTF-8''([^;]+)|filename="?([^";]+)"?/i);
if (m) {
downloadName = decodeURIComponent(m[1] || m[2] || downloadName);
}
} catch {}
const blob = await res.blob();
const url = URL.createObjectURL(blob);
$("#screenshot-preview").html(
`<div class="d-flex flex-column h-100">
<img src="${url}">
<div class="mt-2 text-center">
<a href="${url}" download="${downloadName}" class="btn btn-sm btn-outline-secondary">下载图片</a>
</div>
</div>`
);
await alertAndHide({ icon: "success", title: "截屏完成", timer: 1200 });
} catch (e) {
await alertAndHide({ icon: "error", title: "截屏失败", text: e.message, timer: 3000 });
} finally {
setBusy($btn, false);
}
}
// =========================
// 录屏（基于 screenshot --stream）
// =========================
async function startScreenStream() {
const udid = getUDID();
if (!udid) {
return alertAndHide({ icon: "warning", title: "请选择设备", timer: 2000 });
}
const $btn = $("#record-screen-button");
setBusy($btn, true);
showLoading("正在准备录屏...");
try {
const res = await fetch(API.streamStart, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ udid })
});
if (!res.ok) throw new Error("HTTP " + res.status);
const data = await res.json();
if (!data.ok) throw new Error(data.msg || "启动录屏失败");
_streamSession = { id: data.id, udid, url: data.url };
// 使用 <img> 播放 MJPEG 流
$("#screenshot-preview").html(
`<div class="d-flex flex-column h-100">
<img src="${data.url}">
<div class="mt-2 small text-muted text-center">录屏进行中（MJPEG 流）</div>
</div>`
);
// 添加图片加载错误处理
const streamImg = document.getElementById("screen-stream");
let retryCount = 0;
const maxRetries = 3;
streamImg.addEventListener("error", function() {
retryCount++;
logWarn(`MJPEG 流加载失败 (尝试 ${retryCount}/${maxRetries})`);
if (retryCount <= maxRetries) {
$("#stream-status").text(`连接中断，尝试重连 (${retryCount}/${maxRetries})...`);
// 延迟重试
setTimeout(() => {
streamImg.src = data.url + "?t=" + Date.now(); // 添加时间戳避免缓存
}, 2000 * retryCount); // 递增延迟
} else {
$("#stream-status").html(`<span class="text-danger">流连接失败，请重新开始录屏</span>`);
// 自动停止录屏
setTimeout(stopScreenStream, 1000);
}
});
streamImg.addEventListener("load", function() {
retryCount = 0; // 重置重试计数
$("#stream-status").text("录屏进行中（MJPEG 流）");
});
$btn.text("停止录屏");
await alertAndHide({ icon: "success", title: "录屏已开始", timer: 900 });
} catch (e) {
await alertAndHide({ icon: "error", title: "启动录屏失败", text: e.message || "未知错误", timer: 3000 });
} finally {
hideLoading();
setBusy($btn, false);
}
}
async function stopScreenStream() {
if (!_streamSession) {
logWarn("stopScreenStream: _streamSession 为空");
return;
}
const { id, udid } = _streamSession;
logDebug("stopScreenStream: _streamSession =", _streamSession);
logDebug("stopScreenStream: 发送数据 =", { udid, id });
const $btn = $("#record-screen-button");
setBusy($btn, true);
try {
const res = await fetch(API.streamStop, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ udid, id })
});
if (!res.ok) {
const errorText = await res.text();
console.error("停止录屏HTTP错误:", res.status, errorText);
throw new Error("HTTP " + res.status);
}
const data = await res.json();
if (!data.ok) throw new Error(data.msg || "停止录屏失败");
_streamSession = null;
const dl = data.download_url ? `<a href="${data.download_url}" class="btn btn-sm btn-success" download>下载录屏文件</a>` : "";
$("#screenshot-preview").html(
dl ? `<div class="d-flex flex-column h-100 justify-content-center align-items-center">${dl}</div>` : ""
);
$btn.text("设备录屏");
await alertAndHide({ icon: "success", title: "录屏已停止", timer: 900 });
} catch (e) {
console.error("stopScreenStream 错误:", e);
await alertAndHide({ icon: "error", title: "停止录屏失败", text: e.message || "未知错误", timer: 3000 });
} finally {
setBusy($btn, false);
}
}
async function toggleScreenRecord() {
if (_streamSession) return stopScreenStream();
return startScreenStream();
}
// 启动应用
async function launchApp() {
const udid = getUDID();
if (!udid) return alertAndHide({ icon: "warning", title: "请选择设备", timer: 2000 });
const bundleId = $("#app-select").val();
if (!bundleId) return alertAndHide({ icon: "warning", title: "请选择要启动的应用", timer: 2000 });
const $btn = $("#launch-app-button");
setBusy($btn, true);
try {
const res = await fetch(API.launch, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ udid, bundle_id: bundleId, wait: false })
});
if (!res.ok) throw new Error("HTTP " + res.status);
const data = await res.json();
if (!data.ok) throw new Error(data.msg || data.raw || "启动失败");
await alertAndHide({ icon: "success", title: "已启动", text: bundleId, timer: 1200 });
} catch (e) {
await alertAndHide({ icon: "error", title: "启动失败", text: e.message || "未知错误", timer: 3000 });
} finally {
setBusy($btn, false);
}
}
// 重启设备
async function rebootDevice() {
const udid = getUDID();
if (!udid) return alertAndHide({ icon: "warning", title: "请选择设备", timer: 2000 });
const $btn = $("#reboot-device-button");
setBusy($btn, true);
try {
const ok = await Swal.fire({
icon: "warning",
title: "确认重启设备？",
showCancelButton: true,
confirmButtonText: "确认",
cancelButtonText: "取消",
}).then(r => r.isConfirmed);
if (!ok) return;
const res = await fetch(API.reboot, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ udid })
});
if (!res.ok) throw new Error("HTTP " + res.status);
const data = await res.json();
if (!data.ok) throw new Error(data.msg || data.raw || "重启失败");
await alertAndHide({ icon: "success", title: "重启执行中", timer: 1200 });
} catch (e) {
await alertAndHide({ icon: "error", title: "重启失败", text: e.message || "未知错误", timer: 3000 });
} finally {
setBusy($btn, false);
}
}
async function rebootDeviceFromDM() {
const udid = getUDID();
if (!udid) return alertAndHide({ icon: "warning", title: "请选择设备", timer: 2000 });
const ok = await Swal.fire({ icon: "warning", title: "确认重启设备？", showCancelButton: true }).then(r => r.isConfirmed);
if (!ok) return;
const res = await fetch(API.reboot, { method: "POST", headers: {"Content-Type":"application/json"}, body: JSON.stringify({ udid }) });
const data = await res.json();
if (!data.ok) return alertAndHide({ icon: "error", title: "重启失败", text: data.msg || data.raw || "" });
return alertAndHide({ icon: "success", title: "重启执行中", timer: 1200 });
}
async function listCrashes() {
const udid = getUDID();
if (!udid) return alertAndHide({ icon: "warning", title: "请选择设备", timer: 2000 });
const pattern = (document.getElementById("crash-search-input").value || "").trim();
const url = `${API.crashLs}?udid=${encodeURIComponent(udid)}${pattern ? `&pattern=${encodeURIComponent(pattern)}` : ""}`;
try {
const res = await fetch(url, { cache: "no-store" });
const data = await res.json();
if (!data.ok) throw new Error(data.msg || data.raw || "获取失败");
const items = Array.isArray(data.items) ? data.items : [];
// 自定义下拉多选菜单
const $menu = $("#crash-multi-menu").empty();
$menu.append(`<div class="menu-toolbar d-flex justify-content-between align-items-center"><div class="form-check mb-0"><input class="form-check-input" type="checkbox"><label class="form-check-label" for="crash-multi-all">全选</label></div><div class="d-flex gap-3"><a href="#">清空</a><a href="#">✕</a></div></div>`);
items.forEach((it, idx) => {
if (typeof it !== 'string') return;
const id = `crash-cb-${idx}`;
const $row = $("<div>").addClass("form-check");
const $input = $("<input>").addClass("form-check-input crash-cb").attr({ type: "checkbox", id, value: it });
const $label = $("<label>").addClass("form-check-label").attr("for", id).text(it);
$row.append($input, $label);
$menu.append($row);
});
// 追加局部绑定，确保清空点击必达
$menu.off("click.clearLocal").on("click.clearLocal", "#crash-multi-clear", function(e){
e.preventDefault(); e.stopPropagation(); if (e.stopImmediatePropagation) e.stopImmediatePropagation();
$("#crash-multi-menu .crash-cb").prop("checked", false);
$("#crash-multi-all").prop("checked", false);
$("#crash-select").val([]);
updateCrashTriggerText();
$("#crash-multi-menu").hide();
});
const $trigger = $("#crash-multi-trigger");
if (items.length === 0) {
$menu.append(`<div class="text-muted">无结果</div>`);
$trigger.prop("disabled", true).text("无结果（请更换关键字）");
} else {
$trigger.prop("disabled", false).text("点击选择日志👇（支持多选）");
}
updateCrashTriggerText();
if (items.length === 0) {
await alertAndHide({ icon: "info", title: "无匹配结果", text: "请更换关键字或清空后重试", timer: 2000 });
}
} catch (e) {
await alertAndHide({ icon: "error", title: "获取失败", text: e.message || "" });
}
}
function _getSelected($sel) {
// 自定义下拉多选所选
if ($("#profile-multi").length) {
const arrP = [];
$("#profile-multi-menu .profile-cb:checked").each(function(){ arrP.push(this.value); });
if (arrP.length) return arrP;
}
if ($("#crash-multi").length) {
const arr = [];
$("#crash-multi-menu .crash-cb:checked").each(function(){ arr.push(this.value); });
if (arr.length) return arr;
}
// 兜底：下拉框值
const v = $sel.val();
if (v === null || v === undefined || v === "") return [];
if (Array.isArray(v)) return v;
return [v];
}
async function exportCrashes() {
const udid = getUDID();
if (!udid) return alertAndHide({ icon: "warning", title: "请选择设备", timer: 2000 });
const hasList = $("#crash-multi-menu .crash-cb").length > 0;
if (!hasList) return alertAndHide({ icon: "info", title: "请先获取Crash信息列表", timer: 1800 });
const sels = _getSelected($("#crash-select"));
if (sels.length === 0) {
const okAll = await Swal.fire({ icon: "question", title: "未选择，导出全部？", showCancelButton: true }).then(r => r.isConfirmed);
if (!okAll) return;
}
const body = { udid };
if (sels.length > 1) body.patterns = sels; else if (sels.length === 1) body.pattern = sels[0]; else body.pattern = "*";
const res = await fetch(API.crashCp, { method: "POST", headers: {"Content-Type":"application/json"}, body: JSON.stringify(body) });
const data = await res.json();
if (!data.ok) return alertAndHide({ icon: "error", title: "导出失败", text: data.raw || "" });
// 自动下载：单/多（2）/打包（3+）
if (data.download_url) {
const a = document.createElement("a");
a.href = data.download_url;
a.download = "";
document.body.appendChild(a); a.click(); a.remove();
await alertAndHide({ icon: "success", title: "已开始下载", timer: 1000 });
} else {
await alertAndHide({ icon: "info", title: "无可下载文件", timer: 1200 });
}
}
async function deleteCrashes() {
const udid = getUDID();
if (!udid) return alertAndHide({ icon: "warning", title: "请选择设备", timer: 2000 });
const hasList = $("#crash-multi-menu .crash-cb").length > 0;
if (!hasList) return alertAndHide({ icon: "info", title: "请先获取Crash信息列表", timer: 1800 });
const sels = _getSelected($("#crash-select"));
if (sels.length === 0) {
const okAll = await Swal.fire({ icon: "warning", title: "未选择，删除全部？", text: "将删除所有匹配(*)", showCancelButton: true }).then(r => r.isConfirmed);
if (!okAll) return;
}
const ok = await Swal.fire({ icon: "warning", title: "确认删除所选？", showCancelButton: true }).then(r => r.isConfirmed);
if (!ok) return;
const body = { udid };
if (sels.length > 1) body.patterns = sels; else if (sels.length === 1) body.pattern = sels[0]; else body.pattern = "*";
const res = await fetch(API.crashRm, { method: "POST", headers: {"Content-Type":"application/json"}, body: JSON.stringify(body) });
const data = await res.json();
if (!data.ok) return alertAndHide({ icon: "error", title: "删除失败", text: data.raw || "" });
await alertAndHide({ icon: "success", title: "已删除", timer: 1200 });
listCrashes();
}
async function profileList() {
await withUdid(null, async (udid) => {
const data = await apiFetchJSON(`${API.profileList}?udid=${encodeURIComponent(udid)}`);
if (!data.ok) return notifyError("获取失败", data.raw || "");
const items = Array.isArray(data.items) ? data.items : [];
// 归一化：支持字符串或对象（优先显示 Metadata.PayloadDisplayName / Manifest.Description）
const norm = items.map((it) => {
if (typeof it === "string") return { value: it, label: it };
try {
const name = (it && it.Metadata && (it.Metadata.PayloadDisplayName || it.Metadata.PayloadName))
|| (it && it.Manifest && it.Manifest.Description)
|| (it && it.Identifier)
|| String(it);
// 仅显示名称；不再拼接短ID
return { value: String(name), label: String(name) };
} catch {
const s = String(it);
return { value: s, label: s };
}
});
// 隐藏下拉兜底（用于 _getSelected 兼容）
const $sel = $("#profile-select").empty();
norm.forEach(it => $sel.append(`<option value="${it.value}">${it.label}</option>`));
// 渲染自定义下拉多选
const $menu = $("#profile-multi-menu").empty();
$menu.append(`<div class="menu-toolbar d-flex justify-content-between align-items-center"><div class="form-check mb-0"><input class="form-check-input" type="checkbox"><label class="form-check-label" for="profile-multi-all">全选</label></div><div class="d-flex gap-3"><a href="#">清空</a><a href="#">✕</a></div></div>`);
norm.forEach((it, idx) => {
const id = `profile-cb-${idx}`;
const $row = $("<div>").addClass("form-check");
const $input = $("<input>").addClass("form-check-input profile-cb").attr({ type: "checkbox", id, value: it.value });
const $label = $("<label>").addClass("form-check-label").attr("for", id).text(it.label);
$row.append($input, $label);
$menu.append($row);
});
const $trigger = $("#profile-multi-trigger");
if (norm.length === 0) {
$menu.append(`<div class="text-muted">无结果</div>`);
$trigger.prop("disabled", true).text("无结果（请先获取）");
} else {
$trigger.prop("disabled", false).text("点击选择配置👇（支持多选）");
}
updateProfileTriggerText();
});
}
async function profileRemove() {
const udid = getUDID();
if (!udid) return alertAndHide({ icon: "warning", title: "请选择设备", timer: 2000 });
const sels = _getSelected($("#profile-select"));
if (sels.length === 0) return alertAndHide({ icon: "warning", title: "请选择要删除的配置文件", timer: 2000 });
const ok = await Swal.fire({ icon: "warning", title: "确认移除所选配置文件？", showCancelButton: true }).then(r => r.isConfirmed);
if (!ok) return;
const res = await fetch(API.profileRemove, { method: "POST", headers: {"Content-Type":"application/json"}, body: JSON.stringify({ udid, names: sels }) });
const data = await res.json();
if (!data.ok) return alertAndHide({ icon: "error", title: "移除失败", text: (data.results||[]).map(x=>`${x.name}:${x.ok?'ok':'fail'}`).join("
") });
await alertAndHide({ icon: "success", title: "已移除", timer: 1200 });
profileList();
}
async function assistiveAction(action) {
const udid = getUDID();
if (!udid) return alertAndHide({ icon: "warning", title: "请选择设备", timer: 2000 });
const feature = $("#assistive-feature-select").val();
const res = await fetch(API.assistive(feature, action), { method: "POST", headers: {"Content-Type":"application/json"}, body: JSON.stringify({ udid }) });
const data = await res.json();
if (!data.ok) return alertAndHide({ icon: "error", title: "执行失败", text: data.raw || "" });
await alertAndHide({ icon: "success", title: "已执行", timer: 900 });
}
async function appsRunning() {
const udid = getUDID();
if (!udid) return alertAndHide({ icon: "warning", title: "请选择设备", timer: 2000 });
const res = await fetch(`${API.appsRunning}?udid=${encodeURIComponent(udid)}`);
const data = await res.json();
if (!data.ok) return $("#apps-running-list").text("获取失败");
const list = Array.isArray(data.list) ? data.list : [];
const html = list.length ? list.map(p => `<div>${p.Name || "(unknown)"} · PID ${p.Pid || "?"}</div>`).join("") : '<div class="text-muted">无运行中应用</div>';
$("#apps-running-list").html(html);
}
let _syslogSession = null;
let _syslogES = null;
let _syslogBuf = [];
let _syslogFlushScheduled = false;
const _syslogMaxLines = 1000; // 最大保留行数
const _syslogBatchSize = 300; // 每帧最多渲染条数
let _syslogGen = 0; // 清空/重连时自增，丢弃旧批次
function _escapeHtml(s){
return String(s).replace(/[&<>"']/g, function(ch){
return ({'&':'&','<':'<','>':'>','"':'"',''':'''}[ch]);
});
}
function _buildHighlighter(){
const kwRaw = (document.getElementById('syslog-kw')?.value || '').trim();
if (!kwRaw) return null;
const parts = kwRaw.split(',').map(s=>s.trim()).filter(Boolean).map(s=>s.replace(/[.*+?^${}()|[]\]/g, '\$&'));
if (!parts.length) return null;
const re = new RegExp('(' + parts.join('|') + ')', 'gi');
return function(text){
const safe = _escapeHtml(text);
return safe.replace(re, '<span class="hl">$1</span>');
};
}
function _syslogScheduleFlush() {
if (_syslogFlushScheduled) return;
_syslogFlushScheduled = true;
requestAnimationFrame(_syslogFlush);
}
function _syslogFlush() {
_syslogFlushScheduled = false;
const curGen = _syslogGen; // 捕获当前代次
const $out = $("#syslog-live");
if (!$out.length) { _syslogBuf.length = 0; return; }
const el = $out[0];
const nearBottom = (el.scrollHeight - el.scrollTop - el.clientHeight) < 40;
const frag = document.createDocumentFragment();
const highlighter = _buildHighlighter();
let n = Math.min(_syslogBuf.length, _syslogBatchSize);
for (let i = 0; i < n; i++) {
if (curGen !== _syslogGen) return; // 清空/重连后放弃本批
const line = _syslogBuf[i];
const div = document.createElement('div');
if (highlighter) {
div.innerHTML = highlighter(line);
} else {
div.textContent = line;
}
frag.appendChild(div);
}
_syslogBuf.splice(0, n);
if (curGen !== _syslogGen) return; // 再次检查
el.appendChild(frag);
// 仅当本次确实写入了日志时才隐藏占位文字
if (n > 0) { $("#syslog-placeholder").hide(); }
// 限制最大行数，超出从顶部移除
while (el.childNodes.length > _syslogMaxLines) {
el.removeChild(el.firstChild);
}
if (nearBottom) {
el.scrollTop = el.scrollHeight;
}
if (_syslogBuf.length > 0) {
_syslogScheduleFlush();
}
}
async function syslogStart() {
const udid = getUDID();
if (!udid) return alertAndHide({ icon: "warning", title: "请选择设备", timer: 2000 });
const res = await fetch(API.syslogStart, { method: "POST", headers: {"Content-Type":"application/json"}, body: JSON.stringify({ udid, parse: false }) });
const data = await res.json();
if (!data.ok) return alertAndHide({ icon: "error", title: "启动失败", text: data.msg || data.raw || "" });
_syslogSession = { id: data.id, udid };
$("#syslog-download-link").text(`日志文件：${data.file}`);
// 打开 SSE 流
try {
if (_syslogES) { _syslogES.close(); _syslogES = null; }
const url = _buildSyslogStreamUrl(udid);
_syslogES = new EventSource(url);
const $out = $("#syslog-live");
if ($out.length) {
// 开始前仅清除历史内容，保留占位符，待收到首条日志再隐藏
$out.find('div:not(#syslog-placeholder)').remove();
}
_syslogBuf.length = 0; _syslogGen++;
if (_syslogES) {
_syslogES.onmessage = function(ev){
const g = _syslogGen; // 捕获代次，避免旧事件在清空后生效
_syslogBuf.push(ev.data || "");
if (g === _syslogGen) _syslogScheduleFlush();
};
_syslogES.onerror = function(){ /* 静默 */ };
}
} catch(_) {}
await alertAndHide({ icon: "success", title: "系统日志已开始", timer: 900 });
}
async function syslogStop() {
if (!_syslogSession) return alertAndHide({ icon: "warning", title: "未开始系统日志", timer: 1500 });
const { id, udid } = _syslogSession;
const res = await fetch(API.syslogStop, { method: "POST", headers: {"Content-Type":"application/json"}, body: JSON.stringify({ udid, id }) });
const data = await res.json();
if (!data.ok) return alertAndHide({ icon: "error", title: "停止失败", text: data.msg || data.raw || "" });
_syslogSession = null;
if (_syslogES) { try { _syslogES.close(); } catch(_) {} _syslogES = null; }
_syslogBuf.length = 0; _syslogGen++;
// 停止后显示占位文字
const $out = $('#syslog-live');
if ($out.length) {
// 只清空日志内容，保留占位符
$out.find('div:not(#syslog-placeholder)').remove();
$("#syslog-placeholder").show();
}
if (data.download_url) $("#syslog-download-link").html(`<a href="${data.download_url}" class="link-primary">下载日志</a>`);
await alertAndHide({ icon: "success", title: "系统日志已停止", timer: 900 });
}
function _buildSyslogStreamUrl(udid){
const kw = (document.getElementById('syslog-kw')?.value || '').trim();
const lv = (document.getElementById('syslog-lv')?.value || '').trim();
const params = new URLSearchParams();
params.set('udid', udid);
params.set('parse', '0');
if (kw) params.set('kw', kw);
if (lv) params.set('lv', lv);
return `/api/syslog/stream?${params.toString()}`;
}
function bindDeviceEventsSSE() {
try {
const es = new EventSource(API.deviceEvents);
es.onmessage = function(ev){
// 简单策略：接到任何事件都刷新设备列表
refreshDevices();
};
es.onerror = function(){ /* 静默 */ };
} catch(e) {
// 忽略 SSE 不可用
}
}
function bindDMEvents() {
$("#dm-reboot-button").off("click").on("click", function(e){ e.preventDefault(); rebootDeviceFromDM(); });
$("#crash-list-button").off("click").on("click", function(e){ e.preventDefault(); listCrashes(); });
$("#crash-export-button").off("click").on("click", function(e){ e.preventDefault(); exportCrashes(); });
$("#crash-delete-button").off("click").on("click", function(e){
e.preventDefault();
deleteCrashes();
});
$("#crash-multi-trigger").off("click").on("click", function(e){
e.preventDefault(); e.stopPropagation();
const $m = $("#crash-multi-menu");
$m.toggle();
return false;
});
$(document).off("click.crashmenu").on("click.crashmenu", function(){
$("#crash-multi-menu").hide();
});
$("#crash-multi-menu").off("click").on("click", function(e){ e.stopPropagation(); });
$(document).off("change.crashmenu").on("change.crashmenu", "#crash-multi-all", function(){
const on = this.checked;
$("#crash-multi-menu .crash-cb").prop("checked", on);
updateCrashTriggerText();
});
$(document).off("click.crashclear").on("click.crashclear", "#crash-multi-clear", function(e){
e.preventDefault(); e.stopPropagation(); if (e.stopImmediatePropagation) e.stopImmediatePropagation();
$("#crash-multi-menu .crash-cb").prop("checked", false);
$("#crash-multi-all").prop("checked", false);
$("#crash-select").val([]);
updateCrashTriggerText();
});
$(document).off("change.crashcb").on("change.crashcb", "#crash-multi-menu .crash-cb", function(){
const total = $("#crash-multi-menu .crash-cb").length;
const checked = $("#crash-multi-menu .crash-cb:checked").length;
$("#crash-multi-all").prop("checked", total > 0 && checked === total);
updateCrashTriggerText();
});
$("#crash-multi-menu").off("click.closeLocal").on("click.closeLocal", "#crash-multi-close", function(e){
e.preventDefault(); e.stopPropagation(); if (e.stopImmediatePropagation) e.stopImmediatePropagation();
$("#crash-multi-menu").hide();
return false;
});
$("#crash-checkbox-mode-toggle").off("change").on("change", function(){
const on = this.checked;
$("#crash-select").prop("disabled", on);
$("#crash-checkbox-select-all").prop("disabled", !on);
if (on) {
// 复选模式默认不全选
$("#crash-checkbox-select-all").prop("checked", false);
$("#crash-checkbox-list .crash-cb").prop("checked", false);
}
});
$("#crash-checkbox-select-all").off("change").on("change", function(){
const on = this.checked;
if (!$("#crash-checkbox-mode-toggle").prop("checked")) return;
$("#crash-checkbox-list .crash-cb").prop("checked", on);
});
$("#profile-list-button").off("click").on("click", function(e){ e.preventDefault(); profileList(); });
$("#profile-remove-button").off("click").on("click", function(e){ e.preventDefault(); profileRemove(); });
$("#assistive-enable-button").off("click").on("click", function(e){ e.preventDefault(); assistiveAction("enable"); });
$("#assistive-disable-button").off("click").on("click", function(e){ e.preventDefault(); assistiveAction("disable"); });
$("#apps-running-button").off("click").on("click", function(e){ e.preventDefault(); appsRunning(); });
$("#syslog-start-button").off("click").on("click", function(e){ e.preventDefault(); if (!ensureDeviceSelected()) return; syslogStart(); });
$("#syslog-stop-button").off("click").on("click", function(e){ e.preventDefault(); if (!ensureDeviceSelected() || !ensureSyslogStarted()) return; syslogStop(); });
// Profile 多选触发
$("#profile-multi-trigger").off("click").on("click", function(e){
e.preventDefault(); e.stopPropagation();
const $m = $("#profile-multi-menu");
$m.toggle();
return false;
});
// 点击空白隐藏
bindDocGlobal("click", "profilemenu", function(){ $("#profile-multi-menu").hide(); });
// 菜单内部不冒泡
$("#profile-multi-menu").off("click").on("click", function(e){ e.stopPropagation(); });
// 全选
bindDocNS("change", "profilemenu", "#profile-multi-all", function(){
const on = this.checked;
$("#profile-multi-menu .profile-cb").prop("checked", on);
updateProfileTriggerText();
});
// 清空（使用本地委托，避免被容器 stopPropagation 影响）
$("#profile-multi-menu").off("click.clearLocal").on("click.clearLocal", "#profile-multi-clear", function(e){
e.preventDefault(); e.stopPropagation(); if (e.stopImmediatePropagation) e.stopImmediatePropagation();
$("#profile-multi-menu .profile-cb").prop("checked", false);
$("#profile-multi-all").prop("checked", false);
$("#profile-select").val([]);
updateProfileTriggerText();
$("#profile-multi-menu").hide();
return false;
});
// 单个复选
bindDocNS("change", "profilecb", "#profile-multi-menu .profile-cb", function(){
const total = $("#profile-multi-menu .profile-cb").length;
const checked = $("#profile-multi-menu .profile-cb:checked").length;
$("#profile-multi-all").prop("checked", total > 0 && checked === total);
updateProfileTriggerText();
});
// 关闭按钮
$("#profile-multi-menu").off("click.closeLocal").on("click.closeLocal", "#profile-multi-close", function(e){
e.preventDefault(); e.stopPropagation(); if (e.stopImmediatePropagation) e.stopImmediatePropagation();
$("#profile-multi-menu").hide();
return false;
});
}
// =========================
// 事件绑定（仅 JS；HTML 不写 onclick）
// =========================
function bindEvents() {
$("#device-select").off("change").on("change", function() {
syncHiddenUdid();
const selectedUdid = getUDID();
// 设备选择变化时的处理
if (!selectedUdid) {
// 没有选择设备
showDevModeStatus("no-device");
} else {
// 选择了设备，延迟检测开发者模式
setTimeout(backgroundDevModeCheck, 500);
}
});
$("#refresh-devices-button").off("click").on("click", function (e) {
e.preventDefault(); e.stopImmediatePropagation(); refreshDevices();
});
$("#get-device-info-button").off("click").on("click", function (e) {
e.preventDefault(); e.stopImmediatePropagation(); getDeviceInfo();
});
$("#take-screenshot-button").off("click").on("click", function (e) {
e.preventDefault(); e.stopImmediatePropagation(); takeScreenshot();
});
$("#record-screen-button").off("click").on("click", function (e) {
e.preventDefault(); e.stopImmediatePropagation(); toggleScreenRecord();
});
// 新增：应用管理
$("#get-installed-apps-button").off("click").on("click", function (e) {
e.preventDefault(); e.stopImmediatePropagation(); getInstalledApps();
});
$("#app-search-input").off("input").on("input", filterAppsBySearch);
$("#stop-app-button").off("click").on("click", function (e) {
e.preventDefault(); e.stopImmediatePropagation(); stopApp();
});
$("#install-ipa-button").off("click").on("click", function (e) {
e.preventDefault(); e.stopImmediatePropagation(); installIpa();
});
$("#launch-app-button").off("click").on("click", function (e) {
e.preventDefault(); e.stopImmediatePropagation(); launchApp();
});
$("#reboot-device-button").off("click").on("click", function (e) {
e.preventDefault(); e.stopImmediatePropagation(); rebootDevice();
});
}
// —— 初始化 ——
$(function () {
$("#loading-overlay").hide();
bindEvents();
bindDMEvents();
bindDeviceEventsSSE();
// 初始化时隐藏开发者模式状态提示
showDevModeStatus(null);
refreshDevices();
});
// 暴露少量接口（调试可用）
const IOSApp = {
refreshDevices, getDeviceInfo, takeScreenshot,
getInstalledApps, stopApp, installIpa
};
global.IOSApp = IOSApp;
function updateCrashTriggerText() {
const sels = [];
$("#crash-multi-menu .crash-cb:checked").each(function(){ sels.push(this.value); });
const txt = sels.length ? `${sels.length} 项已选` : "点击选择日志👇（支持多选）";
$("#crash-multi-trigger").text(txt);
}
function updateProfileTriggerText() {
const sels = [];
$("#profile-multi-menu .profile-cb:checked").each(function(){ sels.push(this.value); });
const txt = sels.length ? `${sels.length} 项已选` : "点击选择配置👇（支持多选）";
$("#profile-multi-trigger").text(txt);
}
// 绑定筛选与清空
$(document).off('click.syslogFilter').on('click.syslogFilter', '#syslog-apply-filter', function(e){
e.preventDefault();
// 1. 检查设备是否连接
if (!ensureDeviceSelected()) return;
// 2. 检查关键字输入是否为空
const kw = (document.getElementById('syslog-kw')?.value || '').trim();
const lv = (document.getElementById('syslog-lv')?.value || '').trim();
if (!kw && !lv) {
return alertAndHide({ icon: 'warning', title: '请输入筛选条件', text: '请在关键字输入框中输入筛选条件或选择日志级别', timer: 2000 });
}
// 3. 检查是否已开始系统日志（可选，不影响筛选功能）
if (!ensureSyslogStarted()) return;
// 应用筛选：清空显示内容，重新应用筛选条件到现有日志
_syslogGen++;
syslogClearContentOnly();
// 重新处理现有缓冲区中的日志，应用新的筛选条件
// 如果有筛选条件，重新处理缓冲区
if (kw || lv) {
const filteredLines = [];
for (const line of _syslogBuf) {
let include = true;
// 关键字筛选
if (kw) {
const keywords = kw.split(',').map(k => k.trim().toLowerCase());
const lineLower = line.toLowerCase();
include = keywords.some(keyword => lineLower.includes(keyword));
}
// 级别筛选 - 适配原始 syslog 格式
if (include && lv) {
const level = lv.toLowerCase();
const lineLower = line.toLowerCase();
if (level === 'error') {
include = lineLower.includes('error') || lineLower.includes('fatal') || lineLower.includes('critical') || lineLower.includes('<error>');
} else if (level === 'warning') {
include = lineLower.includes('warning') || lineLower.includes('warn') || lineLower.includes('<warning>');
} else if (level === 'info') {
include = lineLower.includes('info') || lineLower.includes('notice') || lineLower.includes('<notice>') || lineLower.includes('<info>');
}
}
if (include) {
filteredLines.push(line);
}
}
// 显示筛选后的日志
const $out = $("#syslog-live");
if ($out.length) {
const el = $out[0];
for (const line of filteredLines) {
const div = document.createElement('div');
div.textContent = line;
el.appendChild(div);
}
el.scrollTop = el.scrollHeight;
}
} else {
// 无筛选条件，显示所有日志
const $out = $("#syslog-live");
if ($out.length) {
const el = $out[0];
for (const line of _syslogBuf) {
const div = document.createElement('div');
div.textContent = line;
el.appendChild(div);
}
el.scrollTop = el.scrollHeight;
}
}
});
$(document).off('click.syslogClear').on('click.syslogClear', '#syslog-clear-live', function(e){
e.preventDefault();
if (!ensureDeviceSelected() || !ensureSyslogStarted()) return;
_syslogBuf.length = 0; _syslogGen++;
syslogClearContentOnly();
});
// =========================
// 后台自动检测开发者模式
// =========================
// 检测设备连接状态
async function checkDeviceConnection(udid) {
try {
const res = await fetch(`${API.getDeviceInfo}?udid=${encodeURIComponent(udid)}`, { cache: "no-store" });
return res.ok;
} catch (e) {
return false;
}
}
// 检测开发者模式状态
async function checkDevModeStatus(udid) {
try {
const res = await fetch(`${API.devmodeCheck}?udid=${encodeURIComponent(udid)}`, { cache: "no-store" });
if (!res.ok) {
return null;
}
const data = await res.json();
if (!data.ok) {
console.log("checkDevModeStatus: API返回失败", data.raw);
return null;
}
const raw = String(data.raw || "").toLowerCase().trim();
// 更精确的匹配逻辑
if (raw.includes("enabled: true") || raw.includes("enabled: 1") || raw.includes("enabled: yes")) {
return "enabled";
} else if (raw.includes("enabled: false") || raw.includes("enabled: 0") || raw.includes("enabled: no")) {
return "disabled";
} else if (raw.includes("disabled: true") || raw.includes("disabled: 1")) {
return "disabled";
} else if (raw.includes("disabled: false") || raw.includes("disabled: 0")) {
return "enabled";
}
// 兜底匹配
if (raw.includes("true") && !raw.includes("false")) {
return "enabled";
} else if (raw.includes("false") && !raw.includes("true")) {
return "disabled";
}
return null;
} catch (e) {
console.error("checkDevModeStatus: 检测异常", e);
return null;
}
}
// 显示开发者模式状态提示
function showDevModeStatus(status, message) {
const $tip = $("#devmode-status-tip");
if (status === "enabled") {
$tip.html('<span class="text-success">✓ 开发者模式已启用</span>').show();
} else if (status === "disabled") {
$tip.html('<span class="text-warning">⚠️ 开发者模式未启用</span>').show();
} else if (status === "error") {
$tip.html(`<span class="text-danger">❌ 检测失败: ${message}</span>`).show();
} else if (status === "checking") {
$tip.html(`<span class="text-info">🔄 ${message}</span>`).show();
} else if (status === "no-device") {
$tip.html('<span class="text-muted">请连接设备</span>').show();
} else {
$tip.hide();
}
}
// 检测所有设备的开发者模式状态
async function checkAllDevicesDevMode() {
const $select = $("#device-select");
const selectedUdid = getUDID();
// 如果没有选择设备，隐藏状态提示
if (!selectedUdid) {
showDevModeStatus("no-device");
return;
}
// 检查设备连接状态
const isConnected = await checkDeviceConnection(selectedUdid);
if (!isConnected) {
showDevModeStatus("error", "设备连接异常");
return;
}
showDevModeStatus("checking", "正在检测开发者模式...");
try {
const devModeStatus = await checkDevModeStatus(selectedUdid);
if (devModeStatus === "disabled") {
showDevModeStatus("disabled");
// 开发者模式未启用，提醒用户手动开启
await alertAndHide({
icon: "warning",
title: "开发者模式未启用",
text: "当前设备未启用开发者模式，某些调试功能可能无法正常使用。请在设备上手动开启开发者模式。",
timer: 4000
});
} else if (devModeStatus === "enabled") {
showDevModeStatus("enabled");
} else {
showDevModeStatus("error", "无法检测开发者模式状态");
}
} catch (e) {
console.error("checkAllDevicesDevMode: 检测异常", e);
showDevModeStatus("error", e.message);
}
}
// 后台自动检测流程（重命名保持兼容）
async function backgroundDevModeCheck() {
await checkAllDevicesDevMode();
}
// 公共：系统日志占位与内容管理
function syslogShowPlaceholder() {
const $box = $("#syslog-live");
if ($box.length) { $("#syslog-placeholder").show(); }
}
function syslogHidePlaceholder() {
const $box = $("#syslog-live");
if ($box.length) { $("#syslog-placeholder").hide(); }
}
function syslogClearContentOnly() {
const $box = $("#syslog-live");
if ($box.length) { $box.find('div:not(#syslog-placeholder)').remove(); $box[0].scrollTop = 0; }
}
// 公共：操作前置校验
function ensureDeviceSelected() {
const udid = getUDID();
if (!udid) { alertAndHide({ icon: 'warning', title: '请选择设备', timer: 1500 }); return false; }
return true;
}
function ensureSyslogStarted() {
if (!_syslogSession) { alertAndHide({ icon: 'warning', title: '请先开始系统日志', timer: 1500 }); return false; }
return true;
}
// =========================
// Debug功能：三连击触发日志导出
// =========================
// Debug点击计数器
let debugClickCount = 0;
let debugClickTimer = null;
const DEBUG_CLICK_TIMEOUT = 2000; // 2秒内需要完成三连击
const DEBUG_CLICK_REQUIRED = 3; // 需要3次点击
// 初始化Debug触发器
function initDebugTrigger() {
const $trigger = $("#debug-trigger");
if (!$trigger.length) return;
$trigger.on("click", function(e) {
e.preventDefault();
debugClickCount++;
// 清除之前的定时器
if (debugClickTimer) {
clearTimeout(debugClickTimer);
}
// 设置新的定时器
debugClickTimer = setTimeout(() => {
// 超时重置计数器
debugClickCount = 0;
debugClickTimer = null;
}, DEBUG_CLICK_TIMEOUT);
// 检查是否达到三连击
if (debugClickCount >= DEBUG_CLICK_REQUIRED) {
// 重置计数器
debugClickCount = 0;
if (debugClickTimer) {
clearTimeout(debugClickTimer);
debugClickTimer = null;
}
// 触发debug功能
triggerDebugMenu();
}
});
}
// 触发Debug功能选择
async function triggerDebugMenu() {
try {
const result = await Swal.fire({
title: 'Debug功能',
text: '请选择要执行的调试操作',
icon: 'question',
showCancelButton: true,
confirmButtonText: '导出日志',
cancelButtonText: '取消',
showDenyButton: true,
denyButtonText: '清理进程',
reverseButtons: true
});
if (result.isConfirmed) {
await triggerDebugExport();
} else if (result.isDenied) {
await triggerDebugCleanup();
}
} catch (error) {
console.error('Debug菜单异常:', error);
}
}
// 触发Debug日志导出
async function triggerDebugExport() {
try {
// 显示确认对话框
const result = await Swal.fire({
title: '导出日志',
text: '即将导出应用日志文件用于问题分析，是否继续？',
icon: 'question',
showCancelButton: true,
confirmButtonText: '导出日志',
cancelButtonText: '取消',
reverseButtons: true
});
if (!result.isConfirmed) {
return;
}
// 显示加载状态
Swal.fire({
title: '正在导出日志...',
text: '请稍候，正在读取和复制日志文件',
allowOutsideClick: false,
didOpen: () => {
Swal.showLoading();
}
});
// 调用API导出日志
const response = await fetch(API.debugExportLogs, {
method: 'POST',
headers: {
'Content-Type': 'application/json'
}
});
const data = await response.json();
if (data.ok) {
// 导出成功，显示下载链接
await Swal.fire({
title: '日志导出成功',
html: `
<p>日志文件已成功导出：</p>
<p><strong>${data.filename}</strong></p>
<p class="text-muted small">文件已保存到下载目录，1小时内有效</p>
`,
icon: 'success',
confirmButtonText: '下载文件',
showCancelButton: true,
cancelButtonText: '关闭'
}).then((result) => {
if (result.isConfirmed && data.download_url) {
// 触发下载
const link = document.createElement('a');
link.href = data.download_url;
link.download = data.filename;
document.body.appendChild(link);
link.click();
document.body.removeChild(link);
}
});
} else {
// 导出失败
await Swal.fire({
title: '导出失败',
text: data.msg || '未知错误',
icon: 'error',
confirmButtonText: '确定'
});
}
} catch (error) {
console.error('Debug导出异常:', error);
await Swal.fire({
title: '导出异常',
text: '网络错误或服务器异常，请稍后重试',
icon: 'error',
confirmButtonText: '确定'
});
}
}
// 触发Debug进程清理
async function triggerDebugCleanup() {
try {
// 显示确认对话框
const result = await Swal.fire({
title: '清理进程',
text: '即将清理所有遗留的ios.exe进程，是否继续？',
icon: 'warning',
showCancelButton: true,
confirmButtonText: '清理进程',
cancelButtonText: '取消',
reverseButtons: true
});
if (!result.isConfirmed) {
return;
}
// 显示加载状态
Swal.fire({
title: '正在清理进程...',
text: '请稍候，正在终止遗留的ios.exe进程',
allowOutsideClick: false,
didOpen: () => {
Swal.showLoading();
}
});
// 调用API清理进程
const response = await fetch(API.debugCleanupProcesses, {
method: 'POST',
headers: {
'Content-Type': 'application/json'
}
});
const data = await response.json();
if (data.ok) {
await Swal.fire({
title: '进程清理成功',
text: data.msg || '所有遗留的ios.exe进程已清理',
icon: 'success',
confirmButtonText: '确定'
});
} else {
await Swal.fire({
title: '清理失败',
text: data.msg || '未知错误',
icon: 'error',
confirmButtonText: '确定'
});
}
} catch (error) {
console.error('Debug清理异常:', error);
await Swal.fire({
title: '清理异常',
text: '网络错误或服务器异常，请稍后重试',
icon: 'error',
confirmButtonText: '确定'
});
}
}
// 初始化引导文档入口
function initGuideEntry() {
const $guideBtn = $("#guide-entry-btn");
if (!$guideBtn.length) return;
$guideBtn.on("click", function(e) {
e.preventDefault();
openGuideDocument();
});
}
// 打开引导文档
function openGuideDocument() {
// 在新窗口中打开引导文档
const guideUrl = "/guide";
window.open(guideUrl, "_blank", "width=1200,height=800,scrollbars=yes,resizable=yes");
}
// 页面加载完成后初始化Debug功能
$(document).ready(function() {
initDebugTrigger();
initGuideEntry();
});
})(window);5、CSS代码
/* 页面整体样式 */
body {
font-family: Arial, sans-serif; /* 全局字体 */
line-height: 1.6;
background-color: #f8f9fa; /* 背景色 */
padding: 20px;
background-image: url('/static/wresource/background.jpg'); /* 背景图片 */
background-size: cover;
background-position: center;
background-repeat: repeat; /* 背景平铺 */
}
/* 标题样式 */
h1 {
text-align: center;
color: #343a40; /* 主要文字颜色 */
}
/* 引导文档入口样式 */
.guide-entry-wrapper {
position: relative;
z-index: 10;
}
.guide-entry-btn {
background: linear-gradient(135deg, #007bff, #0056b3);
border: 2px solid #007bff;
color: white;
font-weight: 600;
padding: 8px 16px;
border-radius: 20px;
box-shadow: 0 4px 12px rgba(0, 123, 255, 0.3);
transition: all 0.3s ease;
animation: pulse-glow 2s infinite;
}
.guide-entry-btn:hover {
background: linear-gradient(135deg, #0056b3, #004085);
border-color: #0056b3;
color: white;
transform: translateY(-2px);
box-shadow: 0 6px 16px rgba(0, 123, 255, 0.4);
}
.guide-entry-btn:active {
transform: translateY(0);
box-shadow: 0 2px 8px rgba(0, 123, 255, 0.3);
}
.guide-entry-btn {
font-size: 14px;
}
/* 脉冲发光动画 */
@keyframes pulse-glow {
0%, 100% {
box-shadow: 0 4px 12px rgba(0, 123, 255, 0.3);
}
50% {
box-shadow: 0 4px 20px rgba(0, 123, 255, 0.5);
}
}
/* 卡片标题居中 */
.card-title {
text-align: center;
}
/* 卡片样式 */
.card {
border-radius: 8px; /* 圆角 */
box-shadow: 0 4px 8px rgba(0, 0, 0, 0.1); /* 阴影效果 */
}
/* 卡片标题 */
.card-header h2 {
margin-bottom: 0;
font-size: 1.5rem;
}
/* 按钮样式（仅限定 .btn，避免覆盖 Bootstrap 变体） */
.btn {
width: 100%;
margin-bottom: 10px;
padding: 10px;
border-radius: 5px;
transition: background-color 0.3s ease;
}
.btn:hover {
opacity: 0.9;
}
.btn:active {
transform: scale(0.98); /* 按下时缩小 */
will-change: transform; /* 提高动画性能 */
}
/* 表单选择器样式 */
.form-select {
margin-bottom: 10px;
}
/* 边框样式 */
.border {
border: 1px solid #ced4da;
padding: 10px;
border-radius: 4px;
}
/* 新添加的列样式 */
.col-md-3 {
padding: 0 10px; /* 添加内边距 */
margin-bottom: 20px; /* 调整列之间的间距 */
}
/* 加载遮罩和加载图标样式 */
#loading-overlay {
position: fixed;
top: 0;
left: 0;
width: 100%;
height: 100%;
background: rgba(0, 0, 0, 0.7);
z-index: 9999;
display: none; /* 默认隐藏 */
flex-direction: column;
align-items: center;
justify-content: center;
}
.loading-spinner {
width: 40px;
height: 40px;
border: 4px solid transparent; /* 设置光圈边框为透明 */
border-radius: 50%; /* 使光圈呈现圆形 */
animation: spin 1s linear infinite; /* 设置光圈旋转动画 */
background: conic-gradient(
red, orange, yellow, green, blue, indigo, violet, red
);
}
.loading-text {
color: #ffffff;
margin-top: 10px; /* 调整加载文本位置 */
}
/* 旋转动画 */
@keyframes spin {
0% { transform: rotate(0deg); }
100% { transform: rotate(360deg); }
}
/* 多选下拉菜单样式 */
.custom-multiselect {
position: relative;
}
.custom-multiselect .trigger {
text-align: left;
position: relative;
padding-right: 30px;
}
.custom-multiselect .trigger::after {
content: '▼';
position: absolute;
right: 10px;
top: 50%;
transform: translateY(-50%);
font-size: 12px;
color: #6c757d;
}
.custom-multiselect .menu {
position: absolute;
top: 100%;
left: 0;
right: 0;
background: white;
border: 1px solid #ced4da;
border-radius: 4px;
max-height: 200px;
overflow-y: auto;
z-index: 1000;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}
.custom-multiselect .menu label {
display: block;
padding: 8px 12px;
margin: 0;
cursor: pointer;
border-bottom: 1px solid #f8f9fa;
}
.custom-multiselect .menu label:hover {
background-color: #f8f9fa;
}
.custom-multiselect .menu label:last-child {
border-bottom: none;
}
.custom-multiselect .menu input[type="checkbox"] {
margin-right: 8px;
}
/* 系统日志实时显示区域样式 */
#syslog-live {
height: 200px;
overflow: auto;
background: #0b0e11;
color: #c7d5e0;
font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace;
font-size: 12px;
padding: 10px;
border-radius: 4px;
white-space: pre-wrap;
word-wrap: break-word;
}
/* 设备信息展开区域样式 */
#device-info-more {
margin-top: 10px;
padding-top: 10px;
border-top: 1px solid #e9ecef;
}
/* 截图预览区域样式 - 优化占位区域大小 */
#screenshot-preview {
min-height: 610px; /* 调整高度，与系统日志按钮上部平行 */
display: flex;
align-items: center;
justify-content: center;
background-color: #f8f9fa;
border: 1px solid #ced4da;
border-radius: 4px;
margin-top: 10px;
padding: 20px;
transition: all 0.3s ease; /* 平滑过渡效果 */
}
/* 当预览区域有内容时（工作状态）的样式 */
#screenshot-preview:not(:empty) {
min-height: 610px; /* 保持最小高度，确保与占位状态一致 */
align-items: stretch; /* 拉伸内容以填充整个容器 */
justify-content: flex-start; /* 内容左对齐 */
background-color: transparent; /* 移除背景色 */
padding: 0; /* 移除内边距，让内容完全填充 */
}
/* 工作状态下的内部容器样式 */
#screenshot-preview:not(:empty) > div {
width: 100%;
height: 100%;
min-height: 610px;
display: flex;
flex-direction: column;
}
/* 占位文本样式 */
#screenshot-preview:empty::before {
content: "截图与录屏预览区域";
color: #6c757d;
font-size: 0.875rem;
text-align: center;
}
/* 确保截图预览区域内的图片响应式 */
#screenshot-preview img {
max-width: 100%;
height: auto;
border-radius: 4px;
}
/* 录屏流样式 */
#screen-stream {
max-width: 100%;
height: auto;
border-radius: 4px;
}
/* 开发者模式状态提示样式 */
#devmode-status-tip {
font-size: 0.75rem;
color: #6c757d;
margin-top: 8px;
padding: 4px 8px;
border-radius: 4px;
background-color: #f8f9fa;
border: 1px solid #e9ecef;
}
#devmode-status-tip.enabled {
color: #198754;
background-color: #d1e7dd;
border-color: #badbcc;
}
#devmode-status-tip.disabled {
color: #dc3545;
background-color: #f8d7da;
border-color: #f5c2c7;
}
#devmode-status-tip.error {
color: #856404;
background-color: #fff3cd;
border-color: #ffeaa7;
}
#devmode-status-tip.checking {
color: #0c63e4;
background-color: #cfe2ff;
border-color: #b6d4fe;
}
/* 模态框样式 */
.modal {
display: none;
position: fixed;
z-index: 1000;
left: 0;
top: 0;
width: 100%;
height: 100%;
overflow: auto;
background-color: rgba(0, 0, 0, 0.4);
}
.modal-content {
background-color: #fefefe;
margin: 7.2% auto;
padding: 20px;
border: 1px solid #888;
width: 80%;
max-width: 600px;
border-radius: 5px;
}
#install-result h2 {
text-align: center;
color: #333;
margin-bottom: 20px;
}
#install-result h3 {
color: #555;
margin-top: 15px;
margin-bottom: 10px;
}
#failed-list, #success-list {
list-style-type: none;
padding-left: 0;
}
#failed-list li {
color: #d9534f;
margin-bottom: 10px;
}
#success-list li {
color: #5cb85c;
margin-bottom: 5px;
}
#success-list li::before {
content: "• ";
color: #5cb85c;
}
#close-result {
display: block;
width: 100%;
margin-top: 20px;
}
/* 错误信息样式 */
.error-message {
font-size: 0.9em;
color: #777;
margin-top: 5px;
margin-left: 15px;
}
/* footer样式 */
.footer {
color: #ffa500;
width: 100%;
position: fixed;
bottom: 0;
left: 0;
right: 0;
padding: 10px 0;
height: 50px;
font-weight: bold;
z-index: 10;
background-color: #f8f9fa;
}
/* 让“已选中的值”居中（含 Firefox） */
.text-center-select {
text-align-last: center;
-moz-text-align-last: center;
}
/* 让下拉列表里的每一行也尽量居中（多数浏览器有效） */
.text-center-select option {
text-align: center;
}
/* 修正 Bootstrap .form-select 的左右内边距不对称，避免视觉偏左 */
.text-center-select.form-select {
/* Bootstrap 默认：padding: 0.375rem 2.25rem 0.375rem 0.75rem; */
padding-left: 2.25rem; /* 与右侧一致，视觉真正居中 */
padding-right: 2.25rem; /* 保持默认 */
background-position: right 0.75rem center; /* 保持箭头位置 */
}
/* 可选：过长时省略号，别撑破 */
#device-select {
white-space: nowrap;
overflow: hidden;
text-overflow: ellipsis;
}
.crash-checkbox-list {
max-height: 180px;
overflow: auto;
}
.custom-multiselect {
position: relative;
}
.custom-multiselect .menu {
position: absolute;
z-index: 1050;
width: 100%;
max-height: 220px;
overflow: auto;
background: #fff;
border: 1px solid #ced4da;
border-radius: 4px;
padding: 6px 8px;
}
.custom-multiselect .menu .menu-toolbar { position: sticky; top: 0; background: #fff; z-index: 1; padding-bottom: 4px; margin-bottom: 6px; border-bottom: 1px solid #e9ecef; }
.custom-multiselect .menu .form-check {
margin-bottom: 4px;
}
.custom-multiselect .trigger {
display: flex;
align-items: center;
justify-content: space-between;
}
/* 设备信息统一样式 */
#device-info.info-list { min-height: 10px; }
#device-info .info-section-title {
font-weight: 600;
margin-top: .5rem;
margin-bottom: .25rem;
color: #212529;
}
#device-info > div { line-height: 1.5; margin-bottom: .125rem; }
#device-info > div strong { display: inline-block; min-width: 7.5em; font-weight: 600; }
#device-info > div, #device-info > div * { word-break: break-all; }
#device-info .info-k { display:inline-block; min-width:3em; margin-right:.25rem; font-weight:400; color:#212529; }
/* 回退卡片等高与白底相关样式，恢复原有外观 */
/* 移动端优化 */
@media (max-width: 576px) {
body { padding: 10px; -webkit-tap-highlight-color: transparent; }
.card { margin-bottom: 12px; }
.card-header { min-height: 40px; }
.card-header .card-title { font-size: 1rem; }
.card .btn, .card .form-control { min-height: 44px; font-size: 16px; } /* 避免 iOS 自动放大 */
.custom-multiselect .menu { max-height: 60vh; }
.custom-multiselect .menu .menu-toolbar { padding: 6px 4px; }
#device-info .info-k { min-width: 4em; }
}
/* 设备信息底部折叠按钮 */
#device-info .info-collapse-btn-wrapper {
display: flex;
justify-content: center;
margin-top: 8px;
}
.info-collapse-btn {
width: 28px;
height: 28px;
padding: 0;
border-radius: 50%;
display: inline-flex;
align-items: center;
justify-content: center;
background: #f8f9fa;
border: 1px solid #dee2e6;
transition: all .2s ease;
}
.info-collapse-btn:hover {
background: #e9ecef;
border-color: #adb5bd;
}
.info-collapse-btn svg {
width: 14px;
height: 14px;
transition: transform .2s ease;
}
.info-collapse-btn[aria-expanded="true"] svg {
transform: rotate(180deg);
}
@media (max-width: 576px) {
#device-info .info-collapse-btn-wrapper {
margin-top: 6px;
}
.info-collapse-btn {
width: 24px;
height: 24px;
}
.info-collapse-btn svg {
width: 12px;
height: 12px;
}
}
.hl { background: #fffb91; color: #222; padding: 0 1px; border-radius: 2px; }
/* Debug触发器样式 */
#debug-trigger {
transition: all 0.2s ease;
border-radius: 4px;
padding: 2px 6px;
}
#debug-trigger:hover {
background-color: rgba(0, 0, 0, 0.05);
transform: scale(1.02);
}
#debug-trigger:active {
transform: scale(0.98);
background-color: rgba(0, 0, 0, 0.1);
}
/* Debug触发器激活状态（可选） */
#debug-trigger.debug-active {
background-color: rgba(255, 193, 7, 0.1);
border: 1px solid rgba(255, 193, 7, 0.3);
}6、common_utils代码
import re
import os
import cv2
import json
import time
import uuid
import socket
import signal
import zipfile
import logging
import datetime
import requests
import threading
import subprocess
# 条件导入
try:
import psutil
except ImportError:
psutil = None
try:
import numpy as np
except ImportError:
np = None
from typing import Any, Dict, List, Optional
from requests.exceptions import RequestException
def is_tunnel_error(text: Optional[str]) -> bool:
"""粗略判断输出是否与 tunnel/agent 未就绪相关，以触发升级检查。
不依赖精确文案，尽量匹配常见错误关键词。
"""
s = (text or "").lower()
if not s:
return False
patterns = [
"agent is not running",
"failed to get tunnel",
"serve tunnel",
"tunnel server",
"connectex",
"connection refused",
"actively refused",
":60105",
"bind",
"failed to start tunnel",
]
return any(p in s for p in patterns)
def run_with_quick_check_and_escalate(prechecker, udid: str, exec_fn):
"""
先执行 quick_check（不启动 tunnel），执行命令；
若失败且判断为隧道相关，再执行 check_all 启动/修复隧道后重试。
:param prechecker: IOSPrechecker 实例
:param udid: 设备 UDID
:param exec_fn: 可调用，无参，返回 (ok: bool, raw: Any)
:return: (ok, raw)
"""
try:
ok, msg = prechecker.quick_check(udid)
if not ok:
return False, msg
except (AttributeError, ValueError, TypeError, RuntimeError, OSError):
# quick_check 异常时也不阻断，继续尝试执行
pass
try:
ok1, out1 = exec_fn()
except (AttributeError, ValueError, TypeError, RuntimeError, OSError) as e:
# 将异常文案视作 raw 以便后续判断
ok1, out1 = False, str(e)
if ok1:
return True, out1
if is_tunnel_error(str(out1)):
try:
ok_t, _ = prechecker.check_all(udid)
except (AttributeError, ValueError, TypeError, RuntimeError, OSError):
ok_t = False
if ok_t:
try:
return exec_fn()
except (AttributeError, ValueError, TypeError, RuntimeError, OSError) as e:
return False, str(e)
return ok1, out1
def get_local_ip() -> Optional[str]:
"""获取本地IP地址；失败返回 None。"""
sock: Optional[socket.socket] = None
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.connect(("8.8.8.8", 80))
return sock.getsockname()[0]
except Exception as e:
logging.info("获取本地IP失败: %s", e)
return None
finally:
if sock is not None:
sock.close()
# === iOS 设备信息规范化 ===
UNKNOWN = "UNKNOWN"
def _pick(d: Dict[str, Any], keys: List[str], default: Optional[str] = UNKNOWN):
"""从字典 d 里按顺序取第一个有值(key 存在且值不为空)的键。"""
for k in keys:
if isinstance(d, dict) and k in d and d[k] not in (None, "", []):
return d[k]
return default
# 展示顺序
ORDERED_IOS_INFO_LABELS: List[str] = [
"设备品牌",
"产品型号",
"内部型号",
"系统版本",
"CPU 架构",
"激活状态",
"设备区域",
"设备时区",
"UDID",
"IMEI信息①",
"IMEI信息②",
"Wi-Fi地址",
"蓝牙地址",
"以太网地址",
]
def format_product_model(model: Optional[str]) -> str:
s = str(model or "").strip()
return s.split(",")[0] if s else s
def normalize_ios_info(raw: Dict[str, Any]) -> Dict[str, Any]:
"""规范化 go-ios JSON 输出"""
info_map: Dict[str, Any] = {
"设备品牌": _pick(raw, ["ProductName"]),
"产品型号": format_product_model(_pick(raw, ["ProductType"])),
"内部型号": _pick(raw, ["ModelNumber"]),
"系统版本": _pick(raw, ["HumanReadableProductVersionString", "ProductVersion"]),
"CPU 架构": _pick(raw, ["CPUArchitecture"]),
"激活状态": _pick(raw, ["ActivationState"]),
"设备区域": _pick(raw, ["RegionInfo"]),
"设备时区": _pick(raw, ["TimeZone"]),
"UDID": _pick(raw, ["UniqueDeviceID", "UDID"]),
"IMEI信息①": _pick(raw, ["InternationalMobileEquipmentIdentity"]),
"IMEI信息②": _pick(raw, ["InternationalMobileEquipmentIdentity2"]),
"Wi-Fi地址": _pick(raw, ["WiFiAddress"]),
"蓝牙地址": _pick(raw, ["BluetoothAddress"]),
"以太网地址": _pick(raw, ["EthernetAddress"]),
}
for k, v in list(info_map.items()):
if v in (None, "", []):
info_map[k] = UNKNOWN
return info_map
def build_ordered_ios_info(info_map: Dict[str, Any]) -> List[Dict[str, str]]:
"""转成有序数组"""
return [{"label": lab, "value": str(info_map.get(lab, UNKNOWN))} for lab in ORDERED_IOS_INFO_LABELS]
# === 设备列表规范化 ===
def normalize_ios_list_output(raw: str) -> List[Dict[str, Any]]:
"""规整 go-ios list 输出"""
devices: List[Dict[str, Any]] = []
raw = (raw or "").strip()
if not raw:
return devices
try:
data = json.loads(raw)
except json.JSONDecodeError:
data = None
def pick(rec: Dict[str, Any]) -> Dict[str, Any]:
return {
"udid": rec.get("udid") or rec.get("UDID") or rec.get("Udid") or rec.get("UniqueDeviceID") or rec.get(
"serialNumber"),
"name": rec.get("name") or rec.get("DeviceName") or rec.get("ProductName"),
"model": format_product_model(rec.get("model") or rec.get("ProductType") or rec.get("productType")),
"version": rec.get("version") or rec.get("ProductVersion") or rec.get("productVersion"),
}
if isinstance(data, dict):
arr = data.get("deviceList") or data.get("devices") or []
for obj in arr:
if isinstance(obj, dict):
nd = pick(obj)
if nd["udid"]:
devices.append(nd)
elif isinstance(data, list):
for obj in data:
if isinstance(obj, str):
devices.append({"udid": obj, "name": None, "model": None, "version": None})
elif isinstance(obj, dict):
nd = pick(obj)
if nd["udid"]:
devices.append(nd)
else:
for line in raw.splitlines():
val = line.strip()
if val and len(val) >= 16:
devices.append({"udid": val, "name": None, "model": None, "version": None})
seen = set()
uniq: List[Dict[str, Any]] = []
for dev in devices:
uid = dev.get("udid")
if uid and uid not in seen:
seen.add(uid)
uniq.append(dev)
return uniq
# === 应用列表解析 ===
def _is_system_bundle(bundle_id: Optional[str], app_type: Optional[str]) -> bool:
if app_type and isinstance(app_type, str) and "system" in app_type.lower():
return True
return bool(bundle_id) and str(bundle_id).startswith("com.apple.")
def parse_ios_apps(raw: str, third_party_only: bool = True) -> List[Dict[str, str]]:
"""解析 go-ios apps 输出"""
result: List[Dict[str, str]] = []
raw = raw or ""
try:
data = json.loads(raw)
except json.JSONDecodeError:
data = None
def push(bundle_id: str, title: str = "", version: str = "", app_type: str = ""):
if not bundle_id:
return
if third_party_only and _is_system_bundle(bundle_id, app_type):
return
result.append({"bundleId": bundle_id, "name": title, "version": version})
if isinstance(data, list):
for obj in data:
if isinstance(obj, dict):
bid = obj.get("CFBundleIdentifier") or obj.get("bundleID") or obj.get("id")
app_name = obj.get("CFBundleDisplayName") or obj.get("CFBundleName") or obj.get("name") or ""
ver = obj.get("CFBundleShortVersionString") or obj.get("version") or ""
typ = obj.get("ApplicationType") or obj.get("appType") or ""
push(str(bid or ""), str(app_name or ""), str(ver or ""), str(typ or ""))
elif isinstance(data, dict):
possible = data.get("apps") or data.get("applications") or []
for obj in possible:
if isinstance(obj, dict):
bid = obj.get("CFBundleIdentifier") or obj.get("bundleID") or obj.get("id")
app_name = obj.get("CFBundleDisplayName") or obj.get("CFBundleName") or obj.get("name") or ""
ver = obj.get("CFBundleShortVersionString") or obj.get("version") or ""
typ = obj.get("ApplicationType") or obj.get("appType") or ""
push(str(bid or ""), str(app_name or ""), str(ver or ""), str(typ or ""))
else:
for line in raw.splitlines():
m = re.match(r"^s*([a-zA-Z0-9._-]+)s*(?:-s*([^-
]+))?(?:-s*([^
]+))?", line)
if not m:
continue
bid = (m.group(1) or "").strip()
if not bid:
continue
if third_party_only and bid.startswith("com.apple."):
continue
app_name = (m.group(2) or "").strip()
ver = (m.group(3) or "").strip()
result.append({"bundleId": bid, "name": app_name, "version": ver})
seen = set()
uniq: List[Dict[str, str]] = []
for app in result:
bid = app.get("bundleId")
if bid and bid not in seen:
seen.add(bid)
uniq.append(app)
return uniq
# === 工具函数 ===
def find_free_port(start_port=60105, max_tries=20):
for port in range(start_port, start_port + max_tries):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
try:
s.bind(("127.0.0.1", port))
return port
except OSError:
continue
return None
def check_port_available(port: int) -> bool:
"""检查端口是否可用"""
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
result = s.connect_ex(('127.0.0.1', port))
return result != 0 # 连接失败说明端口可用
except (OSError, socket.error):
return False
def wait_for_mjpeg_stream(mjpeg_url: str, process, max_wait_seconds: int = 15, logger=None) -> bool:
"""等待 MJPEG 流就绪"""
if logger:
logger.debug("等待 MJPEG 流就绪: %s", mjpeg_url)
for attempt in range(max_wait_seconds):
try:
response = requests.get(mjpeg_url, timeout=2, stream=True)
if response.status_code == 200:
response.close()
if logger:
logger.debug("MJPEG 流已就绪 (尝试 %d/%d)", attempt + 1, max_wait_seconds)
return True
except (requests.RequestException, OSError, IOError):
pass
if process.poll() is not None:
if logger:
error_msg = f"MJPEG 流进程异常退出 (code: {process.returncode})"
try:
stdout, stderr = process.communicate(timeout=1)
if stdout:
error_msg += f"
stdout: {stdout}"
if stderr:
error_msg += f"
stderr: {stderr}"
except (ValueError, OSError, subprocess.TimeoutExpired):
pass
logger.error(error_msg)
return False
time.sleep(1)
if attempt % 3 == 0 and logger:
logger.debug("等待 MJPEG 流启动 (%d/%d)...", attempt + 1, max_wait_seconds)
return False
def cleanup_old_tunnel_processes(tunnel_manager=None, logger=None) -> None:
"""清理旧 tunnel 进程"""
try:
if tunnel_manager:
try:
tunnel_manager.stop()
if logger:
logger.debug("已停止现有 tunnel 进程")
except Exception as e:
if logger:
logger.debug("停止 tunnel 时异常: %s", e)
if psutil is None:
if logger:
logger.debug("psutil 不可用，跳过进程清理")
return
killed = []
for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
try:
cmdline = ' '.join(proc.info.get('cmdline') or []).lower()
if 'ios' in proc.info.get('name', '').lower() and 'tunnel' in cmdline:
if logger:
logger.warning("发现旧 tunnel 进程 PID %d: %s", proc.info['pid'], cmdline)
proc.terminate()
proc.wait(timeout=5)
killed.append(proc.info['pid'])
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.TimeoutExpired):
continue
if killed and logger:
logger.debug("已清理旧 tunnel 进程: %s", killed)
except Exception as e:
if logger:
logger.warning("清理旧 tunnel 进程异常: %s", e)
def cleanup_all_ios_processes(logger=None) -> None:
"""清理所有ios.exe进程（包括可能遗留的进程）"""
try:
import psutil
killed = []
for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
try:
if proc.info['name'] and proc.info['name'].lower() in ['ios.exe', 'ios']:
# 强制终止所有ios.exe进程
proc.terminate()
try:
proc.wait(timeout=3)
except psutil.TimeoutExpired:
proc.kill()
proc.wait(timeout=2)
killed.append(proc.info['pid'])
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.TimeoutExpired):
continue
if killed and logger:
logger.info("已清理 %d 个ios.exe进程: %s", len(killed), killed)
elif logger:
logger.debug("未发现需要清理的ios.exe进程")
except ImportError:
if logger:
logger.warning("psutil未安装，无法清理ios.exe进程")
except Exception as e:
if logger:
logger.warning("清理ios.exe进程异常: %s", e)
def create_required_directories(config, logger=None) -> None:
"""创建必要的目录"""
dirs = [
getattr(config, "UPLOAD_FOLDER", None),
getattr(config, "LOG_DIR", None),
getattr(config, "GOIOS_DIR", None),
getattr(config, "GOIOS_EXECUTABLE_DIR", None),
getattr(config, "DEVIMAGES_DIR", None),
os.path.join(getattr(config, "IOS_HOME", ""), 'wintun', 'amd64'),
os.path.join(getattr(config, "IOS_HOME", ""), 'wintun', 'arm'),
os.path.join(getattr(config, "IOS_HOME", ""), 'wintun', 'arm64'),
os.path.join(getattr(config, "IOS_HOME", ""), 'wintun', 'x86'),
]
for d in dirs:
if d:
try:
os.makedirs(d, exist_ok=True)
if logger:
logger.debug("确保目录存在: %s", d)
except OSError as e:
if logger:
logger.warning("创建目录失败 %s: %s", d, e)
# 启动下载目录清理守护线程（仅启动一次）
try:
download_dir = getattr(config, "UPLOAD_FOLDER", None)
if download_dir:
_start_downloads_janitor(download_dir, logger=logger)
except Exception as e:
if logger:
logger.debug("启动下载目录清理线程失败: %s", e)
def to_int(value, default: Optional[int]) -> Optional[int]:
try:
return int(value)
except (TypeError, ValueError):
return default
def extract_goios_opts(src: Dict[str, Any]) -> Dict[str, Any]:
return {
"address": (str(src.get("address") or "").strip() or None),
"rsd_port": to_int(src.get("rsd_port"), None),
"userspace_port": to_int(src.get("userspace_port"), None),
"proxyurl": (str(src.get("proxyurl") or "").strip() or None),
"tunnel_info_port": to_int(src.get("tunnel_info_port"), None),
"verbose": bool(src.get("verbose", False)),
"trace": bool(src.get("trace", False)),
"nojson": bool(src.get("nojson", False)),
"pretty": bool(src.get("pretty", False)),
}
def terminate_process(process) -> None:
"""强制终止进程，包括子进程"""
try:
if process and hasattr(process, "poll") and process.poll() is None:
# 使用psutil强制终止进程及其子进程
try:
import psutil
parent = psutil.Process(process.pid)
children = parent.children(recursive=True)
# 先终止子进程
for child in children:
try:
child.terminate()
except (psutil.NoSuchProcess, psutil.AccessDenied):
pass
# 等待子进程终止
try:
psutil.wait_procs(children, timeout=3)
except psutil.TimeoutExpired:
# 强制杀死未终止的子进程
for child in children:
try:
if child.is_running():
child.kill()
except (psutil.NoSuchProcess, psutil.AccessDenied):
pass
# 终止父进程
try:
parent.terminate()
parent.wait(timeout=5)
except (psutil.TimeoutExpired, psutil.NoSuchProcess):
try:
parent.kill()
except (psutil.NoSuchProcess, psutil.AccessDenied):
pass
except ImportError:
# 如果没有psutil，使用传统方法
if os.name == "nt":
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
else:
os.kill(process.pid, signal.SIGTERM)
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
os.kill(process.pid, signal.SIGKILL)
except Exception as e:
logging.warning("终止子进程失败: %s", e)
def safe_filename(text: str) -> str:
return re.sub(r"[^wu4e00-u9fa5.-]", "", str(text or ""))
def get_device_model(udid: str, goios_manager) -> str:
ok, out = goios_manager.list_devices(details=True)
if not ok or not out:
return "iOSDevice"
try:
devices = normalize_ios_list_output(out)
for d in devices:
if d.get("udid") == udid:
return safe_filename(format_product_model(d.get("model") or "iOSDevice"))
except (ValueError, KeyError, TypeError) as e:
logging.debug("解析设备信息失败: %s", e)
return "iOSDevice"
def now_timestamp_str() -> str:
return datetime.datetime.now().strftime("%Y%m%d%H%M%S")
# === MJPEG 录屏 ===
def _start_manual_mjpeg_recording(
mjpeg_url: str,
out_dir: str,
basename: str,
stop_evt: threading.Event,
logger: Optional[logging.Logger] = None,
) -> Dict[str, Any]:
"""
手动解析 MJPEG 流并保存为 MP4 文件。
"""
mp4_name = f"{basename}.mp4"
mp4_path = os.path.join(out_dir, mp4_name)
def _worker():
try:
with requests.get(mjpeg_url, stream=True, timeout=10) as r:
r.raise_for_status()
buffer = b""
writer: Optional[Any] = None
frame: Optional[Any] = None
target_w: Optional[int] = None
target_h: Optional[int] = None
# 目标帧率（可通过环境变量 RECORD_FPS 配置），用于还原真实时长
try:
target_fps = float(os.environ.get("RECORD_FPS", "30"))
if not (1.0 <= target_fps <= 60.0):
target_fps = 12.0
except (ValueError, TypeError):
target_fps = 12.0
last_ts = None
frame_acc = 0.0
max_dup_per_tick = 5 # 一次最多重复写入帧数，限制卡顿时的爆发
for chunk in r.iter_content(chunk_size=4096):
if stop_evt.is_set():
break
if not chunk:
continue
buffer += chunk
# 尝试从 buffer 中提取尽可能多的完整 JPEG 帧
while True:
start = buffer.find(b"xffxd8")
end = buffer.find(b"xffxd9")
if start != -1 and end != -1 and end > start:
jpg_bytes = buffer[start:end + 2]
buffer = buffer[end + 2:]
else:
break
arr = np.frombuffer(jpg_bytes, dtype=np.uint8)
frame = cv2.imdecode(arr, cv2.IMREAD_COLOR)
if frame is None:
continue
h, w = frame.shape[:2]
# 初始化写入器，统一视频尺寸（强制偶数，提升兼容性）
if writer is None:
target_w = w - (w % 2)
target_h = h - (h % 2)
if target_w <= 0 or target_h <= 0:
target_w, target_h = w, h
if target_w != w or target_h != h:
frame = cv2.resize(frame, (target_w, target_h))
h, w = frame.shape[:2]
fourcc = _safe_fourcc("mp4v")
writer = cv2.VideoWriter(mp4_path, fourcc, target_fps, (w, h))
last_ts = time.time()
if logger:
logger.debug("初始化 VideoWriter: %s [%dx%d @ %.1f fps]",
mp4_path, w, h, target_fps)
else:
# 若帧尺寸变化（如旋转），统一到初始尺寸
if (w != target_w) or (h != target_h):
frame = cv2.resize(frame, (target_w, target_h))
# 按时间间隔平滑计算应写入帧数，保持生成视频时长 ≈ 实际时长
now_ts = time.time()
if last_ts is None:
frame_acc += 1.0
else:
dt = max(0.0, now_ts - last_ts)
frame_acc += dt * target_fps
last_ts = now_ts
# 平滑写入，限幅避免卡顿时爆量重复
frames_to_write = int(frame_acc)
if frames_to_write > 0:
frames_to_write = min(frames_to_write, max_dup_per_tick)
frame_acc -= frames_to_write
for _ in range(frames_to_write):
writer.write(frame)
except requests.RequestException as e:
if logger:
if stop_evt.is_set():
# 停止录制导致的连接中断，视为正常结束
logger.debug("MJPEG 流已结束")
else:
logger.error("MJPEG 请求失败: %s", e)
except (OSError, IOError) as e:
if logger:
logger.error("文件写入失败: %s", e)
except Exception as e: # 更具体的兜底
if logger:
logger.error("MJPEG 手动录制异常: %s", e, exc_info=True)
finally:
# 确保写入器被正确释放，写入 moov 等元数据
try:
if 'writer' in locals() and writer is not None:
writer.release()
if logger:
logger.debug("已关闭视频写入器: %s", mp4_path)
except Exception as e:
if logger:
logger.warning("释放 VideoWriter 失败: %s", e)
th = threading.Thread(target=_worker, daemon=True, name=f"ManualMJPEG-{basename}")
th.start()
return {"stop_evt": stop_evt, "thread": th, "path": mp4_path, "name": mp4_name, "format": "mp4"}
def _record_mjpeg_fallback(mjpeg_url: str, out_path: str, stop_event: threading.Event,
logger: Optional[logging.Logger] = None) -> None:
"""回退：直接把 MJPEG 流保存为 .mjpeg"""
try:
with requests.get(mjpeg_url, stream=True, timeout=15) as r:
r.raise_for_status()
with open(out_path, "wb") as f:
for chunk in r.iter_content(chunk_size=8192):
if stop_event.is_set():
return
if chunk:
f.write(chunk)
if logger:
logger.debug("MJPEG 录制完成: %s", out_path)
except RequestException as e:
if logger:
logger.warning("MJPEG 请求错误: %s", e)
except (OSError, IOError) as e:
if logger:
logger.error("MJPEG 文件写入错误: %s", e)
def _safe_fourcc(codec_str: str):
try:
if hasattr(cv2, "VideoWriter_fourcc"):
return cv2.VideoWriter_fourcc(*codec_str)
return sum(ord(c) << (8 * i) for i, c in enumerate(codec_str[:4]))
except (AttributeError, TypeError, ValueError) as e:
logging.debug("生成 fourcc 失败: %s, 使用默认 mp4v", e)
return 0x34766D70 # 默认 mp4v
def _start_mjpeg_fallback(mjpeg_url: str, out_dir: str, basename: str,
logger: Optional[logging.Logger] = None) -> Dict[str, Any]:
stop_evt = threading.Event()
mjpeg_name = f"{basename}.mjpeg"
mjpeg_path = os.path.join(out_dir, mjpeg_name)
def _worker():
_record_mjpeg_fallback(mjpeg_url, mjpeg_path, stop_evt, logger)
th = threading.Thread(target=_worker, daemon=True, name=f"MJPEG-{basename}")
th.start()
return {"stop_evt": stop_evt, "thread": th, "path": mjpeg_path, "name": mjpeg_name, "format": "mjpeg"}
def start_mjpeg_to_mp4(mjpeg_url: str, out_dir: str, basename: str,
logger: Optional[logging.Logger] = None) -> Dict[str, Any]:
"""
启动录屏：默认使用手动 MJPEG 解析转 MP4。
如果 numpy/cv2 不可用，则退回保存为 .mjpeg。
"""
os.makedirs(out_dir, exist_ok=True)
if np is None or cv2 is None:
if logger:
logger.warning("缺少 numpy 或 OpenCV，回退到 .mjpeg 保存方案")
return _start_mjpeg_fallback(mjpeg_url, out_dir, basename, logger)
stop_evt = threading.Event()
return _start_manual_mjpeg_recording(mjpeg_url, out_dir, basename, stop_evt, logger)
def stop_recorder(rec_ctx: Dict[str, Any], join_timeout: float = 5.0) -> None:
if not isinstance(rec_ctx, dict):
return
stop_evt = rec_ctx.get("stop_evt")
th = rec_ctx.get("thread")
if isinstance(stop_evt, threading.Event):
stop_evt.set()
if isinstance(th, threading.Thread) and th.is_alive():
th.join(timeout=join_timeout)
if th.is_alive():
logging.warning("录制线程 %s 在 %.1f 秒后仍在运行", th.name, join_timeout)
# === 下载目录自动清理 ===
_downloads_janitor_started = False
_downloads_janitor_lock = threading.Lock()
def _downloads_cleanup_once(download_dir: str, max_age_seconds: int = 1800,
logger: Optional[logging.Logger] = None) -> None:
"""执行一次清理：删除下载目录内超过 max_age_seconds 的普通文件。
- 忽略子目录
- 文件被占用/权限错误时跳过
- 仅输出 debug 日志，避免噪声
"""
now = time.time()
try:
for name in os.listdir(download_dir):
path = os.path.join(download_dir, name)
try:
if not os.path.isfile(path):
continue
# 计算文件年龄（按 mtime）
try:
mtime = os.path.getmtime(path)
except OSError:
continue
if (now - float(mtime)) <= max_age_seconds:
continue
# 尝试删除；被占用或无权限时跳过
try:
os.remove(path)
if logger:
logger.debug("已清理过期文件: %s", path)
except PermissionError:
# 可能仍在写入/占用，跳过
if logger:
logger.debug("文件占用，跳过清理: %s", path)
except OSError as e:
if logger:
logger.debug("删除失败，跳过 %s: %s", path, e)
except (OSError, PermissionError) as e:
if logger:
logger.debug("处理文件异常，跳过 %s: %s", path, e)
continue
except FileNotFoundError:
# 目录被外部删除
if logger:
logger.debug("下载目录不存在，跳过清理: %s", download_dir)
except Exception as e:
if logger:
logger.debug("清理任务异常: %s", e)
def _start_downloads_janitor(download_dir: str,
logger: Optional[logging.Logger] = None,
max_age_seconds: int = 300,
scan_interval_seconds: int = 60) -> None:
"""启动后台清理线程（仅启动一次）。"""
global _downloads_janitor_started
if not download_dir:
return
with _downloads_janitor_lock:
if _downloads_janitor_started:
return
_downloads_janitor_started = True
def _worker():
# 首次延迟，避免与启动阶段 I/O 冲突
time.sleep(5)
while True:
_downloads_cleanup_once(download_dir, max_age_seconds=max_age_seconds, logger=logger)
time.sleep(max(15, scan_interval_seconds))
th = threading.Thread(target=_worker, daemon=True, name="DownloadsJanitor")
th.start()
# === 签名下载令牌（通用） ===
_SIGNED_DOWNLOADS_STORE: Dict[str, Dict[str, Any]] = {}
def create_signed_download_token(filename: str, ttl_seconds: int = 600) -> str:
token = uuid.uuid4().hex if 'uuid' in globals() else str(int(time.time() * 1000))
expire_at = time.time() + max(30, ttl_seconds)
rel = str(filename).replace("\", "/").lstrip("/")
_SIGNED_DOWNLOADS_STORE[token] = {"file": rel, "expire": expire_at}
return token
def consume_signed_download_token(token: str) -> str:
info = _SIGNED_DOWNLOADS_STORE.pop(token, None)
if not info:
raise KeyError("invalid token")
if time.time() > float(info.get("expire", 0)):
raise KeyError("expired token")
return str(info.get("file"))
# === ps --apps 解析 ===
def parse_ps_apps_raw(raw: str) -> List[Dict[str, Any]]:
result: List[Dict[str, Any]] = []
try:
data = json.loads(raw or "")
if isinstance(data, list):
for p in data:
if not isinstance(p, dict):
continue
result.append({
"Name": p.get("Name"),
"Pid": p.get("Pid") or p.get("PID") or p.get("pid"),
"RealAppName": p.get("RealAppName"),
})
except json.JSONDecodeError:
pass
return result
# === 电池详情合并 ===
def build_battery_detail(goios_manager, udid: str, **opts) -> Dict[str, Any]:
ok1, raw1 = goios_manager.battery_info(udid, **opts)
ok2, raw2 = getattr(goios_manager, 'battery_registry', lambda *a, **k: (False, None))(udid, **opts)
detail = {"batterycheck": None, "batteryregistry": None}
try:
if ok1 and isinstance(raw1, str) and raw1.strip():
detail["batterycheck"] = json.loads(raw1)
except json.JSONDecodeError:
detail["batterycheck"] = None
try:
if ok2 and isinstance(raw2, str) and raw2.strip():
detail["batteryregistry"] = json.loads(raw2)
except json.JSONDecodeError:
detail["batteryregistry"] = None
return detail
# === 磁盘信息解析（支持 JSON 或文本） ===
def parse_diskspace_summary(raw: str) -> Dict[str, Any]:
summary = {"BlockSize": None, "FreeSpace": None, "UsedSpace": None, "TotalSpace": None}
if not raw:
return summary
try:
data = json.loads(raw)
if isinstance(data, dict):
summary["BlockSize"] = data.get("BlockSize")
summary["FreeSpace"] = data.get("FreeSpace") or data.get("free")
summary["UsedSpace"] = data.get("UsedSpace") or data.get("used")
summary["TotalSpace"] = data.get("TotalSpace") or data.get("total")
return summary
except json.JSONDecodeError:
pass
text = str(raw)
import re as _re
mb = _re.search(r"BlockSize:s*([0-9.]+)", text, _re.I)
mf = _re.search(r"FreeSpace:s*([^
]+)", text, _re.I)
mu = _re.search(r"UsedSpace:s*([^
]+)", text, _re.I)
mt = _re.search(r"TotalSpace:s*([^
]+)", text, _re.I)
summary["BlockSize"] = (mb.group(1) + "KB") if mb else None
summary["FreeSpace"] = mf.group(1).strip() if mf else None
summary["UsedSpace"] = mu.group(1).strip() if mu else None
summary["TotalSpace"] = mt.group(1).strip() if mt else None
return summary
# === Crash 导出/删除 ===
def crash_export_zip(goios_manager, udid: str, patterns: List[str], crash_root: str,
logger: Optional[logging.Logger] = None) -> Dict[str, Any]:
os.makedirs(crash_root, exist_ok=True)
batch_dir = os.path.join(crash_root, uuid.uuid4().hex if 'uuid' in globals() else str(int(time.time())))
os.makedirs(batch_dir, exist_ok=True)
ok_all = True
raws: List[str] = []
pats = patterns or ["*"]
for pat in pats:
ok_one, raw_one = goios_manager.crash_cp(udid, str(pat), batch_dir)
ok_all = ok_all and ok_one
raws.append(raw_one)
zip_name = os.path.basename(batch_dir) + ".zip"
zip_path = os.path.join(crash_root, zip_name)
try:
with zipfile.ZipFile(zip_path, 'w', compression=zipfile.ZIP_DEFLATED) as z:
for root, _, files in os.walk(batch_dir):
for f in files:
full = os.path.join(root, f)
arc = os.path.relpath(full, batch_dir)
z.write(full, arcname=arc)
except Exception as exc:
if logger:
logger.warning("打包 crash 目录失败: %s", exc)
return {"ok": ok_all, "zip_path": zip_path, "raw": "
".join(raws)}
def crash_remove_many(goios_manager, udid: str, patterns: List[str], cwd: str = ".", recursive: bool = False) -> Dict[
str, Any]:
pats = patterns or ["*"]
all_ok = True
raws: List[str] = []
results: List[Dict[str, Any]] = []
for pat in pats:
try:
ok_rm, raw_rm = goios_manager.crash_rm(udid, cwd, str(pat), recursive=recursive)
all_ok = all_ok and ok_rm
raws.append(raw_rm)
# 记录每个pattern的删除结果
result = {
"pattern": pat,
"success": ok_rm,
"output": raw_rm,
"cwd": cwd,
"recursive": recursive
}
# 分析失败原因
if not ok_rm:
if "permission denied" in raw_rm.lower() or "access denied" in raw_rm.lower():
result["error_type"] = "permission_denied"
result["suggestion"] = "检查文件权限或设备锁定状态"
elif "no such file" in raw_rm.lower() or "file not found" in raw_rm.lower():
result["error_type"] = "file_not_found"
result["suggestion"] = "文件可能已被删除或路径不正确"
elif "usage:" in raw_rm.lower() or "help" in raw_rm.lower():
result["error_type"] = "invalid_parameters"
result["suggestion"] = "检查pattern和cwd参数格式"
else:
result["error_type"] = "unknown_error"
result["suggestion"] = "检查go-ios命令输出获取更多信息"
results.append(result)
except Exception as e:
all_ok = False
error_msg = f"删除pattern '{pat}'时发生异常: {str(e)}"
raws.append(error_msg)
results.append({
"pattern": pat,
"success": False,
"output": error_msg,
"error_type": "exception",
"suggestion": "检查go-ios进程状态和网络连接"
})
return {
"ok": all_ok,
"raw": "
".join(raws),
"results": results,
"summary": {
"total_patterns": len(pats),
"successful_deletions": sum(1 for r in results if r["success"]),
"failed_deletions": sum(1 for r in results if not r["success"]),
"cwd": cwd,
"recursive": recursive
}
}
def crash_export_collect(goios_manager, udid: str, patterns: List[str], crash_root: str,
logger: Optional[logging.Logger] = None) -> Dict[str, Any]:
os.makedirs(crash_root, exist_ok=True)
batch_dir = os.path.join(crash_root, uuid.uuid4().hex if 'uuid' in globals() else str(int(time.time())))
os.makedirs(batch_dir, exist_ok=True)
ok_all = True
raws: List[str] = []
pats = patterns or ["*"]
for pat in pats:
pat_str = str(pat)
tried = []
# 先按原样
ok_one, raw_one = goios_manager.crash_cp(udid, pat_str, batch_dir)
tried.append((pat_str, ok_one, raw_one))
# 若像目录名（不含通配符且不含点后缀），追加 "/*" 重试
if not ok_one:
if all(ch not in pat_str for ch in ("*", "?")) and "." not in pat_str:
pat_dir = pat_str.rstrip("/") + "/*"
ok_two, raw_two = goios_manager.crash_cp(udid, pat_dir, batch_dir)
tried.append((pat_dir, ok_two, raw_two))
ok_one = ok_two
raw_one = (raw_one or "") + ("
" + (raw_two or ""))
ok_all = ok_all and ok_one
raws.append(raw_one)
# 列出 batch_dir 下的所有文件（扁平化相对路径）
files: List[str] = []
for root, _, fnames in os.walk(batch_dir):
for f in fnames:
full = os.path.join(root, f)
files.append(full)
if logger:
logger.debug("Crash 导出收集：udid=%s, patterns=%s, files=%d, dir=%s", udid, pats, len(files), batch_dir)
return {"ok": ok_all, "batch_dir": batch_dir, "files": files, "raw": "
".join(raws)}
def crash_zip_dir(batch_dir: str, crash_root: str, logger: Optional[logging.Logger] = None) -> Optional[str]:
try:
name = os.path.basename(batch_dir.rstrip(os.sep)) + ".zip"
zip_path = os.path.join(crash_root, name)
with zipfile.ZipFile(zip_path, 'w', compression=zipfile.ZIP_DEFLATED) as z:
for root, _, files in os.walk(batch_dir):
for f in files:
full = os.path.join(root, f)
arc = os.path.relpath(full, batch_dir)
z.write(full, arcname=arc)
return zip_path
except Exception as exc:
if logger:
logger.warning("打包 crash 目录失败: %s", exc)
return None
# === 系统日志会话（写文件） ===
def _drain_syslog_to_file(proc, path: str, logger: Optional[logging.Logger] = None) -> None:
try:
if not proc or not getattr(proc, 'stdout', None):
return
with open(path, 'a', encoding='utf-8', errors='replace') as f:
for line in iter(proc.stdout.readline, ''):
if not line:
break
f.write(line)
except (OSError, IOError) as exc:
if logger:
logger.debug("syslog 写入失败: %s", exc)
except Exception as exc:
if logger:
logger.debug("syslog 写入未知异常: %s", exc)
def syslog_start_session(goios_manager, udid: str, out_dir: str, parse: bool = True,
logger: Optional[logging.Logger] = None) -> Dict[str, Any]:
p = goios_manager.syslog_stream_popen_parsed(udid=udid, parse=parse)
if p is None:
return {"ok": False, "msg": "启动 syslog 失败"}
name = f"{udid}_syslog_{now_timestamp_str()}.log"
path = os.path.join(out_dir, name)
t = threading.Thread(target=_drain_syslog_to_file, args=(p, path, logger), daemon=True, name=f"Syslog-{udid}")
t.start()
return {"ok": True, "p": p, "name": name, "path": path}
# === 设备事件监听（SSE生成器） ===
def listen_event_stream(goios_manager):
p = goios_manager.listen_popen()
try:
if not p or not p.stdout:
yield "event: error
data: failed to start listen
"
return
for line in p.stdout:
if not line:
break
data = line.strip()
yield f"data: {data}
"
finally:
try:
if p:
p.terminate()
except (OSError, AttributeError) as exc:
logging.debug("listen_popen 终止异常: %s", exc)
def parse_crash_ls_items(raw: str) -> List[str]:
items: List[str] = []
raw = raw or ""
# 1) 先尝试整体 JSON
def _push_files(files_in):
for x in files_in or []:
if not isinstance(x, str):
continue
name_in = x.strip()
if not name_in or name_in in (".", ".."):
continue
items.append(name_in)
data = None
try:
data = json.loads(raw)
except json.JSONDecodeError:
pass
if isinstance(data, dict):
files_found = data.get("files") or data.get("list")
if isinstance(files_found, list):
_push_files(files_found)
elif isinstance(data, list) and all(isinstance(x, str) for x in data):
_push_files(data)
# 2) 若未取到，逐行解析（应对多段 JSON / 告警 JSON + 结果 JSON）
if not items:
for line in str(raw).splitlines():
s = line.strip()
if not s:
continue
try:
obj = json.loads(s)
except json.JSONDecodeError:
# 不是 JSON，跳过（避免把整段 JSON 文本当成文件项）
continue
if isinstance(obj, dict):
files_found = obj.get("files") or obj.get("list")
if isinstance(files_found, list):
_push_files(files_found)
# 逐行提取后可能仍为空，则再按纯文本回退
if not items:
for line in str(raw).splitlines():
line_name = line.strip()
if not line_name or line_name in (".", ".."):
continue
# 仅接受看起来像文件名的项（.ips 或 非 JSON 格式）
if line_name.startswith("{") and line_name.endswith("}"):
continue
items.append(line_name)
# 去重
seen = set()
uniq: List[str] = []
for unique_name in items:
if unique_name not in seen:
seen.add(unique_name)
uniq.append(unique_name)
return uniq
def stream_syslog_sse(goios_manager, udid: str, parse: bool = True, logger: Optional[logging.Logger] = None,
keywords: Optional[List[str]] = None,
levels: Optional[List[str]] = None,
existing_process=None):
"""基于 go-ios syslog 的 SSE 生成器。客户端断开后终止进程。"""
# 如果提供了现有进程，使用它；否则创建新的
p = existing_process
if not p:
if logger:
logger.warning("没有提供现有进程，创建新的syslog进程: udid=%s", udid)
p = goios_manager.syslog_stream_popen_parsed(udid=udid, parse=parse)
kw_list = [str(k or "").lower() for k in (keywords or []) if str(k or "").strip()]
lv_set = set([str(l or "").lower() for l in (levels or []) if str(l or "").strip()])
def _bucketize_level(val: str) -> str:
s = str(val or "").strip().lower()
# 数字优先级：0=emerg ... 7=debug
try:
n = int(s)
if n <= 3:
return "error"
if n == 4:
return "warning"
return "info"
except (ValueError, TypeError):
pass
if not s:
return "info"
if "warn" in s:
return "warning"
if s in ("error", "err", "fault", "critical", "crit", "emergency", "emerg", "alert", "fatal"):
return "error"
# 常见信息级别：info/notice/default/debug
return "info"
try:
if not p or not getattr(p, 'stdout', None):
yield "event: error
data: failed to start syslog
"
return
# 立即发送一次注释/心跳，触发浏览器接收响应头并建立 EventSource
yield ": heartbeat
"
for line in p.stdout:
if not line:
break
data = line.rstrip("
")
include = True
if parse:
try:
obj = json.loads(data)
except json.JSONDecodeError:
obj = None
if isinstance(obj, dict):
msg_val = str(obj.get("Message") or obj.get("message") or obj.get("msg") or "")
proc_val = str(
obj.get("Process") or obj.get("process") or obj.get("Sender") or obj.get("Image") or obj.get(
"Program") or "")
raw_level = obj.get("Level") or obj.get("level") or obj.get("Priority") or ""
bucket = _bucketize_level(raw_level)
low_msg = msg_val.lower()
low_proc = proc_val.lower()
if kw_list:
include = any((k in low_msg) or (k in low_proc) for k in kw_list)
if include and lv_set:
include = (bucket in lv_set)
else:
# 解析失败时按原始文本关键字
if kw_list:
include = any(k in data.lower() for k in kw_list)
else:
# 非 parse 模式仅做关键字过滤
if kw_list:
include = any(k in data.lower() for k in kw_list)
if not include:
continue
# 简单限长，避免超长行撑爆前端
if len(data) > 8000:
data = data[:8000] + " …"
yield f"data: {data}
"
finally:
# 只有当进程是我们创建的（不是传入的现有进程）时才终止
if not existing_process and p:
try:
if logger:
logger.debug("终止新创建的syslog进程: udid=%s", udid)
p.terminate()
except (OSError, AttributeError) as exc:
if logger:
logger.debug("syslog_sse 终止异常: %s", exc)7、config代码
# -*- coding: utf-8 -*-
# config.py
import os
class Config:
# Flask
SECRET_KEY = os.environ.get("SECRET_KEY", "dev-secret")
DEBUG = os.environ.get("FLASK_DEBUG", "1") == "1"
HOST = os.environ.get("FLASK_HOST", "0.0.0.0")
PORT = int(os.environ.get("FLASK_PORT", "5000"))
# 路径
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
PROJECT_ROOT = os.path.abspath(os.path.join(BASE_DIR, '..'))
# 统一到项目根目录
IOS_HOME = os.path.join(PROJECT_ROOT, 'IOSPrechecker')
UPLOAD_FOLDER = os.path.join(PROJECT_ROOT, "ios_downloads")
LOG_DIR = os.path.join(PROJECT_ROOT, "ios_logs")
# go-ios 资源与镜像统一归档
GOIOS_DIR = os.path.join(IOS_HOME, 'utils') # 放置 go-ios 压缩包 (go-ios-*.zip)
GOIOS_EXECUTABLE_DIR = os.path.join(IOS_HOME, 'executable') # 解压后的可执行文件存放处（避免与wintun/bin冲突）
DEVIMAGES_DIR = os.environ.get('DEVIMAGES_DIR', os.path.join(IOS_HOME, 'devimages'))
# 可选：若你想使用系统里已安装的 ios 可执行文件，直接设置此环境变量即可覆盖
GOIOS_BIN_PATH = os.environ.get("GOIOS_BIN_PATH", "")
# Flask 上传文件大小限制
MAX_CONTENT_LENGTH = 512 * 1024 * 1024 # 512MB
# 可选：HTTPS 证书与私钥（若同时提供则启用 HTTPS）
SSL_CERT_FILE = os.environ.get("SSL_CERT_FILE", "")
SSL_KEY_FILE = os.environ.get("SSL_KEY_FILE", "")8、goios_wrapper代码
import os
import json
import stat
import zipfile
import logging
import platform
import threading
import subprocess
from pathlib import Path
from typing import List, Optional, Tuple, Dict, Any
logger = logging.getLogger(__name__)
class GoIOSManager:
"""
负责：
1) 按操作系统自动解压 go-ios 压缩包到 bin 目录；
2) 找到可执行文件（可能叫 ios 或 ios.exe）；
3) 提供统一的 run() 封装与常用功能方法（list/info/screenshot/apps/install/launch/syslog 等）；
4) 多设备并发安全：对同一 UDID 的命令串行化执行。
"""
def __init__(self, goios_root: str, bin_dir: str, bin_path_override: str = ""):
self.goios_root = goios_root
self.bin_dir = bin_dir
self.bin_path_override = bin_path_override
self._ensure_dirs()
self.ios_bin = self._ensure_ios_binary()
self._locks: Dict[str, threading.Lock] = {}
# ---------------- 基础能力：目录、可执行文件 ----------------
def _ensure_dirs(self) -> None:
for d in (self.goios_root, self.bin_dir):
os.makedirs(d, exist_ok=True)
@staticmethod
def _zip_map_for_os() -> Tuple[str, str]:
"""
根据宿主系统返回 (zip 文件名, 解压到的子目录名)
"""
sys_name = platform.system().lower()
if sys_name.startswith("win"):
return "go-ios-win.zip", "win"
if sys_name == "darwin":
return "go-ios-mac.zip", "mac"
return "go-ios-linux.zip", "linux"
@staticmethod
def _arch_preferred_names() -> list:
"""
根据宿主 CPU 架构给出优先候选名（Linux 包里常见：ios-amd64 / ios-arm64）。
同时包含跨平台常见命名：ios / ios.exe / go-ios / go-ios.exe。
"""
m = (platform.machine() or "").lower()
preferred = []
if any(x in m for x in ("aarch64", "arm64")):
preferred += ["ios-arm64", "ios"] # Apple Silicon / ARM Linux
elif any(x in m for x in ("x86_64", "amd64", "x64")):
preferred += ["ios-amd64", "ios"] # x86_64 Linux
else:
preferred += ["ios"]
# 跨平台常见别名一并加入（用于 Mac/Win 或历史包）
preferred += ["go-ios", "ios.exe", "go-ios.exe", "ios-arm64.exe", "ios-amd64.exe"]
# 去重但保留顺序
seen = set()
ordered = []
for n in preferred:
if n not in seen:
seen.add(n)
ordered.append(n)
return ordered
@staticmethod
def _find_executable(search_dir: str) -> Optional[str]:
"""
在 search_dir 下递归寻找 go-ios 可执行文件。
兼容：
- Linux: ios-amd64 / ios-arm64 （无扩展名）
- macOS: ios
- Windows: ios.exe
- 旧名： go-ios / go-ios.exe
优先选择与宿主架构匹配的名称。
"""
# 先收集所有文件 -> 路径
file_map: Dict[str, str] = {}
for root, _, files in os.walk(search_dir):
for f in files:
file_map[f] = os.path.join(root, f)
# 按优先顺序挑选
for name in GoIOSManager._arch_preferred_names():
if name in file_map:
path = file_map[name]
logger.debug("发现 go-ios 可执行文件: %s", path)
return path
# 兜底：找名为 'ios'（无扩展）的文件
if "ios" in file_map:
path = file_map["ios"]
logger.debug("发现 go-ios 可执行文件（fallback）: %s", path)
return path
return None
@staticmethod
def _ensure_executable_perm(file_path: str) -> None:
"""
为 *nix 系统赋予执行权限；Windows 忽略
"""
try:
if os.name != "nt":
st_mode = os.stat(file_path).st_mode
os.chmod(file_path, st_mode | stat.S_IEXEC)
except OSError as exc:
logger.warning("为 %s 赋予执行权限失败：%s", file_path, exc)
def _ensure_ios_binary(self) -> str:
"""
确保 go-ios 可执行文件可用：优先使用覆盖路径，否则从 zip 解压/复用已解压文件。
"""
# 优先使用外部覆盖路径（如果指定）
if self.bin_path_override:
if not os.path.exists(self.bin_path_override):
raise FileNotFoundError(f"GOIOS_BIN_PATH 不存在: {self.bin_path_override}")
logger.debug("Using go-ios binary (override): %s", self.bin_path_override)
return self.bin_path_override
zip_name, os_dir = self._zip_map_for_os()
target_dir = os.path.join(self.bin_dir, os_dir)
os.makedirs(target_dir, exist_ok=True)
# 若目录中已存在可执行文件，直接使用
existing = self._find_executable(target_dir)
if existing:
self._ensure_executable_perm(existing)
return existing
# 否则尝试从 zip 解压
zip_path = os.path.join(self.goios_root, zip_name)
if not os.path.exists(zip_path):
raise FileNotFoundError(f"未找到 go-ios 压缩包: {zip_path}")
logger.debug("首次使用：正在解压 %s 到 %s ...", zip_path, target_dir)
try:
with zipfile.ZipFile(zip_path, "r") as zf:
zf.extractall(target_dir)
except zipfile.BadZipFile as exc:
logger.exception("go-ios 压缩包损坏：%s", exc)
raise
exe = self._find_executable(target_dir)
if not exe:
raise FileNotFoundError(
f"解压后仍未找到 ios 可执行文件，请检查压缩包内容: {zip_path}"
)
self._ensure_executable_perm(exe)
logger.debug("Using go-ios binary at: %s", exe)
return exe
# ---------------- 命令拼接与执行 ----------------
def _lock_for(self, udid: Optional[str]) -> threading.Lock:
"""
返回某 UDID 对应的锁；未指定 UDID 的命令共享“匿名锁”
"""
key = udid or "_default_"
if key not in self._locks:
self._locks[key] = threading.Lock()
return self._locks[key]
@staticmethod
def _build_common_opts(
udid: Optional[str] = None,
address: Optional[str] = None,
rsd_port: Optional[int] = None,
userspace_port: Optional[int] = None,
proxyurl: Optional[str] = None,
tunnel_info_port: Optional[int] = None,
verbose: bool = False,
trace: bool = False,
nojson: bool = False,
pretty: bool = False,
) -> List[str]:
"""
将 go-ios 全局 options（--udid 等）转成参数列表
"""
opts: List[str] = []
if verbose:
opts.append("-v")
if trace:
opts.append("--trace")
if nojson:
opts.append("--nojson")
if pretty:
opts.append("--pretty")
if udid:
opts += ["--udid", udid]
if address:
opts += ["--address", address]
if rsd_port:
opts += ["--rsd-port", str(rsd_port)]
if userspace_port:
opts += ["--userspace-port", str(userspace_port)]
if proxyurl:
opts += ["--proxyurl", proxyurl]
if tunnel_info_port:
opts += ["--tunnel-info-port", str(tunnel_info_port)]
return opts
def run(
self,
args: List[str],
timeout: int = 120,
**opts: Any,
) -> Tuple[int, str, str]:
"""
通用执行器：支持 go-iOS 全局 options（--udid/--address/...）。
所有“针对具体设备”的方法都应把 udid 传进来（用于加锁与选中目标设备）。
返回 (returncode, stdout, stderr)；不抛异常，让上层决定如何提示。
"""
# 允许调用方注入额外环境变量（例如 ENABLE_GO_IOS_AGENT）。
# 注意：需先从 opts 中弹出，避免传入 _build_common_opts。
extra_env_raw = opts.get("extra_env")
if isinstance(extra_env_raw, dict):
extra_env: Dict[str, str] = extra_env_raw
del opts["extra_env"]
else:
extra_env = {}
common = self._build_common_opts(**opts)
cmd = [self.ios_bin] + common + args
udid = opts.get("udid") # 仅用于加锁
env = os.environ.copy()
env.update({k: str(v) for k, v in extra_env.items()})
logger.debug("[GoIOSManager] 执行命令: %s", " ".join(cmd))
lock = self._lock_for(udid)
with lock:
try:
cp = subprocess.run(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
timeout=timeout,
check=False,
encoding="utf-8",
errors="replace",
env=env,
creationflags=(subprocess.CREATE_NO_WINDOW if os.name == 'nt' else 0),
)
if cp.returncode != 0:
logger.warning(
"命令失败（返回码 %s）：stderr=%s",
cp.returncode,
(cp.stderr or "").strip(),
)
return cp.returncode, cp.stdout, cp.stderr
except subprocess.TimeoutExpired:
logger.error("命令超时：%s", " ".join(cmd))
return 124, "", "命令执行超时"
except (FileNotFoundError, OSError, ValueError) as exc:
# FileNotFoundError: 可执行文件不存在；OSError/ValueError: 参数或环境错误
logger.exception("命令执行异常：%s", exc)
return 125, "", f"命令执行异常: {exc}"
# ---------------- 常用操作（全部支持 **opts 透传） ----------------
def device_pair(
self,
udid: Optional[str] = None,
p12file: Optional[str] = None,
password: Optional[str] = None,
**opts: Any,
) -> Tuple[bool, str]:
"""
go-ios 配对：
- 普通设备：直接 `ios pair`
- 受监督设备：传 p12 与密码可静默配对
"""
args = ["pair"]
if p12file:
path = Path(p12file)
try:
# 仅做“存在且是文件”的健壮性检查，避免把路径错误带到 go-ios
if not path.is_file():
return False, f"p12 文件不存在: {path}"
except (OSError, PermissionError) as exc:
# 只捕获文件系统相关异常；记录后继续让 go-ios 尝试
logger.warning("检查 p12 文件时出错（%s）：%s，将继续尝试 go-ios。",
type(exc).__name__, exc)
args += ["--p12file", str(path)]
if password:
# 注意：不要把密码写入日志
args += ["--password", password]
code, out, err = self.run(args, udid=udid, timeout=120, **opts)
return code == 0, (out.strip() if (out or "").strip() else err)
def list_devices(self, details: bool = False, **opts: Any) -> Tuple[bool, str]:
args = ["list"]
if details:
args.append("--details")
code, out, err = self.run(args, **opts)
return code == 0, out if out.strip() else err
def device_info(self, udid: str, **opts: Any) -> Tuple[bool, str]:
code, out, err = self.run(["info"], udid=udid, **opts)
out_str = out.strip() or err
try:
data = json.loads(out_str)
return code == 0, json.dumps(data, ensure_ascii=False, indent=2)
except json.JSONDecodeError:
return code == 0, out_str
def screenshot(self, udid: str, save_path: str, **opts: Any) -> Tuple[bool, str]:
"""
截图并保存到指定路径，自动确保目录存在。
"""
try:
os.makedirs(os.path.dirname(save_path), exist_ok=True)
except Exception as e:
logger.warning("创建截图目录失败: %s", e)
args = ["screenshot", f"--output={save_path}"]
code, out, err = self.run(args, udid=udid, timeout=180, **opts)
if code == 0 and os.path.exists(save_path):
return True, save_path
return False, (out.strip() or err or "截屏失败")
def apps_list(self, udid: str, only_list: bool = True, **opts: Any) -> Tuple[bool, str]:
# ios apps [--list]
args = ["apps"]
if only_list:
args.append("--list")
code, out, err = self.run(args, udid=udid, **opts)
return code == 0, out if out.strip() else err
def install_ipa(self, udid: str, ipa_path: str, **opts: Any) -> Tuple[bool, str]:
if not os.path.exists(ipa_path):
return False, f"IPA 文件不存在: {ipa_path}"
code, out, err = self.run(["install", "--path", ipa_path], udid=udid, timeout=1800, **opts)
return code == 0, out if out.strip() else err
def launch_app(self, udid: str, bundle_id: str, wait: bool = False, **opts: Any) -> Tuple[bool, str]:
# ios launch <bundleID> [--wait]
args = ["launch", bundle_id]
if wait:
args.append("--wait")
code, out, err = self.run(args, udid=udid, timeout=300, **opts)
return code == 0, out if out.strip() else err
def kill_app(self, udid: str, bundle_id: str, **opts: Any) -> Tuple[bool, str]:
"""
更稳健的停止逻辑：
先尝试按文档直接使用 bundleID；若失败，再回退到 ps/apps 映射进程名方案。
文档参考：ios kill (<bundleID> | --pid | --process=<processName>)
"""
# 0. 首先直接按 bundleID 尝试（与官方文档一致）
code0, out0, err0 = self.run(["kill", bundle_id], udid=udid, timeout=60, **opts)
if code0 == 0:
return True, (out0.strip() or "停止应用成功")
# 继续回退到进程名匹配方案
"""
1) 先 ios ps --apps (JSON) 获取进程列表；
2) 直接用 RealAppName/Name 匹配；
3) 若未匹配到，则 ios apps 获取 CFBundleExecutable/Name，映射到进程 Name 或路径；
4) 命中后使用 --process 结束。
"""
# 1. ps 检查（JSON）
code, out, err = self.run(["ps", "--apps"], udid=udid, timeout=60, **opts)
if code != 0:
return False, (out.strip() or err or "无法获取进程列表")
try:
processes = json.loads(out)
if not isinstance(processes, list):
return False, f"进程列表格式异常: {out or err}"
except json.JSONDecodeError:
return False, f"进程列表解析失败: {out or err}"
def pick_target(proc_list, bundle: str, exec_names: list[str]) -> Optional[Dict[str, Any]]:
b = (bundle or "").strip()
for p in proc_list:
real = str(p.get("RealAppName") or "")
name = str(p.get("Name") or "")
if b and b in real:
return p
for en in exec_names:
if not en:
continue
if name == en:
return p
if real.endswith(f"/{en}") or f"/{en}.app/{en}" in real:
return p
return None
# 2. 直接尝试用 bundle_id 在 RealAppName 中匹配
target = pick_target(processes, bundle_id, exec_names=[])
# 3. 若失败，从 apps 列表推断可执行名再匹配
if not target:
code_apps, out_apps, err_apps = self.run(["apps"], udid=udid, timeout=90, **opts)
exec_candidates: list[str] = []
if code_apps == 0 and out_apps:
try:
data = json.loads(out_apps)
except json.JSONDecodeError:
data = None
# data 可以是 list 或包含 apps 的 dict
app_list = []
if isinstance(data, list):
app_list = data
elif isinstance(data, dict):
maybe = data.get("apps") or data.get("Apps") or data.get("applications") or data.get("list")
if isinstance(maybe, list):
app_list = maybe
# 提取可执行名
for obj in app_list:
if not isinstance(obj, dict):
continue
bid = str(obj.get("CFBundleIdentifier") or obj.get("bundleID") or obj.get("bundleId") or obj.get(
"Bundle") or obj.get("id") or "")
if bid == bundle_id:
exec_candidates.append(str(obj.get("CFBundleExecutable") or "").strip())
# 兜底：名称也作为候选
name_candidate = str(obj.get("CFBundleDisplayName") or obj.get("CFBundleName") or obj.get(
"BundleName") or obj.get("name") or obj.get("Name") or "").strip()
if name_candidate:
exec_candidates.append(name_candidate)
break
# 去重并清理
exec_candidates = [e for e in {e for e in exec_candidates if e}]
target = pick_target(processes, bundle_id, exec_candidates)
if not target:
return False, f"{bundle_id} 未运行"
process_name = target.get("Name")
if not process_name:
return False, f"未找到 {bundle_id} 对应的进程名"
# 4. kill by process name
code, out, err = self.run(["kill", "--process", process_name], udid=udid, timeout=60, **opts)
if code != 0:
return False, (out.strip() or err or "停止应用失败")
return True, out.strip() or "停止应用成功"
def reboot(self, udid: str, **opts: Any) -> Tuple[bool, str]:
code, out, err = self.run(["reboot"], udid=udid, timeout=60, **opts)
return code == 0, out if out.strip() else err
def battery_info(self, udid: str, **opts: Any) -> Tuple[bool, str]:
code, out, err = self.run(["batterycheck"], udid=udid, timeout=30, **opts)
return code == 0, out if out.strip() else err
def battery_registry(self, udid: str, **opts: Any) -> Tuple[bool, str]:
code, out, err = self.run(["batteryregistry"], udid=udid, timeout=30, **opts)
return code == 0, out if out.strip() else err
def diskspace(self, udid: str, **opts: Any) -> Tuple[bool, str]:
code, out, err = self.run(["diskspace"], udid=udid, timeout=30, **opts)
return code == 0, out if out.strip() else err
def image_auto(self, udid: str, basedir: Optional[str] = None, **opts: Any) -> Tuple[bool, str]:
args = ["image", "auto"]
if basedir:
try:
os.makedirs(basedir, exist_ok=True)
except Exception as e:
logger.warning("创建 image_auto basedir 失败: %s", e)
args.append(f"--basedir={basedir}")
code, out, err = self.run(args, udid=udid, timeout=300, **opts)
output = out if out.strip() else err
if "error" in output.lower() or "failed" in output.lower():
return False, output
return code == 0, output
def image_mount(self, udid: str, path: str, **opts: Any) -> Tuple[bool, str]:
if not os.path.exists(path):
return False, f"镜像文件不存在: {path}"
args = ["image", "mount", f"--path={path}"]
code, out, err = self.run(args, udid=udid, timeout=300, **opts)
return code == 0, out if out.strip() else err
def image_list(self, udid: Optional[str] = None, **opts: Any) -> Tuple[bool, str]:
# ios image list
args = ["image", "list"]
code, out, err = self.run(args, udid=udid, timeout=30, **opts)
return code == 0, out if out.strip() else err
def devicestate_list(self, udid: Optional[str] = None, **opts: Any) -> Tuple[bool, str]:
# ios devicestate list
code, out, err = self.run(["devicestate", "list"], udid=udid, timeout=30, **opts)
return code == 0, out if out.strip() else err
def devicestate_enable(self, udid: str, profile_type_id: str, profile_id: str, **opts: Any) -> Tuple[bool, str]:
# ios devicestate enable <profileTypeId> <profileId>
code, out, err = self.run(
["devicestate", "enable", profile_type_id, profile_id],
udid=udid,
timeout=30,
**opts,
)
return code == 0, out if out.strip() else err
def set_location(self, udid: str, lat: float, lon: float, **opts: Any) -> Tuple[bool, str]:
# ios setlocation --lat=.. --lon=..
args = ["setlocation", f"--lat={lat}", f"--lon={lon}"]
code, out, err = self.run(args, udid=udid, timeout=30, **opts)
return code == 0, out if out.strip() else err
# ---------------- 长进程（Popen）：日志/流/转发 ----------------
def syslog_stream_popen(self, udid: str, **opts: Any):
"""
启动 syslog 流；返回 Popen。调用方需在断连时自行终止进程。
"""
cmd = [self.ios_bin] + self._build_common_opts(udid=udid, **opts) + ["syslog"]
try:
return subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
bufsize=1,
universal_newlines=True,
encoding="utf-8",
errors="replace",
creationflags=(
subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_PROCESS_GROUP) if os.name == 'nt' else 0,
)
except (OSError, ValueError) as exc:
logger.exception("启动 syslog 失败：%s", exc)
return None
def screenshot_stream_popen(self, udid: str, port: int = 3333,
save_dir: Optional[str] = None, **opts: Any):
"""
启动 MJPEG 截屏流到 0.0.0.0:<port>；返回 Popen。
save_dir 用于保证临时录屏目录存在。
"""
# ✅ 确保保存目录存在
if save_dir:
try:
os.makedirs(save_dir, exist_ok=True)
logger.debug("确保录屏目录存在: %s", save_dir)
except Exception as e:
logger.warning("创建录屏目录失败: %s", e)
# 提取环境变量，避免传递给 _build_common_opts
extra_env = opts.pop("extra_env", {})
cmd = (
[self.ios_bin]
+ self._build_common_opts(udid=udid, **opts)
+ ["screenshot", "--stream", "--port", str(port)]
)
env = os.environ.copy()
env.update({k: str(v) for k, v in extra_env.items()})
try:
return subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
encoding="utf-8",
errors="replace",
env=env,
bufsize=0,
creationflags=(subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_PROCESS_GROUP)
if os.name == 'nt' else 0,
)
except (OSError, ValueError) as exc:
logger.exception("启动 screenshot --stream 失败：%s", exc)
return None
def forward_popen(self, udid: str, host_port: int, target_port: int, **opts: Any):
"""
启动端口转发（host_port -> target_port）；返回 Popen。
"""
cmd = (
[self.ios_bin]
+ self._build_common_opts(udid=udid, **opts)
+ ["forward", str(host_port), str(target_port)]
)
try:
return subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True,
encoding="utf-8",
errors="replace",
)
except (OSError, ValueError) as exc:
logger.exception("启动 forward 失败：%s", exc)
return None
def crash_ls(self, udid: str, pattern: Optional[str] = None, **opts: Any) -> Tuple[bool, str]:
"""
列出设备上的 crash 文件。
pattern 可选，用于按模式过滤。
"""
args = ["crash", "ls"]
if pattern:
args.append(pattern)
code, out, err = self.run(args, udid=udid, timeout=120, **opts)
return code == 0, out if out.strip() else err
def crash_cp(self, udid: str, srcpattern: str, target_dir: str, **opts: Any) -> Tuple[bool, str]:
try:
os.makedirs(target_dir, exist_ok=True)
except Exception as e:
logger.warning("创建 crash_cp 目标目录失败: %s", e)
code, out, err = self.run(["crash", "cp", srcpattern, target_dir], udid=udid, timeout=300, **opts)
return code == 0, out if out.strip() else err
def crash_rm(self, udid: str, cwd: str, pattern: str, recursive: bool = False, **opts: Any) -> Tuple[bool, str]:
candidates: List[List[str]] = []
# 根据 go-ios 命令格式：ios crash rm <cwd> <pattern> [options]
# 递归参数应该作为选项，不是位置参数
# 基础命令：ios crash rm <cwd> <pattern>
base_cmd = ["crash", "rm", cwd, pattern]
candidates.append(base_cmd)
# 如果指定了递归，添加递归选项
if recursive:
# 尝试不同的递归选项格式
recursive_cmds = [
["crash", "rm", "-r", cwd, pattern], # -r 选项
["crash", "rm", "--r", cwd, pattern], # --r 选项
["crash", "rm", "--recursive", cwd, pattern] # --recursive 选项
]
candidates.extend(recursive_cmds)
# 记录尝试的命令用于调试
logger.debug("crash_rm 将尝试以下命令: %s", [candidates])
last_msg = ""
for args in candidates:
try:
logger.debug("尝试 crash_rm 命令: %s", " ".join(args))
code, out, err = self.run(args, udid=udid, timeout=120, **opts)
msg = out if (out and out.strip()) else (err or "")
if code == 0:
logger.debug("crash_rm 命令成功: %s", " ".join(args))
return True, msg
logger.debug("crash_rm 命令失败 (code=%d): %s", code, msg)
last_msg = msg
except (OSError, ValueError) as exc:
last_msg = str(exc)
logger.debug("crash_rm 命令异常: %s", exc)
continue
logger.warning("所有 crash_rm 命令尝试都失败了")
return False, last_msg
def devmode_get(self, udid: str, **opts: Any) -> Tuple[bool, str]:
code, out, err = self.run(["devmode", "get"], udid=udid, timeout=60, **opts)
return code == 0, out if out.strip() else err
def devmode_enable(self, udid: str, enable_post_restart: bool = True, **opts: Any) -> Tuple[bool, str]:
args = ["devmode", "enable"]
if enable_post_restart:
args.append("--enable-post-restart")
code, out, err = self.run(args, udid=udid, timeout=120, **opts)
return code == 0, out if out.strip() else err
def profile_list(self, udid: str, **opts: Any) -> Tuple[bool, str]:
code, out, err = self.run(["profile", "list"], udid=udid, timeout=60, **opts)
return code == 0, out if out.strip() else err
def profile_remove(self, udid: str, profile_name: str, **opts: Any) -> Tuple[bool, str]:
code, out, err = self.run(["profile", "remove", profile_name], udid=udid, timeout=60, **opts)
return code == 0, out if out.strip() else err
def ps_apps(self, udid: str, **opts: Any) -> Tuple[bool, str]:
code, out, err = self.run(["ps", "--apps"], udid=udid, timeout=60, **opts)
return code == 0, out if out.strip() else err
def syslog_stream_popen_parsed(self, udid: str, parse: bool = True, **opts: Any):
cmd = [self.ios_bin] + self._build_common_opts(udid=udid, **opts) + ["syslog"]
if parse:
cmd.append("--parse")
try:
return subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True,
encoding="utf-8",
errors="replace",
creationflags=(
subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_PROCESS_GROUP) if os.name == 'nt' else 0,
)
except (OSError, ValueError) as exc:
logger.exception("启动 syslog 失败：%s", exc)
return None
def listen_popen(self, **opts: Any):
cmd = [self.ios_bin] + self._build_common_opts(**opts) + ["listen"]
try:
return subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True,
encoding="utf-8",
errors="replace",
creationflags=(
subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_PROCESS_GROUP) if os.name == 'nt' else 0,
)
except (OSError, ValueError) as exc:
logger.exception("启动 listen 失败：%s", exc)
return None
def assistive(self, udid: str, feature: str, action: str, force: bool = False, **opts: Any) -> Tuple[bool, str]:
# feature: assistivetouch | voiceover | zoom ; action: enable|disable|toggle|get
args = [feature, action]
if force:
args.append("--force")
code, out, err = self.run(args, udid=udid, timeout=60, **opts)
return code == 0, out if out.strip() else err8、ios_prechecker代码
import time
import logging
import threading
from typing import Tuple
try:
from .tunnel_manager import TunnelManager
from .goios_wrapper import GoIOSManager
from .common_utils import find_free_port
from .config import Config
except ImportError:
# 当作为独立模块运行时，使用绝对导入
from tunnel_manager import TunnelManager
from goios_wrapper import GoIOSManager
from common_utils import find_free_port
from config import Config
logger = logging.getLogger(__name__)
class IOSPrechecker:
"""
iOS 操作前的统一检查器：
1. UDID 是否存在
2. go-ios tunnel 是否已启动（iOS17+ 必需）
3. 是否挂载了 Developer Image
"""
def __init__(self, manager: GoIOSManager, tunnel: TunnelManager):
self.m = manager
self.tunnel = tunnel
self._tunnel_lock = threading.Lock()
self._tunnel_start_time = 0
def check_all(self, udid: str, skip_tunnel_check: bool = False) -> Tuple[bool, str]:
"""
执行全量检查
:param udid: 设备UDID
:param skip_tunnel_check: 是否跳过tunnel检查（用于某些不需要tunnel的操作）
返回: (ok, msg)
"""
# 1. 检查 UDID 是否存在
ok, out = self.m.list_devices(details=False)
if not ok:
return False, f"无法获取设备列表，请检查:
1. iOS设备是否已连接
2. 是否已信任此电脑
3. go-ios是否正确安装"
if not out or udid not in out:
return False, f"设备 {udid} 未连接或未被识别"
# 2. 检查 tunnel 状态（可选跳过）
if not skip_tunnel_check:
ok, tunnel_msg = self._ensure_tunnel_running()
if not ok:
return False, tunnel_msg
# 3. 检查是否挂载开发者镜像（对于某些操作是必需的）
try:
# 先获取隧道参数再挂载镜像
extra_opts = self.tunnel.get_goios_opts(udid) if hasattr(self.tunnel, 'get_goios_opts') else {}
ok, out = self.m.image_auto(udid, basedir=Config.DEVIMAGES_DIR, extra_env={"ENABLE_GO_IOS_AGENT": "user"}, **extra_opts)
if not ok:
logger.warning("开发者镜像挂载失败，某些功能可能受限: %s", out)
# 不作为致命错误，某些操作不需要开发者镜像
except Exception as exc:
logger.warning("检查开发者镜像时异常: %s", exc)
return True, "设备检查通过"
def _ensure_tunnel_running(self) -> Tuple[bool, str]:
"""
确保tunnel正在运行，包含智能重试逻辑
"""
with self._tunnel_lock:
# 首先检查tunnel状态
is_running, status_msg = self.tunnel.status()
if is_running:
logger.info("Tunnel 已在运行")
return True, "Tunnel 正常运行"
# 如果最近刚尝试启动过，避免频繁重试
current_time = time.time()
if current_time - self._tunnel_start_time < 60: # 60秒内不重复启动
return False, "Tunnel 启动中或最近启动失败，请稍后重试"
logger.warning("Tunnel 未运行，尝试启动...")
self._tunnel_start_time = current_time
# 寻找可用端口
port = find_free_port(start_port=60105, max_tries=20)
if not port:
return False, "无可用端口，无法启动 tunnel"
# 启动tunnel
ok, msg = self.tunnel.start(userspace=True, tunnel_info_port=port, retry_count=2)
if not ok:
return False, f"Tunnel 启动失败: {msg}
可能的解决方案:
1. 重新连接设备
2. 重启应用
3. 检查设备是否为iOS17+"
logger.info("Tunnel 启动成功，监听端口 %s", port)
return True, f"Tunnel 启动成功 (端口: {port})"
def check_device_only(self, udid: str) -> Tuple[bool, str]:
"""
仅检查设备连接状态，不检查tunnel
"""
return self.check_all(udid, skip_tunnel_check=True)
def quick_check(self, udid: str) -> Tuple[bool, str]:
"""
快速检查：只验证设备连接，不启动tunnel
"""
ok, out = self.m.list_devices(details=False)
if not ok or not out:
return False, "无法获取设备列表"
if udid not in out:
return False, f"设备 {udid} 未连接"
return True, "设备连接正常"9、tunnel_manager代码
import os
import json
import time
import logging
import platform
from typing import Tuple, Dict, Any, Optional
try:
from .goios_wrapper import GoIOSManager
from .config import Config
except ImportError:
# 当作为独立模块运行时，使用绝对导入
from goios_wrapper import GoIOSManager
from config import Config
logger = logging.getLogger(__name__)
class TunnelManager:
"""
管理 go-ios tunnel（隧道）的启动、停止和状态查询。
主要用于 iOS 17+ 设备的通信。
"""
def __init__(self, goios_manager: GoIOSManager):
self.goios = goios_manager
self.info_port: Optional[int] = None
self._device_opts: Dict[str, Any] = {}
@staticmethod
def _check_system_requirements() -> Tuple[bool, str]:
"""检查系统要求"""
system = platform.system().lower()
if system == "windows":
# 检查wintun.dll
wintun_path = os.path.join(os.environ.get('SystemRoot', 'C:\Windows'), 'system32', 'wintun.dll')
if not os.path.exists(wintun_path):
return False, (
"Windows系统缺少wintun.dll依赖
"
"tunnel需要wintun.dll才能正常工作
"
"请检查文件是否存在: C:/Windows/system32/wintun.dll"
)
# 检查管理员权限（可选，因为可以使用userspace模式）
try:
import ctypes
is_admin = ctypes.windll.shell32.IsUserAnAdmin()
if not is_admin:
logger.debug("未检测到管理员权限，将使用userspace模式")
except (ImportError, AttributeError, OSError):
logger.warning("无法检查Windows管理员权限")
elif system in ["linux", "darwin"]:
# 检查是否为root或sudo权限
if os.geteuid() != 0:
return False, (
f"{system.title()}系统需要root权限启动tunnel
"
"请使用 sudo 权限运行应用"
)
return True, "系统要求检查通过"
@staticmethod
def _get_windows_arch() -> str:
"""获取Windows系统架构"""
arch = platform.machine().lower()
# 映射架构名称
arch_mapping = {
'amd64': 'amd64',
'x86_64': 'amd64',
'x86': 'x86',
'i386': 'x86',
'i686': 'x86',
'arm64': 'arm64',
'aarch64': 'arm64',
'arm': 'arm'
}
return arch_mapping.get(arch, 'amd64') # 默认amd64
def _auto_install_wintun(self) -> Tuple[bool, str]:
"""自动安装 wintun.dll 到系统目录"""
try:
arch = self._get_windows_arch()
logger.debug("开始安装 wintun.dll (架构: %s)", arch)
# 构建源文件路径（新结构：IOSPrechecker/wintun/{arch}/wintun.dll）
wintun_source = os.path.join(Config.IOS_HOME, 'wintun', arch, 'wintun.dll')
wintun_target = os.path.join(os.environ.get('SystemRoot', 'C:\Windows'), 'system32', 'wintun.dll')
if not os.path.exists(wintun_source):
return False, (
f"未找到适合当前架构({arch})的 wintun.dll
"
f"源文件路径: {wintun_source}
"
"请检查 IOSPrechecker/wintun/{amd64|x86|arm|arm64} 目录结构"
)
# 检查管理员权限
import ctypes
if not ctypes.windll.shell32.IsUserAnAdmin():
return False, (
"需要管理员权限才能安装 wintun.dll
"
"请以管理员身份重新启动应用"
)
# 复制文件
import shutil
shutil.copy2(wintun_source, wintun_target)
# 验证复制成功
if os.path.exists(wintun_target):
logger.debug("成功复制 wintun.dll: %s -> %s", wintun_source, wintun_target)
return True, f"wintun.dll 已自动安装 (架构: {arch})"
else:
return False, "wintun.dll 复制失败，目标文件不存在"
except ImportError as e:
return False, f"缺少必要模块: {e}"
except PermissionError:
return False, (
"权限不足，无法复制到系统目录
"
"请确保以管理员权限运行应用"
)
except Exception as e:
logger.exception("自动安装 wintun.dll 失败")
return False, f"安装 wintun.dll 时出错: {e}"
def start(self, userspace: bool = True, tunnel_info_port: Optional[int] = None, retry_count: int = 3) -> Tuple[
bool, str]:
"""
启动 tunnel，支持重试机制
:param userspace: 是否使用 --userspace 模式（推荐Windows使用）
:param tunnel_info_port: 可选，指定 tunnel_info_port
:param retry_count: 重试次数
"""
# 检查系统要求
system = platform.system().lower()
if not userspace and system != "windows":
# 非用户态模式需要检查权限
ok, msg = TunnelManager._check_system_requirements()
if not ok:
logger.warning("系统要求检查失败: %s", msg)
# 不直接返回失败，而是强制使用用户态模式
logger.debug("自动切换到用户态模式(--userspace)")
userspace = True
cmd = ["tunnel", "start"]
if userspace:
cmd.append("--userspace")
if tunnel_info_port:
cmd.append(f"--tunnel-info-port={tunnel_info_port}")
logger.debug("TunnelManager.start(): cmd=%s (系统: %s)", " ".join(cmd), system)
for attempt in range(retry_count):
logger.debug("尝试启动tunnel (第%d次)...", attempt + 1)
# 先检查是否已经在运行
if attempt > 0:
is_running, status_msg = self.status(_tunnel_info_port=tunnel_info_port)
if is_running:
logger.debug("Tunnel 已在运行: %s", status_msg)
# 若调用方传入端口，记住它；否则保留原值
if tunnel_info_port:
self.info_port = tunnel_info_port
return True, "Tunnel 已启动"
# 增加超时时间，Windows下tunnel启动可能比较慢
timeout = 60 if attempt == 0 else 90
# 在 Windows 用户态模式下注入 agent 变量以提升稳定性
extra_env = {"ENABLE_GO_IOS_AGENT": "user"} if userspace else {}
code, out, err = self.goios.run(cmd, timeout=timeout, extra_env=extra_env)
if code == 0:
logger.debug("Tunnel 启动命令返回成功，等待就绪...")
# 启动后等待就绪：轮询 ls 最多 10 秒（仅打印首尾日志）
ready = False
for i in range(10):
ok, _msg = self.status(_tunnel_info_port=tunnel_info_port)
if ok:
ready = True
break
time.sleep(1)
if ready:
logger.debug("Go-iOS Agent is ready")
# 记录端口（若未传入，则保留现有值或 None）
if tunnel_info_port:
self.info_port = tunnel_info_port
return True, "Tunnel 启动成功"
logger.warning("Tunnel 启动命令成功但状态检查失败，继续重试...")
else:
error_msg = err.strip() or out.strip() or "未知错误"
logger.warning("Tunnel 启动失败 (第%d次): %s", attempt + 1, error_msg)
# 如果是端口冲突，尝试不同端口
if "address already in use" in error_msg.lower() or "端口" in error_msg:
if tunnel_info_port:
tunnel_info_port += 1
cmd = ["tunnel", "start"]
if userspace:
cmd.append("--userspace")
cmd.append(f"--tunnel-info-port={tunnel_info_port}")
logger.debug("端口冲突，尝试新端口: %d", tunnel_info_port)
if attempt < retry_count - 1:
time.sleep(3) # 重试前等待
return False, f"Tunnel 启动失败，已重试{retry_count}次"
def _cache_from_ls_raw(self, raw: str) -> None:
try:
data = json.loads(raw)
except json.JSONDecodeError:
return
items = []
if isinstance(data, list):
items = data
elif isinstance(data, dict):
maybe = data.get("tunnels") or data.get("list") or []
if isinstance(maybe, list):
items = maybe
for obj in items:
if not isinstance(obj, dict):
continue
# 兼容不同大小写/命名
udid = obj.get("udid") or obj.get("UDID") or obj.get("Udid")
if not udid:
continue
address = obj.get("address") or obj.get("Address")
rsd_port = obj.get("rsdPort") or obj.get("rsd_port") or obj.get("RsdPort")
userspace_port = (obj.get("userspaceTunPort") or obj.get("userspacePort") or
obj.get("userspace_port") or obj.get("UserspacePort"))
opts: Dict[str, Any] = {}
if address:
opts["address"] = address
if rsd_port:
try:
opts["rsd_port"] = int(rsd_port)
except (ValueError, TypeError):
pass
if userspace_port:
try:
opts["userspace_port"] = int(userspace_port)
except (ValueError, TypeError):
pass
if opts:
self._device_opts[udid] = opts
def status(self, _tunnel_info_port: Optional[int] = None) -> Tuple[bool, str]:
"""
查询 tunnel 状态
使用 `ios tunnel ls`，并解析输出判断是否 agent 已运行
"""
args = ["tunnel", "ls"]
# 为了兼容 go-ios 使用的同一环境，用户态时也注入 agent 环境变量
extra_env = {"ENABLE_GO_IOS_AGENT": "user"}
code, out, err = self.goios.run(args, timeout=15, extra_env=extra_env)
raw = out.strip() or err or ""
logger.debug("Tunnel 状态输出: %s", raw)
# 缓存 JSON 信息
if raw.startswith("[") or raw.startswith("{"):
self._cache_from_ls_raw(raw)
# 判断 agent 是否未运行
if "not running" in raw.lower():
return False, "agent 未运行"
if "failed to get tunnel info" in raw.lower():
return False, "tunnel server 未响应"
if "connectex: No connection could be made" in raw:
return False, "tunnel服务连接失败"
# 返回码 0 且包含 JSON 列表，说明 agent 已运行
if code == 0:
# 检查是否包含有效的tunnel信息
if raw.startswith("[") or raw.startswith("{"):
try:
data = json.loads(raw)
# 空数组表示没有活动的 tunnel
if isinstance(data, list) and len(data) == 0:
return False, "没有活动的tunnel"
# 非空数组或有效对象表示有活动的 tunnel
if isinstance(data, list) and len(data) > 0:
logger.debug("Tunnel 状态: 已运行")
return True, raw
if isinstance(data, dict) and data:
logger.debug("Tunnel 状态: 已运行")
return True, raw
except json.JSONDecodeError:
pass
elif "no tunnels running" in raw.lower():
return False, "没有活动的tunnel"
# 默认兜底
return code == 0, raw or "未知 tunnel 状态"
def stop(self) -> Tuple[bool, str]:
"""
停止 tunnel
"""
code, out, err = self.goios.run(["tunnel", "stopagent"], timeout=30)
ok = (code == 0)
msg = out.strip() or err or ("停止成功" if ok else "停止失败")
if ok:
logger.debug("Tunnel 停止成功: %s", msg)
else:
logger.warning("Tunnel 停止失败: %s", msg)
return ok, msg
def get_info_port(self) -> Optional[int]:
return self.info_port
def check_windows_wintun(self) -> Tuple[bool, str]:
"""检查Windows系统的wintun.dll是否存在，缺失时尝试自动安装"""
if platform.system().lower() != "windows":
return True, "非Windows系统，无需检查wintun.dll"
wintun_path = os.path.join(os.environ.get('SystemRoot', 'C:\Windows'), 'system32', 'wintun.dll')
if os.path.exists(wintun_path):
return True, "wintun.dll 已存在"
else:
# 尝试自动安装
logger.debug("检测到 wintun.dll 缺失，尝试自动安装...")
install_ok, install_msg = self._auto_install_wintun()
if install_ok:
return True, f"wintun.dll 自动安装成功: {install_msg}"
else:
return False, (
"检测到Windows系统缺少 wintun.dll
"
f"自动安装失败: {install_msg}
"
"手动安装步骤：
"
"1. 确保以管理员权限运行应用
"
"2. 检查 IOSPrechecker/wintun/{amd64|x86|arm|arm64} 目录结构
"
"3. 或访问 https://git.zx2c4.com/wintun 手动下载"
)
def get_goios_opts(self, udid: str) -> Dict[str, Any]:
opts = dict(self._device_opts.get(udid, {}))
if self.info_port:
opts["tunnel_info_port"] = self.info_port
return opts10、requirements.txt
altgraph==0.17.4
blinker==1.9.0
certifi==2025.8.3
charset-normalizer==3.4.3
click==8.2.1
colorama==0.4.6
Flask==3.1.1
greenlet==3.2.4
idna==3.10
itsdangerous==2.2.0
Jinja2==3.1.6
MarkupSafe==3.0.2
numpy==2.2.6
opencv-python==4.12.0.88
packaging==25.0
pefile==2023.2.7
psutil==6.1.0
pyee==13.0.0
pyinstaller==6.15.0
pyinstaller-hooks-contrib==2025.8
pywin32-ctypes==0.2.3
requests==2.32.4
setuptools==80.9.0
typing_extensions==4.14.1
urllib3==2.5.0
Werkzeug==3.1.312、⚠️NOTICE⚠️
本项目基于开源软件开发，不以盈利为目的，仅用于学术交流，使用时请注意遵守开源协议与Apple 开发者协议以及相关的知识产权法规❗❗❗
© 版权声明
文章版权归作者所有，未经允许请勿转载。
相关文章
暂无评论...
