RED = '\033[0;31m'

GREEN = '\033[0;32m'

YELLOW = '\033[0;33m'

BLUE = '\033[0;34m'

CYAN = '\033[0;36m'

MAGENTA = '\033[0;35m'

"""Magisk/KernelSU/APatch 模块安装器"""

def __init__(self, module_path: str, device_serial: Optional[str] = None):

self.module_path = Path(module_path)

self.device_serial = device_serial

self.root_method = None

self.device = None

self.module_info = {}

def ui_print(self, text: str) -> None:

"""打印成功/信息消息"""

print(f"{Colors.GREEN}{text}{Colors.NC}")

def ui_error(self, text: str) -> None:

"""打印错误消息"""

print(f"{Colors.RED}{text}{Colors.NC}")

def ui_warning(self, text: str) -> None:

"""打印警告消息"""

print(f"{Colors.YELLOW}{text}{Colors.NC}")

def ui_info(self, text: str) -> None:

"""打印信息消息"""

print(f"{Colors.BLUE}{text}{Colors.NC}")

def ui_output(self, text: str) -> None:

"""打印原始输出（用于显示命令输出）"""

print(f"{Colors.CYAN}{text}{Colors.NC}")

def ui_title(self, text: str) -> None:

"""打印标题"""

print(f"{Colors.MAGENTA}{text}{Colors.NC}")

def print_title(self, title: str, subtitle: str = "") -> None:

"""打印标题框"""

# 计算中文字符长度（中文算2个字符宽度）

def get_display_len(text: str) -> int:

"""获取字符串显示宽度（中文算2，英文算1）"""

length = 0

for char in text:

if '\u4e00' <= char <= '\u9fff': # 中文字符范围

length += 2

else:

length += 1

return length

title_len = get_display_len(title)

sub_len = get_display_len(subtitle) if subtitle else 0

max_len = max(title_len, sub_len) + 2

bar = "*" * max_len

self.ui_print(bar)

self.ui_print(f" {title} ")

if subtitle:

self.ui_print(f" {subtitle} ")

self.ui_print(bar)

def print_module_info(self) -> None:

"""打印模块详细信息"""

if not self.module_info:

return

self.ui_title("════════════════════════════════════════════════════════════")

self.ui_title(" 模块信息 ")

self.ui_title("════════════════════════════════════════════════════════════")

# 模块 ID

if self.module_info.get('id'):

print(f" {Colors.CYAN}模块 ID:{Colors.NC} {self.module_info['id']}")

# 模块名称

if self.module_info.get('name'):

print(f" {Colors.CYAN}模块名称:{Colors.NC} {self.module_info['name']}")

# 版本信息

version_str = ""

if self.module_info.get('version'):

version_str += f"版本: {self.module_info['version']}"

if self.module_info.get('versionCode'):

if version_str:

version_str += f" ({self.module_info['versionCode']})"

else:

version_str += f"版本号: {self.module_info['versionCode']}"

if version_str:

print(f" {Colors.CYAN}{version_str}{Colors.NC}")

# 作者

if self.module_info.get('author'):

print(f" {Colors.CYAN}作者:{Colors.NC} {self.module_info['author']}")

# 简介

if self.module_info.get('description'):

print(f" {Colors.CYAN}简介:{Colors.NC}")

# 格式化简介，每行不超过 60 个字符

desc = self.module_info['description']

words = desc.split()

line = " "

for word in words:

if len(line + word) > 64: # 每行最多 60 字符加上缩进

print(line)

line = " " + word

else:

if line == " ":

line += word

else:

line += " " + word

if line != " ":

print(line)

# 更新配置

if self.module_info.get('updateJson'):

print(f" {Colors.CYAN}更新配置:{Colors.NC}")

print(f" {self.module_info['updateJson']}")

# 其他可选字段

if self.module_info.get('support'):

print(f" {Colors.CYAN}技术支持:{Colors.NC} {self.module_info['support']}")

print()

def abort(self, message: str) -> None:

"""终止脚本并显示错误"""

self.ui_error(f"✗ {message}")

sys.exit(1)

def run_adb(self, args: List[str], check: bool = True, capture: bool = True) -> Optional[subprocess.CompletedProcess]:

"""执行 ADB 命令"""

cmd = ["adb"]

if self.device_serial:

cmd.extend(["-s", self.device_serial])

cmd.extend(args)

try:

if capture:

# 在 Windows 下使用 utf-8 编码

if sys.platform == 'win32':

result = subprocess.run(cmd, capture_output=True, text=True,

encoding='utf-8', errors='ignore', check=check)

else:

result = subprocess.run(cmd, capture_output=True, text=True, check=check)

else:

# 不捕获输出，直接显示到终端

# 设置编码为 utf-8 避免解码错误

if sys.platform == 'win32':

result = subprocess.run(cmd, check=check, encoding='utf-8', errors='ignore')

else:

result = subprocess.run(cmd, check=check)

return result

except subprocess.CalledProcessError as e:

if check:

self.abort(f"ADB 命令执行失败: {' '.join(cmd)}\n错误: {e.stderr}")

return e

except FileNotFoundError:

self.abort("未找到 ADB 命令，请确保已安装 Android Debug Bridge 并添加到 PATH")

except Exception as e:

if check:

self.abort(f"执行命令时出错: {e}")

return None

def run_su(self, command: str, check: bool = True, capture: bool = True) -> Optional[subprocess.CompletedProcess]:

"""以 root 权限执行命令"""

return self.run_adb(["shell", "su", "-c", command], check, capture)

def check_adb(self) -> None:

"""检查 ADB 是否可用"""

try:

subprocess.run(["adb", "version"], capture_output=True, check=True)

except (subprocess.CalledProcessError, FileNotFoundError):

self.abort("未找到 ADB 命令，请确保已安装 Android Debug Bridge 并添加到 PATH")

def get_devices(self) -> List[str]:

"""获取已连接的设备列表"""

result = self.run_adb(["devices"], check=False)

if not result or not result.stdout:

return []

devices = []

for line in result.stdout.split('\n'):

# 匹配设备序列号和状态

match = re.match(r'^([a-zA-Z0-9_.:-]+)\s+device$', line.strip())

if match:

devices.append(match.group(1))

return devices

def select_device(self) -> str:

"""选择要使用的设备"""

devices = self.get_devices()

if not devices:

self.abort("未检测到设备，请连接您的 Android 设备")

if len(devices) == 1 and not self.device_serial:

self.ui_info(f"✓ 检测到设备: {devices[0]}")

return devices[0]

if self.device_serial:

# 检查指定的设备是否存在（支持部分匹配）

if self.device_serial in devices:

return self.device_serial

# 如果没有精确匹配，尝试找到包含指定字符串的设备

matches = [d for d in devices if self.device_serial in d]

if len(matches) == 1:

self.ui_warning(f"⚠ 使用设备: {matches[0]} (匹配自 '{self.device_serial}')")

return matches[0]

elif len(matches) > 1:

self.ui_error(f"✗ 有多个设备匹配 '{self.device_serial}':")

for match in matches:

print(f" {match}")

self.abort("请指定完整的设备序列号")

else:

self.abort(f"未找到设备 '{self.device_serial}'")

# 多个设备且未指定

self.ui_error("检测到多个设备:")

for i, device in enumerate(devices, 1):

print(f" {i}. {device}")

self.abort("请使用 -d 参数指定设备序列号")

def check_root_method(self) -> str:

"""检查 root 方法和权限"""

self.ui_info("正在检查 root 权限...")

# 测试 root 权限

result = self.run_su("echo test", check=False)

if not result or "test" not in (result.stdout or ""):

self.abort("设备未 root 或 root 权限被拒绝")

self.ui_info("✓ Root 权限验证通过")

# 检查 APatch

result = self.run_su("which apd", check=False)

if result and result.stdout and "apd" in result.stdout:

version_result = self.run_su("apd -V", check=False)

version = (version_result.stdout or "").split('\n')[0].strip()

self.ui_info(f"✓ 检测到 APatch: {version}")

return "APatch"

# 检查 Magisk

result = self.run_su("which magisk", check=False)

if result and result.stdout and "magisk" in result.stdout:

version_result = self.run_su("magisk -v", check=False)

version = (version_result.stdout or "").split('\n')[0].strip()

self.ui_info(f"✓ 检测到 Magisk: {version}")

return "Magisk"

# 检查 KernelSU

result = self.run_su("which ksud", check=False)

if result and result.stdout and "ksud" in result.stdout:

version_result = self.run_su("ksud -V", check=False)

version = (version_result.stdout or "").split('\n')[0].strip()

self.ui_info(f"✓ 检测到 KernelSU: {version}")

return "KernelSU"

self.abort("设备已 root，但未找到 Magisk、KernelSU 或 APatch")

def parse_module_prop(self, content: str) -> Dict[str, str]:

"""解析 module.prop 文件内容"""

info = {}

for line in content.split('\n'):

line = line.strip()

if not line or line.startswith('#'):

continue

if '=' in line:

key, value = line.split('=', 1)

key = key.strip()

value = value.strip()

info[key] = value

return info

def validate_module(self) -> Tuple[bool, Optional[str]]:

"""验证模块文件，返回 (是否成功, 模块ID)"""

if not self.module_path.exists():

self.abort(f"模块文件不存在: {self.module_path}")

if not self.module_path.suffix.lower() == '.zip':

self.ui_warning("⚠ 警告: 文件扩展名不是 .zip")

module_id = None

# 检查 ZIP 文件完整性

try:

with zipfile.ZipFile(self.module_path, 'r') as zip_ref:

# 测试 ZIP 文件完整性

bad_file = zip_ref.testzip()

if bad_file:

self.abort(f"ZIP 文件损坏: {bad_file} 已损坏")

# 检查 module.prop

if 'module.prop' not in zip_ref.namelist():

self.abort("无效的模块: 在 ZIP 根目录未找到 module.prop")

# 读取模块信息

with zip_ref.open('module.prop') as prop_file:

content = prop_file.read().decode('utf-8', errors='ignore')

self.module_info = self.parse_module_prop(content)

# 提取常用字段

module_id = self.module_info.get('id')

module_name = self.module_info.get('name')

module_version = self.module_info.get('version')

module_version_code = self.module_info.get('versionCode')

module_author = self.module_info.get('author')

module_description = self.module_info.get('description')

module_update_json = self.module_info.get('updateJson')

# 显示模块信息

self.print_module_info()

except zipfile.BadZipFile:

self.abort("无效或损坏的 ZIP 文件")

except Exception as e:

self.abort(f"验证模块失败: {e}")

return True, module_id

def push_module(self) -> None:

"""推送模块到设备"""

self.ui_info("正在推送模块到设备...")

# 创建临时目录

cache_dir = Path.home() / ".cache" / "module_installer"

cache_dir.mkdir(parents=True, exist_ok=True)

temp_zip = cache_dir / "module.zip"

try:

# 复制模块文件

shutil.copy2(self.module_path, temp_zip)

# 清理设备上的旧文件

self.run_adb(["shell", "rm", "-rf", "/data/local/tmp/module.zip"], check=False)

# 推送新文件

result = self.run_adb(["push", str(temp_zip), "/data/local/tmp/module.zip"], check=False)

if not result or result.returncode != 0:

self.abort("推送模块到设备失败")

self.ui_info("✓ 模块推送成功")

finally:

# 清理临时文件

if temp_zip.exists():

temp_zip.unlink()

def get_module_id_from_device(self) -> Optional[str]:

"""从设备获取模块 ID"""

return self.module_info.get('id')

def install_module(self, module_id: Optional[str]) -> None:

"""安装模块并显示原始输出"""

print() # 添加空行

self.ui_info(f"正在使用 {self.root_method} 安装模块...")

self.ui_info("=" * 50)

if self.root_method == "KernelSU":

self.ui_output("执行命令: /data/adb/ksud module install /data/local/tmp/module.zip")

print() # 添加空行

# 不捕获输出，直接显示到终端

self.run_su("/data/adb/ksud module install /data/local/tmp/module.zip",

check=False, capture=False)

elif self.root_method == "APatch":

self.ui_output("执行命令: /data/adb/apd module install /data/local/tmp/module.zip")

print() # 添加空行

# 不捕获输出，直接显示到终端

self.run_su("/data/adb/apd module install /data/local/tmp/module.zip",

check=False, capture=False)

else:

self.ui_output("执行命令: magisk --install-module /data/local/tmp/module.zip")

print() # 添加空行

# 不捕获输出，直接显示到终端

self.run_su("magisk --install-module /data/local/tmp/module.zip",

check=False, capture=False)

self.ui_info("=" * 50)

print() # 添加空行

# 清理临时文件

self.run_adb(["shell", "rm", "-rf", "/data/local/tmp/module.zip"], check=False)

# 等待安装完成

time.sleep(2)

# 检查安装结果

self.ui_info("正在检查安装结果...")

# 获取模块 ID 用于检查

if not module_id:

module_id = self.get_module_id_from_device()

if module_id:

# 检查模块是否已安装

if self.root_method == "KernelSU":

result = self.run_su("/data/adb/ksud module list", check=False)

output = result.stdout if result and result.stdout else ""

elif self.root_method == "APatch":

result = self.run_su("/data/adb/apd module list", check=False)

output = result.stdout if result and result.stdout else ""

else:

result = self.run_su("magisk --list", check=False)

output = result.stdout if result and result.stdout else ""

if module_id in output:

self.ui_print("✓ 模块安装成功！")

self.ui_warning("⚠ 请重启设备以应用更改")

else:

self.ui_warning("⚠ 模块可能未成功安装，请检查上面的错误信息")

if "Failed to request license" in output or "license" in output.lower():

self.ui_warning("提示: 此模块需要联网验证授权，请确保设备已连接互联网")

elif "magisk: not found" in output:

self.ui_warning(f"提示: 此模块可能需要 Magisk 环境，但当前使用的是 {self.root_method}")

elif "ksud: not found" in output:

self.ui_warning(f"提示: 此模块可能需要 KernelSU 环境，但当前使用的是 {self.root_method}")

elif "apd: not found" in output:

self.ui_warning(f"提示: 此模块可能需要 APatch 环境，但当前使用的是 {self.root_method}")

else:

self.ui_warning("⚠ 无法获取模块 ID 进行验证，请手动检查安装结果")

self.ui_warning("如果看到错误信息，请根据错误提示解决问题")

def ask_reboot(self) -> None:

"""询问是否重启设备"""

print()

response = input("是否现在重启设备？(y/N): ").strip().lower()

if response in ['y', 'yes']:

self.ui_info("正在重启设备...")

self.run_adb(["reboot"], check=False)

self.ui_print("✓ 设备正在重启，请等待设备重新启动")

else:

self.ui_info("请记得手动重启设备以应用模块更改")

def confirm_installation(self) -> bool:

"""确认是否继续安装"""

print()

response = input("确认继续安装？(y/N): ").strip().lower()

return response in ['y', 'yes']

def run(self) -> None:

"""执行安装流程"""

# 清屏

os.system('clear' if os.name == 'posix' else 'cls')

# 打印标题

self.print_title("ModuleInstaller", "Magisk/KernelSU/APatch 模块安装工具")

# 检查 ADB

self.check_adb()

# 验证模块（会显示模块信息）

success, module_id = self.validate_module()

if not success:

return

# 选择设备

self.device = self.select_device()

self.ui_info(f"当前设备: {self.device}")

# 检查 root 方法

self.root_method = self.check_root_method()

# self.ui_info(f"Root 方式: {self.root_method}")

# 确认安装

if not self.confirm_installation():

self.abort("用户取消安装")

# 推送模块

self.push_module()

# 安装模块

self.install_module(module_id)

# 询问重启

"""主函数"""

parser = argparse.ArgumentParser(

description="Magisk/KernelSU/APatch 模块安装器",

formatter_class=argparse.RawDescriptionHelpFormatter,

epilog="""

)

parser.add_argument(

'ZipPath',

nargs='?',

help='Magisk/KernelSU/APatch 模块 ZIP 文件路径'

)

parser.add_argument(

'-d', '--device',

help='ADB 设备序列号（支持 USB 和无线设备）'

)

parser.add_argument(

'-v', '--version',

action='version',

version='模块安装器 v2.1 (Python)',

help='显示版本信息'

)

args = parser.parse_args()

# 检查是否提供了模块文件

if not args.ZipPath:

parser.print_help()

sys.exit(1)

# 创建安装器并运行

installer = ModuleInstaller(args.ZipPath, args.device)