1213 lines
49 KiB
Python
1213 lines
49 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
import time
|
||
import tkinter as tk
|
||
import traceback
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from tkinter import filedialog, messagebox, scrolledtext, simpledialog, ttk
|
||
from urllib.parse import urlencode
|
||
from urllib.request import Request, urlopen
|
||
|
||
|
||
|
||
def get_app_dir():
|
||
return Path(sys.executable).resolve().parent if getattr(sys, 'frozen', False) else Path(__file__).resolve().parent
|
||
|
||
|
||
def resource_candidates(file_name):
|
||
base_dir = get_app_dir()
|
||
candidates = []
|
||
if getattr(sys, 'frozen', False):
|
||
candidates.append(Path(getattr(sys, '_MEIPASS', base_dir)) / file_name)
|
||
candidates.extend([
|
||
base_dir / file_name,
|
||
base_dir / 'tools' / file_name,
|
||
base_dir / 'shared' / file_name,
|
||
base_dir.parent / 'tools' / file_name,
|
||
base_dir.parent / 'shared' / file_name,
|
||
base_dir.parent / file_name,
|
||
])
|
||
unique = []
|
||
for candidate in candidates:
|
||
if candidate not in unique:
|
||
unique.append(candidate)
|
||
return unique
|
||
|
||
|
||
def find_resource(file_name):
|
||
candidates = resource_candidates(file_name)
|
||
for candidate in candidates:
|
||
if candidate.exists():
|
||
return candidate
|
||
return candidates[0]
|
||
|
||
|
||
def find_tool(file_name, fallback=None):
|
||
path = find_resource(file_name)
|
||
if path.exists():
|
||
return str(path)
|
||
return fallback or str(path)
|
||
class UNIZLanguageGUI:
|
||
PASSWORD_API_URL = "https://api.changan.softwindy.cn/api/authorizations/get-uni-z-pwd"
|
||
PACKAGE_KEY_API_URL = "https://api.changan.softwindy.cn/api/authorizations/package-key"
|
||
DOWNLOAD_DIR = "/storage/self/primary/Download/"
|
||
APPS_DIR = "/storage/self/primary/Download/apps/"
|
||
THIRD_APPS_DIR = "/storage/self/primary/Download/Third-apps/"
|
||
MANAGER_APK_NAME = "yingyongguanjia.apk"
|
||
|
||
def __init__(self):
|
||
self.root = tk.Tk()
|
||
self.root.title("UNI-Z语言文件推送工具")
|
||
self.root.geometry("650x560")
|
||
self.root.resizable(True, True)
|
||
|
||
self.colors = {
|
||
"bg_dark": "#1e1e2e",
|
||
"bg_light": "#2a2a3e",
|
||
"accent": "#6c5ce7",
|
||
"accent_hover": "#5b4bc4",
|
||
"success": "#00b894",
|
||
"error": "#d63031",
|
||
"warning": "#fdcb6e",
|
||
"info": "#0984e3",
|
||
"text": "#dfe6e9",
|
||
"text_secondary": "#b2bec3",
|
||
"border": "#3d3d5e",
|
||
}
|
||
|
||
self.base_dir = get_app_dir()
|
||
self.adb = find_tool("adb.exe", "adb")
|
||
self.sz = find_tool("7za.exe")
|
||
|
||
self.package_file = find_resource("package.bin")
|
||
self.temp_dir = self.get_cache_dir()
|
||
self.apps_dir = None
|
||
self.manager_apk = None
|
||
self.device_connected = False
|
||
self.vin = None
|
||
self.query_result = None
|
||
self.debug_password = None
|
||
self.extract_password = None
|
||
self.last_resource_error = ""
|
||
self.debug_mode = False
|
||
self._refreshing = False
|
||
self.vin_placeholder = "请输入VIN"
|
||
|
||
self.setup_styles()
|
||
self.setup_ui()
|
||
self.center_window()
|
||
self.check_environment()
|
||
self.start_device_monitor()
|
||
|
||
def setup_styles(self):
|
||
style = ttk.Style()
|
||
style.theme_use("clam")
|
||
style.configure("TFrame", background=self.colors["bg_dark"])
|
||
style.configure("TLabel", background=self.colors["bg_dark"], foreground=self.colors["text"])
|
||
style.configure(
|
||
"TProgressbar",
|
||
background=self.colors["accent"],
|
||
troughcolor=self.colors["bg_light"],
|
||
borderwidth=0,
|
||
)
|
||
|
||
def setup_ui(self):
|
||
self.root.configure(bg=self.colors["bg_dark"])
|
||
main_frame = tk.Frame(self.root, bg=self.colors["bg_dark"])
|
||
main_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
|
||
|
||
title_frame = tk.Frame(main_frame, bg=self.colors["bg_dark"], height=64)
|
||
title_frame.pack(fill=tk.X, pady=(0, 10))
|
||
title_frame.pack_propagate(False)
|
||
|
||
tk.Label(
|
||
title_frame,
|
||
text="🚀 UNI-Z语言文件推送工具",
|
||
font=("Microsoft YaHei", 18, "bold"),
|
||
fg=self.colors["accent"],
|
||
bg=self.colors["bg_dark"],
|
||
).pack()
|
||
tk.Label(
|
||
title_frame,
|
||
text="宜宾科宜科技有限公司 - 出口改装一站式服务",
|
||
font=("Microsoft YaHei", 9),
|
||
fg=self.colors["text_secondary"],
|
||
bg=self.colors["bg_dark"],
|
||
).pack()
|
||
|
||
query_frame = tk.Frame(main_frame, bg=self.colors["bg_light"], relief=tk.RAISED, bd=1)
|
||
query_frame.pack(fill=tk.X, padx=5, pady=(0, 8))
|
||
|
||
tk.Label(
|
||
query_frame,
|
||
text="VIN码:",
|
||
font=("Microsoft YaHei", 9),
|
||
fg=self.colors["text"],
|
||
bg=self.colors["bg_light"],
|
||
).grid(row=0, column=0, padx=(10, 5), pady=8, sticky="w")
|
||
|
||
self.vin_input = tk.Entry(
|
||
query_frame,
|
||
font=("Consolas", 10),
|
||
bg="#2d2d3d",
|
||
fg="#636e72",
|
||
insertbackground="white",
|
||
relief=tk.FLAT,
|
||
width=24,
|
||
)
|
||
self.vin_input.insert(0, self.vin_placeholder)
|
||
self.vin_input.bind("<FocusIn>", self._on_vin_input_focus_in)
|
||
self.vin_input.bind("<FocusOut>", self._on_vin_input_focus_out)
|
||
self.vin_input.grid(row=0, column=1, padx=5, pady=8, sticky="w")
|
||
|
||
self.btn_query = tk.Button(
|
||
query_frame,
|
||
text="查询密码",
|
||
command=self.query_password_by_vin,
|
||
font=("Microsoft YaHei", 9),
|
||
fg="white",
|
||
bg=self.colors["accent"],
|
||
relief=tk.FLAT,
|
||
cursor="hand2",
|
||
width=12,
|
||
)
|
||
self.btn_query.grid(row=0, column=2, padx=5, pady=8)
|
||
|
||
self.factory_pwd_label = tk.Label(
|
||
query_frame,
|
||
text="工厂模式密码: 未查询",
|
||
font=("Microsoft YaHei", 9, "bold"),
|
||
fg=self.colors["text_secondary"],
|
||
bg=self.colors["bg_light"],
|
||
anchor="w",
|
||
width=24,
|
||
)
|
||
self.factory_pwd_label.grid(row=1, column=0, columnspan=2, padx=10, pady=(0, 8), sticky="w")
|
||
|
||
self.debug_pwd_label = tk.Label(
|
||
query_frame,
|
||
text="调试密码: 未查询",
|
||
font=("Microsoft YaHei", 9, "bold"),
|
||
fg=self.colors["text_secondary"],
|
||
bg=self.colors["bg_light"],
|
||
anchor="w",
|
||
width=24,
|
||
)
|
||
self.debug_pwd_label.grid(row=1, column=2, padx=5, pady=(0, 8), sticky="w")
|
||
|
||
status_frame = tk.Frame(main_frame, bg=self.colors["bg_light"], relief=tk.RAISED, bd=1)
|
||
status_frame.pack(fill=tk.X, padx=5, pady=(0, 8))
|
||
tk.Label(
|
||
status_frame,
|
||
text="设备状态:",
|
||
font=("Microsoft YaHei", 9),
|
||
fg=self.colors["text"],
|
||
bg=self.colors["bg_light"],
|
||
).pack(side=tk.LEFT, padx=(10, 4), pady=8)
|
||
self.device_status_label = tk.Label(
|
||
status_frame,
|
||
text="未检测",
|
||
font=("Microsoft YaHei", 9, "bold"),
|
||
fg=self.colors["text_secondary"],
|
||
bg=self.colors["bg_light"],
|
||
)
|
||
self.device_status_label.pack(side=tk.LEFT, padx=(0, 10), pady=8)
|
||
|
||
button_frame = tk.Frame(main_frame, bg=self.colors["bg_light"], relief=tk.RAISED, bd=1)
|
||
button_frame.pack(fill=tk.X, padx=5, pady=(0, 8))
|
||
|
||
self.btn_push = tk.Button(
|
||
button_frame,
|
||
text="📦 推送资源文件",
|
||
command=self.push_language_package,
|
||
font=("Microsoft YaHei", 10),
|
||
fg="white",
|
||
bg=self.colors["accent"],
|
||
relief=tk.FLAT,
|
||
cursor="hand2",
|
||
width=13,
|
||
)
|
||
self.btn_push.pack(side=tk.LEFT, padx=10, pady=8)
|
||
|
||
self.btn_push_third = tk.Button(
|
||
button_frame,
|
||
text="📲 推送第三方APK",
|
||
command=self.push_third_party_apks,
|
||
font=("Microsoft YaHei", 10),
|
||
fg="white",
|
||
bg=self.colors["success"],
|
||
relief=tk.FLAT,
|
||
cursor="hand2",
|
||
width=14,
|
||
)
|
||
self.btn_push_third.pack(side=tk.LEFT, padx=5, pady=8)
|
||
|
||
self.btn_settings = tk.Button(
|
||
button_frame,
|
||
text="⚙️ 原生设置",
|
||
command=self.open_android_settings,
|
||
font=("Microsoft YaHei", 10),
|
||
fg="white",
|
||
bg=self.colors["info"],
|
||
relief=tk.FLAT,
|
||
cursor="hand2",
|
||
width=12,
|
||
)
|
||
self.btn_settings.pack(side=tk.LEFT, padx=5, pady=8)
|
||
|
||
self.btn_reboot = tk.Button(
|
||
button_frame,
|
||
text="🔄 重启设备",
|
||
command=self.reboot_device,
|
||
font=("Microsoft YaHei", 10),
|
||
fg="white",
|
||
bg=self.colors["warning"],
|
||
relief=tk.FLAT,
|
||
cursor="hand2",
|
||
width=14,
|
||
)
|
||
self.btn_reboot.pack(side=tk.LEFT, padx=5, pady=8)
|
||
|
||
tips_frame = tk.Frame(main_frame, bg=self.colors["bg_light"], relief=tk.RAISED, bd=1)
|
||
tips_frame.pack(fill=tk.X, padx=5, pady=(0, 8))
|
||
tips = [
|
||
"1. 请先输入 VIN 查询密码，再推送文件到 /storage/self/primary/Download/。",
|
||
"2. 推送完成后会自动打开设备文件管理。",
|
||
"3. 第三方 APK 会推送到 /storage/self/primary/Download/Third-apps/。",
|
||
]
|
||
for tip in tips:
|
||
tk.Label(
|
||
tips_frame,
|
||
text=tip,
|
||
font=("Microsoft YaHei", 9),
|
||
fg=self.colors["warning"],
|
||
bg=self.colors["bg_light"],
|
||
anchor="w",
|
||
).pack(fill=tk.X, padx=10, pady=(4, 0))
|
||
|
||
progress_frame = tk.Frame(main_frame, bg=self.colors["bg_dark"])
|
||
progress_frame.pack(fill=tk.X, padx=5, pady=(0, 8))
|
||
self.progress_label = tk.Label(
|
||
progress_frame,
|
||
text="",
|
||
font=("Microsoft YaHei", 9),
|
||
fg=self.colors["text_secondary"],
|
||
bg=self.colors["bg_dark"],
|
||
)
|
||
self.progress_label.pack()
|
||
self.progress = ttk.Progressbar(progress_frame, mode="determinate", style="TProgressbar")
|
||
self.progress.pack(fill=tk.X)
|
||
|
||
log_card = tk.Frame(main_frame, bg=self.colors["bg_light"], relief=tk.RAISED, bd=1)
|
||
log_card.pack(fill=tk.BOTH, expand=True, padx=5)
|
||
|
||
log_title = tk.Frame(log_card, bg=self.colors["bg_dark"], height=30)
|
||
log_title.pack(fill=tk.X)
|
||
log_title.pack_propagate(False)
|
||
tk.Label(
|
||
log_title,
|
||
text="📋 运行日志",
|
||
font=("Microsoft YaHei", 10, "bold"),
|
||
fg=self.colors["accent"],
|
||
bg=self.colors["bg_dark"],
|
||
).pack(side=tk.LEFT, padx=10)
|
||
tk.Button(
|
||
log_title,
|
||
text="清空日志",
|
||
command=self.clear_log,
|
||
font=("Microsoft YaHei", 8),
|
||
fg=self.colors["text_secondary"],
|
||
bg=self.colors["bg_dark"],
|
||
relief=tk.FLAT,
|
||
cursor="hand2",
|
||
).pack(side=tk.RIGHT, padx=10)
|
||
|
||
self.log_text = scrolledtext.ScrolledText(
|
||
log_card,
|
||
height=10,
|
||
wrap=tk.WORD,
|
||
font=("Consolas", 9),
|
||
bg="#2d2d3d",
|
||
fg="#e0e0e0",
|
||
insertbackground="white",
|
||
relief=tk.FLAT,
|
||
borderwidth=0,
|
||
)
|
||
self.log_text.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
|
||
self.log_text.tag_config("INFO", foreground="#74b9ff")
|
||
self.log_text.tag_config("SUCCESS", foreground="#55efc4")
|
||
self.log_text.tag_config("ERROR", foreground="#ff7675")
|
||
self.log_text.tag_config("WARNING", foreground="#ffeaa7")
|
||
self.log_text.tag_config("CMD", foreground="#a29bfe")
|
||
self.root.bind("<Control-Shift-D>", self._toggle_debug)
|
||
self.root.bind("<Control-Shift-E>", self._debug_extract_package)
|
||
|
||
def center_window(self):
|
||
self.root.update_idletasks()
|
||
screen_w = self.root.winfo_screenwidth()
|
||
screen_h = self.root.winfo_screenheight()
|
||
win_w = self.root.winfo_reqwidth()
|
||
win_h = self.root.winfo_reqheight()
|
||
x = (screen_w - win_w) // 2
|
||
y = (screen_h - win_h) // 2
|
||
self.root.geometry(f"+{x}+{y}")
|
||
|
||
def run_on_ui_thread(self, func, *args, **kwargs):
|
||
self.root.after(0, func, *args, **kwargs)
|
||
|
||
def _adb_cmd(self):
|
||
return subprocess.list2cmdline([self.adb])
|
||
|
||
def _log_impl(self, message, level="INFO"):
|
||
timestamp = datetime.now().strftime("%H:%M:%S")
|
||
self.log_text.insert(tk.END, f"[{timestamp}] [{level}] {message}\n", level)
|
||
self.log_text.see(tk.END)
|
||
|
||
def log(self, message, level="INFO"):
|
||
self.run_on_ui_thread(self._log_impl, message, level)
|
||
|
||
def clear_log(self):
|
||
self.log_text.delete(1.0, tk.END)
|
||
self.log("日志已清空", "INFO")
|
||
|
||
def _toggle_debug(self, event=None):
|
||
if self.debug_mode:
|
||
self.debug_mode = False
|
||
self.log("调试模式已关闭", "WARNING")
|
||
return
|
||
|
||
pwd = simpledialog.askstring("调试模式", "请输入调试密码:", show="*", parent=self.root)
|
||
if pwd == "zxch5200":
|
||
self.debug_mode = True
|
||
self.log("调试模式已开启，将显示完整命令和输出", "WARNING")
|
||
elif pwd is not None:
|
||
messagebox.showwarning("错误", "密码错误")
|
||
|
||
def _debug_extract_package(self, event=None):
|
||
if not self.debug_mode:
|
||
messagebox.showwarning("调试模式", "请先按 Ctrl+Shift+D 开启调试模式")
|
||
return
|
||
|
||
password = simpledialog.askstring(
|
||
"测试解压资源包",
|
||
"请输入 package.bin 解压密码:",
|
||
show="*",
|
||
parent=self.root,
|
||
)
|
||
if not password:
|
||
return
|
||
|
||
self.extract_password = password
|
||
|
||
def do_extract():
|
||
try:
|
||
self.log("调试模式：开始离线测试解压 package.bin，跳过设备检测和云端密码请求", "WARNING")
|
||
if self.temp_dir and self.temp_dir.exists():
|
||
self.log("调试模式：将重新解压并覆盖现有缓存", "WARNING")
|
||
shutil.rmtree(str(self.temp_dir), ignore_errors=True)
|
||
time.sleep(0.3)
|
||
|
||
if self.extract_package(force=True):
|
||
apps_count = len(self.collect_apps_apks())
|
||
self.log(
|
||
f"调试解压成功：apps 目录包含 {apps_count} 个 APK，已找到 {self.MANAGER_APK_NAME}",
|
||
"SUCCESS",
|
||
)
|
||
self.run_on_ui_thread(messagebox.showinfo, "测试解压完成", "package.bin 解压成功")
|
||
else:
|
||
self.run_on_ui_thread(
|
||
messagebox.showerror,
|
||
"测试解压失败",
|
||
self.last_resource_error or "请检查 package.bin 或解压密码",
|
||
)
|
||
except Exception as e:
|
||
self.log(f"调试解压异常: {e}", "ERROR")
|
||
self.log(traceback.format_exc(), "ERROR")
|
||
self.run_on_ui_thread(messagebox.showerror, "测试解压异常", str(e))
|
||
|
||
threading.Thread(target=do_extract, daemon=True).start()
|
||
|
||
def _update_progress_impl(self, value, max_value=100, label=""):
|
||
percent = 0 if max_value <= 0 else min(100, (value / max_value) * 100)
|
||
self.progress["value"] = percent
|
||
self.progress_label.config(text=f"{label}: {value}/{max_value} ({percent:.1f}%)")
|
||
self.root.update_idletasks()
|
||
|
||
def update_progress(self, value, max_value=100, label=""):
|
||
self.run_on_ui_thread(self._update_progress_impl, value, max_value, label)
|
||
|
||
def check_environment(self):
|
||
if not self.package_file.exists():
|
||
self.log(f"未找到资源包: {self.package_file}", "WARNING")
|
||
if not os.path.exists(self.sz):
|
||
self.log(f"未找到 7za.exe: {self.sz}", "ERROR")
|
||
ok, output = self.run_adb_command("adb version")
|
||
if ok:
|
||
self.log("ADB 环境正常", "SUCCESS")
|
||
else:
|
||
self.log(f"ADB 不可用: {output}", "ERROR")
|
||
|
||
def run_adb_command(self, command, log_debug=True):
|
||
command = command.replace("adb", self._adb_cmd(), 1)
|
||
if self.debug_mode and log_debug:
|
||
self.log(f"CMD: {command}", "CMD")
|
||
try:
|
||
result = subprocess.run(
|
||
command,
|
||
shell=True,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
universal_newlines=True,
|
||
)
|
||
output = (result.stdout or "").strip()
|
||
error = (result.stderr or "").strip()
|
||
if self.debug_mode and log_debug:
|
||
level = "CMD" if result.returncode == 0 else "ERROR"
|
||
self.log(f"RET: {result.returncode}", level)
|
||
if output:
|
||
self.log(f"STDOUT:\n{output}", "CMD")
|
||
if error:
|
||
self.log(f"STDERR:\n{error}", "ERROR")
|
||
if result.returncode == 0:
|
||
return True, output
|
||
return False, error or output
|
||
except Exception as e:
|
||
if self.debug_mode and log_debug:
|
||
self.log(f"CMD 异常: {e}\n{traceback.format_exc()}", "ERROR")
|
||
return False, str(e)
|
||
|
||
def refresh_device_status(self):
|
||
if self._refreshing:
|
||
return
|
||
self._refreshing = True
|
||
|
||
def refresh():
|
||
try:
|
||
was_connected = self.device_connected
|
||
ok, output = self.run_adb_command("adb -d devices", log_debug=False)
|
||
connected = False
|
||
if ok:
|
||
lines = output.splitlines()
|
||
devices = [
|
||
line for line in lines[1:]
|
||
if line.strip() and "\tdevice" in line and "offline" not in line
|
||
]
|
||
connected = bool(devices)
|
||
self.update_device_status(connected)
|
||
if connected and not was_connected:
|
||
self.log("设备已连接", "SUCCESS")
|
||
elif not connected and was_connected:
|
||
self.log("设备已断开连接", "WARNING")
|
||
finally:
|
||
self._refreshing = False
|
||
|
||
threading.Thread(target=refresh, daemon=True).start()
|
||
|
||
def update_device_status(self, connected):
|
||
self.device_connected = connected
|
||
self.run_on_ui_thread(self._update_device_status_impl, connected)
|
||
|
||
def _update_device_status_impl(self, connected):
|
||
if connected:
|
||
self.device_status_label.config(text="已连接", fg=self.colors["success"])
|
||
else:
|
||
self.device_status_label.config(text="未连接", fg=self.colors["error"])
|
||
|
||
def start_device_monitor(self):
|
||
def monitor():
|
||
while True:
|
||
try:
|
||
self.refresh_device_status()
|
||
time.sleep(5)
|
||
except Exception:
|
||
time.sleep(5)
|
||
|
||
threading.Thread(target=monitor, daemon=True).start()
|
||
|
||
def _on_vin_input_focus_in(self, event):
|
||
if self.vin_input.get() == self.vin_placeholder:
|
||
self.vin_input.delete(0, tk.END)
|
||
self.vin_input.config(fg="#e0e0e0")
|
||
|
||
def _on_vin_input_focus_out(self, event):
|
||
if not self.vin_input.get().strip():
|
||
self.vin_input.delete(0, tk.END)
|
||
self.vin_input.insert(0, self.vin_placeholder)
|
||
self.vin_input.config(fg="#636e72")
|
||
|
||
def query_password_by_vin(self):
|
||
vin = self.vin_input.get().strip()
|
||
if not vin or vin == self.vin_placeholder:
|
||
messagebox.showwarning("提示", "请输入 VIN 码")
|
||
return
|
||
self.vin = None
|
||
self.query_result = None
|
||
self.debug_password = None
|
||
self.extract_password = None
|
||
self.factory_pwd_label.config(text="工厂模式密码: 查询中...", fg=self.colors["text_secondary"])
|
||
self.debug_pwd_label.config(text="调试密码: 查询中...", fg=self.colors["text_secondary"])
|
||
|
||
def query():
|
||
try:
|
||
url = f"{self.PASSWORD_API_URL}?{urlencode({'vin': vin})}"
|
||
req = Request(url, method="GET", headers={"User-Agent": "Mozilla/5.0"})
|
||
with urlopen(req, timeout=10) as response:
|
||
data = json.loads(response.read().decode("utf-8"))
|
||
|
||
if not data.get("success"):
|
||
msg = data.get("message", "查询失败")
|
||
self.log(f"密码查询失败: {msg}", "ERROR")
|
||
self.run_on_ui_thread(messagebox.showerror, "查询失败", msg)
|
||
return
|
||
|
||
payload = data.get("data") or {}
|
||
factory_pwd = payload.get("factoryPwd")
|
||
debug_pwd = payload.get("password")
|
||
authorized = bool(payload.get("authorized"))
|
||
if not authorized or not debug_pwd:
|
||
error_msg = payload.get("error") or "设备未授权"
|
||
self.query_result = payload
|
||
self.vin = vin
|
||
self.debug_password = None
|
||
self.extract_password = None
|
||
|
||
def update_unauthorized():
|
||
self.factory_pwd_label.config(
|
||
text="工厂模式密码: 未授权",
|
||
fg=self.colors["warning"],
|
||
)
|
||
self.debug_pwd_label.config(
|
||
text="调试密码: 无可用密码",
|
||
fg=self.colors["warning"],
|
||
)
|
||
self.log("设备未授权，无法获取调试密码", "WARNING")
|
||
messagebox.showwarning("设备未授权", error_msg)
|
||
|
||
self.run_on_ui_thread(update_unauthorized)
|
||
return
|
||
|
||
self.query_result = payload
|
||
self.vin = vin
|
||
self.debug_password = debug_pwd
|
||
self.extract_password = None
|
||
|
||
def update_ui():
|
||
self.factory_pwd_label.config(
|
||
text=f"工厂模式密码: {factory_pwd or '未知'}",
|
||
fg=self.colors["success"],
|
||
)
|
||
self.debug_pwd_label.config(
|
||
text=f"调试密码: {debug_pwd}",
|
||
fg=self.colors["success"],
|
||
)
|
||
self.log(f"密码查询成功 VIN={vin}", "SUCCESS")
|
||
|
||
self.run_on_ui_thread(update_ui)
|
||
except Exception as e:
|
||
self.log(f"密码查询请求失败: {str(e)}", "ERROR")
|
||
self.run_on_ui_thread(messagebox.showerror, "请求失败", str(e))
|
||
|
||
threading.Thread(target=query, daemon=True).start()
|
||
|
||
def _decode_7z_output(self, output):
|
||
for enc in ("gbk", "utf-8"):
|
||
try:
|
||
return output.decode(enc)
|
||
except UnicodeDecodeError:
|
||
continue
|
||
return output.decode("utf-8", errors="replace")
|
||
|
||
def _seven_zip_supports_progress_stream(self):
|
||
try:
|
||
result = subprocess.run(
|
||
[self.sz],
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
universal_newlines=True,
|
||
creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0,
|
||
)
|
||
return "-bs{o|e|p}" in ((result.stdout or "") + (result.stderr or ""))
|
||
except Exception:
|
||
return False
|
||
|
||
def get_cache_dir(self):
|
||
local_appdata = os.environ.get("LOCALAPPDATA", os.path.expanduser("~\\AppData\\Local"))
|
||
return Path(local_appdata) / ".cache" / "system" / ".android" / "apps_cache_UNIZ"
|
||
|
||
def reuse_cached_package(self):
|
||
self.temp_dir = self.get_cache_dir()
|
||
self.apps_dir = None
|
||
self.manager_apk = None
|
||
if not self.temp_dir.exists():
|
||
return False
|
||
|
||
self.apps_dir = self.find_apps_dir()
|
||
if not self.apps_dir:
|
||
return False
|
||
self.manager_apk = self.find_manager_apk()
|
||
if not self.manager_apk:
|
||
return False
|
||
|
||
ok, apk_count, reason = self.validate_cached_resources()
|
||
if not ok:
|
||
self.log(f"已解压缓存无效，将重新解压: {reason}", "WARNING")
|
||
self.clear_cached_package()
|
||
return False
|
||
|
||
self.log(
|
||
f"检测到已解压缓存，复用 apps 目录（{apk_count} 个 APK）和 {self.MANAGER_APK_NAME}",
|
||
"SUCCESS",
|
||
)
|
||
self.update_progress(100, 100, "资源缓存可用")
|
||
return True
|
||
|
||
def validate_cached_resources(self):
|
||
if not self.apps_dir or not self.apps_dir.exists():
|
||
return False, 0, "缺少 apps 目录"
|
||
if not self.manager_apk or not self.manager_apk.exists():
|
||
return False, 0, f"缺少 {self.MANAGER_APK_NAME}"
|
||
|
||
apks = list(self.apps_dir.rglob("*.apk"))
|
||
if not apks:
|
||
return False, 0, "apps 目录没有 APK"
|
||
|
||
zero_apks = [apk.name for apk in apks if apk.stat().st_size <= 0]
|
||
if zero_apks:
|
||
preview = ", ".join(zero_apks[:5])
|
||
suffix = "..." if len(zero_apks) > 5 else ""
|
||
return False, len(apks), f"发现 0KB APK: {preview}{suffix}"
|
||
if self.manager_apk.stat().st_size <= 0:
|
||
return False, len(apks), f"{self.MANAGER_APK_NAME} 是 0KB"
|
||
return True, len(apks), ""
|
||
|
||
def detect_extract_error(self, output):
|
||
text = output.lower()
|
||
password_markers = [
|
||
"wrong password",
|
||
"incorrect password",
|
||
"password is incorrect",
|
||
"can not open encrypted archive",
|
||
"data error in encrypted file",
|
||
]
|
||
if any(marker in text for marker in password_markers):
|
||
return "准备资源出错"
|
||
if "data error" in text:
|
||
return "资源包数据错误，可能是密码错误或 package.bin 损坏"
|
||
if "headers error" in text or "unexpected end" in text:
|
||
return "资源包损坏或不完整，请检查 package.bin"
|
||
return ""
|
||
|
||
def clear_cached_package(self):
|
||
if self.temp_dir and self.temp_dir.exists():
|
||
shutil.rmtree(str(self.temp_dir), ignore_errors=True)
|
||
time.sleep(0.3)
|
||
self.apps_dir = None
|
||
self.manager_apk = None
|
||
|
||
def fetch_package_password(self):
|
||
if not self.vin:
|
||
self.log("请先输入 VIN 查询密码", "ERROR")
|
||
return False
|
||
|
||
try:
|
||
url = f"{self.PACKAGE_KEY_API_URL}?{urlencode({'vin': self.vin})}"
|
||
if self.debug_mode:
|
||
self.log(f"HTTP GET: {url}", "CMD")
|
||
req = Request(url, method="GET", headers={"User-Agent": "Mozilla/5.0"})
|
||
with urlopen(req, timeout=10) as response:
|
||
data = json.loads(response.read().decode("utf-8"))
|
||
if self.debug_mode:
|
||
self.log(
|
||
"PACKAGE-KEY RESPONSE:\n"
|
||
+ json.dumps(data, ensure_ascii=False, indent=2),
|
||
"CMD",
|
||
)
|
||
|
||
payload = data.get("data") or {}
|
||
password = payload.get("password")
|
||
if data.get("success") and password:
|
||
self.extract_password = password
|
||
self.log("资源包解压密码获取成功", "SUCCESS")
|
||
return True
|
||
|
||
msg = payload.get("error") or data.get("message") or "无法获取资源包解压密码"
|
||
self.extract_password = None
|
||
self.log(f"资源包解压密码获取失败: {msg}", "ERROR")
|
||
return False
|
||
except Exception as e:
|
||
self.extract_password = None
|
||
self.log(f"资源包解压密码请求失败: {e}", "ERROR")
|
||
if self.debug_mode:
|
||
self.log(traceback.format_exc(), "ERROR")
|
||
return False
|
||
|
||
def extract_package(self, force=False):
|
||
self.last_resource_error = ""
|
||
if not force and self.reuse_cached_package():
|
||
return True
|
||
|
||
if not self.package_file.exists():
|
||
self.log(f"错误：未找到资源包 ({self.package_file})", "ERROR")
|
||
return False
|
||
if not self.extract_password:
|
||
self.log("请先获取资源包解压密码", "ERROR")
|
||
return False
|
||
if not os.path.exists(self.sz):
|
||
self.log(f"错误：未找到 7za.exe ({self.sz})", "ERROR")
|
||
return False
|
||
|
||
try:
|
||
self.temp_dir = self.get_cache_dir()
|
||
hidden_path = self.temp_dir.parent
|
||
hidden_path.mkdir(parents=True, exist_ok=True)
|
||
|
||
if self.temp_dir.exists():
|
||
shutil.rmtree(str(self.temp_dir), ignore_errors=True)
|
||
time.sleep(0.3)
|
||
self.temp_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
if sys.platform == "win32":
|
||
subprocess.run(f'attrib +h "{self.temp_dir.parent}"', shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||
subprocess.run(f'attrib +h "{self.temp_dir}"', shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||
|
||
self.log("正在解压资源包...", "INFO")
|
||
self.update_progress(0, 100, "资源加载中")
|
||
cmd = [
|
||
self.sz,
|
||
"x",
|
||
str(self.package_file),
|
||
f"-p{self.extract_password}",
|
||
f"-o{self.temp_dir}",
|
||
"-y",
|
||
]
|
||
if self._seven_zip_supports_progress_stream():
|
||
if self.debug_mode:
|
||
cmd.extend(["-bsp1", "-bse1"])
|
||
else:
|
||
cmd.extend(["-bsp1", "-bso0", "-bse1"])
|
||
if self.debug_mode:
|
||
self.log(f"7ZA CMD: {subprocess.list2cmdline(cmd)}", "CMD")
|
||
|
||
proc = subprocess.Popen(
|
||
cmd,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.STDOUT,
|
||
stdin=subprocess.DEVNULL,
|
||
creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0,
|
||
)
|
||
|
||
output = bytearray()
|
||
progress_window = bytearray()
|
||
last_percent = -1
|
||
while True:
|
||
chunk = proc.stdout.read(1) if proc.stdout else b""
|
||
if not chunk:
|
||
if proc.poll() is not None:
|
||
break
|
||
time.sleep(0.05)
|
||
continue
|
||
|
||
output.extend(chunk)
|
||
if not self.debug_mode and len(output) > 60000:
|
||
del output[:-60000]
|
||
progress_window.extend(chunk)
|
||
if len(progress_window) > 1024:
|
||
del progress_window[:-1024]
|
||
|
||
matches = re.findall(rb"(\d{1,3})%", bytes(progress_window[-512:]))
|
||
if matches:
|
||
percent = min(100, int(matches[-1]))
|
||
if percent != last_percent:
|
||
last_percent = percent
|
||
self.update_progress(percent, 100, "资源加载中")
|
||
|
||
return_code = proc.wait()
|
||
decoded_output = self._decode_7z_output(bytes(output))
|
||
if self.debug_mode:
|
||
level = "CMD" if return_code == 0 else "ERROR"
|
||
self.log(f"7ZA RET: {return_code}", level)
|
||
if decoded_output.strip():
|
||
self.log(f"7ZA OUTPUT:\n{decoded_output.strip()}", level)
|
||
if return_code != 0:
|
||
error_msg = self.detect_extract_error(decoded_output) or f"解压失败: {decoded_output.strip()[:300]}"
|
||
self.last_resource_error = error_msg
|
||
self.log(error_msg, "ERROR")
|
||
self.clear_cached_package()
|
||
return False
|
||
|
||
self.update_progress(100, 100, "资源加载完成")
|
||
self.apps_dir = self.find_apps_dir()
|
||
if not self.apps_dir:
|
||
self.log("未找到 apps 目录", "ERROR")
|
||
self.clear_cached_package()
|
||
return False
|
||
self.manager_apk = self.find_manager_apk()
|
||
if not self.manager_apk:
|
||
self.log(f"未找到 {self.MANAGER_APK_NAME}", "ERROR")
|
||
self.clear_cached_package()
|
||
return False
|
||
|
||
ok, apk_count, reason = self.validate_cached_resources()
|
||
if not ok:
|
||
self.last_resource_error = f"解压后的资源无效: {reason}。已停止推送，请检查解压密码或资源包。"
|
||
self.log(self.last_resource_error, "ERROR")
|
||
self.clear_cached_package()
|
||
return False
|
||
self.log(
|
||
f"资源准备完成，apps 目录包含 {apk_count} 个 APK，已找到 {self.MANAGER_APK_NAME}",
|
||
"SUCCESS",
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
self.last_resource_error = f"资源准备失败: {str(e)}"
|
||
self.log(self.last_resource_error, "ERROR")
|
||
self.clear_cached_package()
|
||
return False
|
||
|
||
def find_apps_dir(self):
|
||
if not self.temp_dir:
|
||
return None
|
||
candidates = [candidate for candidate in self.temp_dir.rglob("apps") if candidate.is_dir()]
|
||
for candidate in candidates:
|
||
if (candidate.parent / self.MANAGER_APK_NAME).is_file():
|
||
return candidate
|
||
return candidates[0] if candidates else None
|
||
|
||
def find_manager_apk(self):
|
||
if not self.temp_dir or not self.apps_dir:
|
||
return None
|
||
sibling = self.apps_dir.parent / self.MANAGER_APK_NAME
|
||
if sibling.is_file():
|
||
return sibling
|
||
return None
|
||
|
||
def collect_apps_apks(self):
|
||
if not self.apps_dir:
|
||
return []
|
||
return sorted(self.apps_dir.rglob("*.apk"), key=lambda p: str(p.relative_to(self.apps_dir)).lower())
|
||
|
||
def ensure_device_connected(self):
|
||
ok, output = self.run_adb_command("adb -d devices", log_debug=False)
|
||
if not ok:
|
||
self.log(f"设备检查失败: {output}", "ERROR")
|
||
return False
|
||
lines = output.splitlines()
|
||
devices = [
|
||
line for line in lines[1:]
|
||
if line.strip() and "\tdevice" in line and "offline" not in line
|
||
]
|
||
self.device_connected = bool(devices)
|
||
return self.device_connected
|
||
|
||
def open_android_settings(self):
|
||
def do_open():
|
||
try:
|
||
self.log("正在检查 ADB 设备...", "INFO")
|
||
device_ready = self.ensure_device_connected()
|
||
if not device_ready and not self.debug_mode:
|
||
self.log("未检测到可用设备，无法打开原生设置", "ERROR")
|
||
self.run_on_ui_thread(messagebox.showerror, "设备未连接", "未检测到可用设备")
|
||
return
|
||
if not device_ready and self.debug_mode:
|
||
self.log("调试模式：设备检查未通过，继续执行打开原生设置命令", "WARNING")
|
||
|
||
self.log("正在打开安卓原生设置...", "INFO")
|
||
ok, output = self.run_adb_command(
|
||
"adb -d exec-out am start -a android.settings.SETTINGS"
|
||
)
|
||
if ok:
|
||
self.log("已发送打开安卓原生设置命令", "SUCCESS")
|
||
else:
|
||
self.log(f"打开安卓原生设置失败: {output}", "ERROR")
|
||
self.run_on_ui_thread(messagebox.showerror, "打开失败", output or "打开安卓原生设置失败")
|
||
except Exception as e:
|
||
self.log(f"打开安卓原生设置异常: {e}", "ERROR")
|
||
if self.debug_mode:
|
||
self.log(traceback.format_exc(), "ERROR")
|
||
self.run_on_ui_thread(messagebox.showerror, "打开异常", str(e))
|
||
|
||
threading.Thread(target=do_open, daemon=True).start()
|
||
|
||
def reboot_device(self):
|
||
if not messagebox.askyesno("确认重启", "确定要重启设备吗？"):
|
||
return
|
||
|
||
def do_reboot():
|
||
try:
|
||
self.log("正在检查 ADB 设备...", "INFO")
|
||
device_ready = self.ensure_device_connected()
|
||
if not device_ready and not self.debug_mode:
|
||
self.log("未检测到可用设备，无法重启", "ERROR")
|
||
self.run_on_ui_thread(messagebox.showerror, "设备未连接", "未检测到可用设备")
|
||
return
|
||
if not device_ready and self.debug_mode:
|
||
self.log("调试模式：设备检查未通过，继续执行 adb -d reboot", "WARNING")
|
||
|
||
self.log("正在重启设备...", "INFO")
|
||
ok, output = self.run_adb_command("adb -d reboot")
|
||
if ok:
|
||
self.device_connected = False
|
||
self.run_on_ui_thread(
|
||
lambda: self.device_status_label.config(
|
||
text="正在重启",
|
||
fg=self.colors["warning"],
|
||
)
|
||
)
|
||
self.log("设备正在重启...", "SUCCESS")
|
||
self.run_on_ui_thread(messagebox.showinfo, "重启设备", "设备正在重启")
|
||
else:
|
||
self.log(f"重启设备失败: {output}", "ERROR")
|
||
self.run_on_ui_thread(messagebox.showerror, "重启失败", output or "adb -d reboot 执行失败")
|
||
except Exception as e:
|
||
self.log(f"重启过程异常: {e}", "ERROR")
|
||
if self.debug_mode:
|
||
self.log(traceback.format_exc(), "ERROR")
|
||
self.run_on_ui_thread(messagebox.showerror, "重启异常", str(e))
|
||
|
||
threading.Thread(target=do_reboot, daemon=True).start()
|
||
|
||
def ensure_remote_dir(self, remote_dir, label):
|
||
ok, output = self.run_adb_command(
|
||
f"adb -d exec-out mkdir {remote_dir.rstrip('/')}"
|
||
)
|
||
if ok:
|
||
return True
|
||
if "File exists" in output or "exists" in output:
|
||
self.log(f"{label} 目录已存在，继续推送", "INFO")
|
||
return True
|
||
self.log(f"创建 {label} 目录失败: {output}", "ERROR")
|
||
return False
|
||
|
||
def push_third_party_apks(self):
|
||
files = filedialog.askopenfilenames(
|
||
title="选择需要推送的 APK 文件",
|
||
filetypes=[("APK 文件", "*.apk"), ("所有文件", "*.*")],
|
||
)
|
||
if not files:
|
||
return
|
||
|
||
apk_paths = [Path(file) for file in files]
|
||
invalid_files = [path for path in apk_paths if path.suffix.lower() != ".apk"]
|
||
if invalid_files:
|
||
messagebox.showwarning("文件类型错误", "请选择 APK 文件")
|
||
return
|
||
|
||
if not messagebox.askyesno(
|
||
"确认推送",
|
||
f"将推送 {len(apk_paths)} 个 APK 到：{self.THIRD_APPS_DIR}\n\n是否开始？",
|
||
):
|
||
return
|
||
|
||
def do_push():
|
||
try:
|
||
self.log(f"开始推送第三方 APK，目标目录 {self.THIRD_APPS_DIR}", "INFO")
|
||
self.log("正在检查 ADB 设备...", "INFO")
|
||
device_ready = self.ensure_device_connected()
|
||
if not device_ready and not self.debug_mode:
|
||
self.log("未检测到可用设备，无法推送第三方 APK", "ERROR")
|
||
self.run_on_ui_thread(messagebox.showerror, "设备未连接", "未检测到可用设备")
|
||
return
|
||
if not device_ready and self.debug_mode:
|
||
self.log("调试模式：设备检查未通过，继续执行第三方 APK 推送", "WARNING")
|
||
|
||
self.log("正在创建 Third-apps 目录...", "INFO")
|
||
if not self.ensure_remote_dir(self.THIRD_APPS_DIR, "Third-apps"):
|
||
self.run_on_ui_thread(messagebox.showerror, "推送失败", "无法创建 Third-apps 目录")
|
||
return
|
||
|
||
total = len(apk_paths)
|
||
success_count = 0
|
||
for i, apk in enumerate(apk_paths, 1):
|
||
self.update_progress(i - 1, total, "正在推送第三方 APK")
|
||
target = self.THIRD_APPS_DIR + apk.name
|
||
ok, output = self.run_adb_command(f'adb -d push "{apk}" "{target}"')
|
||
if ok:
|
||
success_count += 1
|
||
self.log(f"已推送第三方 APK: {apk.name}", "SUCCESS")
|
||
else:
|
||
self.log(f"推送第三方 APK 失败 {apk.name}: {output}", "ERROR")
|
||
self.update_progress(i, total, "正在推送第三方 APK")
|
||
|
||
if success_count == total:
|
||
self.log(f"第三方 APK 推送完成，共 {total} 个", "SUCCESS")
|
||
self.run_on_ui_thread(
|
||
messagebox.showinfo,
|
||
"推送完成",
|
||
f"成功推送 {total} 个第三方 APK",
|
||
)
|
||
elif success_count > 0:
|
||
self.log(f"第三方 APK 部分推送成功（{success_count}/{total}）", "WARNING")
|
||
self.run_on_ui_thread(
|
||
messagebox.showwarning,
|
||
"部分成功",
|
||
f"成功: {success_count}\n失败: {total - success_count}",
|
||
)
|
||
else:
|
||
self.log("第三方 APK 推送失败，所有文件均未成功", "ERROR")
|
||
self.run_on_ui_thread(messagebox.showerror, "推送失败", "所有第三方 APK 均未成功")
|
||
except Exception as e:
|
||
self.log(f"第三方 APK 推送过程异常: {e}", "ERROR")
|
||
if self.debug_mode:
|
||
self.log(traceback.format_exc(), "ERROR")
|
||
self.run_on_ui_thread(messagebox.showerror, "推送异常", str(e))
|
||
|
||
threading.Thread(target=do_push, daemon=True).start()
|
||
|
||
def push_language_package(self):
|
||
if not self.debug_password:
|
||
messagebox.showwarning("请先查询密码", "请先输入 VIN 查询调试密码")
|
||
return
|
||
|
||
if not messagebox.askyesno(
|
||
"确认推送",
|
||
f"apps 内 APK 将逐个推送到：{self.APPS_DIR}\n"
|
||
f"{self.MANAGER_APK_NAME} 将推送到：{self.DOWNLOAD_DIR}\n\n是否开始？",
|
||
):
|
||
return
|
||
self.log(f"开始推送资源，目标目录 {self.DOWNLOAD_DIR}", "INFO")
|
||
|
||
def do_push():
|
||
try:
|
||
if self.debug_mode:
|
||
self.log(
|
||
f"DEBUG: debug_password_set={bool(self.debug_password)}, "
|
||
f"extract_password_set={bool(self.extract_password)}, vin={self.vin}, "
|
||
f"package={self.package_file}, 7za={self.sz}, adb={self.adb}",
|
||
"CMD",
|
||
)
|
||
|
||
self.log("正在检查 ADB 设备...", "INFO")
|
||
device_ready = self.ensure_device_connected()
|
||
if not device_ready and not self.debug_mode:
|
||
self.log("未检测到可用设备，请确认 USB 调试连接", "ERROR")
|
||
self.run_on_ui_thread(messagebox.showerror, "设备未连接", "未检测到可用设备")
|
||
return
|
||
if not device_ready and self.debug_mode:
|
||
self.log("调试模式：设备检查未通过，继续执行以捕获 adb push 输出", "WARNING")
|
||
|
||
if device_ready:
|
||
self.run_on_ui_thread(
|
||
lambda: self.device_status_label.config(
|
||
text="已连接",
|
||
fg=self.colors["success"],
|
||
)
|
||
)
|
||
|
||
if not self.reuse_cached_package():
|
||
if not self.fetch_package_password():
|
||
self.run_on_ui_thread(messagebox.showerror, "资源准备失败", "无法获取资源包解压密码")
|
||
return
|
||
|
||
if not self.extract_package():
|
||
self.run_on_ui_thread(
|
||
messagebox.showerror,
|
||
"资源准备失败",
|
||
self.last_resource_error or "请检查 package.bin 或资源包解压密码",
|
||
)
|
||
return
|
||
|
||
apps_apks = self.collect_apps_apks()
|
||
if not apps_apks or not self.manager_apk:
|
||
self.log("没有找到完整的推送资源", "ERROR")
|
||
self.run_on_ui_thread(messagebox.showerror, "推送失败", "没有找到完整的推送资源")
|
||
return
|
||
if self.debug_mode:
|
||
self.log(
|
||
"待推送 apps APK:\n" + "\n".join(str(apk) for apk in apps_apks)
|
||
+ f"\n{self.MANAGER_APK_NAME}: {self.manager_apk}",
|
||
"CMD",
|
||
)
|
||
|
||
self.log("正在创建 apps 目录...", "INFO")
|
||
if not self.ensure_remote_dir(self.APPS_DIR, "apps"):
|
||
self.run_on_ui_thread(messagebox.showerror, "推送失败", "无法创建 apps 目录")
|
||
return
|
||
|
||
total = len(apps_apks) + 1
|
||
success_count = 0
|
||
for i, apk in enumerate(apps_apks, 1):
|
||
self.update_progress(i - 1, total, "正在推送 apps APK")
|
||
target = self.APPS_DIR + apk.name
|
||
ok, output = self.run_adb_command(f'adb -d push "{apk}" "{target}"')
|
||
if ok:
|
||
success_count += 1
|
||
self.log(f"已推送 apps APK: {apk.name}", "SUCCESS")
|
||
else:
|
||
self.log(f"推送 apps APK 失败 {apk.name}: {output}", "ERROR")
|
||
self.update_progress(i, total, "正在推送 apps APK")
|
||
|
||
manager_index = len(apps_apks) + 1
|
||
self.update_progress(manager_index - 1, total, "正在推送应用管家")
|
||
manager_target = self.DOWNLOAD_DIR + self.manager_apk.name
|
||
ok, output = self.run_adb_command(f'adb -d push "{self.manager_apk}" "{manager_target}"')
|
||
if ok:
|
||
success_count += 1
|
||
self.log(f"已推送: {self.manager_apk.name}", "SUCCESS")
|
||
else:
|
||
self.log(f"推送失败 {self.manager_apk.name}: {output}", "ERROR")
|
||
self.update_progress(manager_index, total, "正在推送应用管家")
|
||
|
||
if success_count == total:
|
||
self.log("推送完成，apps APK 和应用管家 APK 已写入 Download", "SUCCESS")
|
||
self.log("正在打开文件管理...", "INFO")
|
||
ok, output = self.run_adb_command(
|
||
"adb -d exec-out am start -n com.android.documentsui/.files.FilesActivity"
|
||
)
|
||
if ok:
|
||
self.log("已发送打开文件管理命令", "SUCCESS")
|
||
else:
|
||
self.log(f"打开文件管理失败: {output}", "WARNING")
|
||
self.run_on_ui_thread(
|
||
messagebox.showinfo,
|
||
"推送完成",
|
||
"apps APK 和应用管家 APK 已推送完成，正在打开文件管理",
|
||
)
|
||
elif success_count > 0:
|
||
self.log(f"部分推送成功（{success_count}/{total}）", "WARNING")
|
||
self.run_on_ui_thread(
|
||
messagebox.showwarning,
|
||
"部分成功",
|
||
f"成功: {success_count}\n失败: {total - success_count}",
|
||
)
|
||
else:
|
||
self.log("推送失败，所有资源均未成功", "ERROR")
|
||
self.run_on_ui_thread(messagebox.showerror, "推送失败", "所有资源均未成功")
|
||
except Exception as e:
|
||
self.log(f"推送过程异常: {e}", "ERROR")
|
||
if self.debug_mode:
|
||
self.log(traceback.format_exc(), "ERROR")
|
||
self.run_on_ui_thread(messagebox.showerror, "推送异常", str(e))
|
||
|
||
threading.Thread(target=do_push, daemon=True).start()
|
||
|
||
def run(self):
|
||
self.root.mainloop()
|
||
|
||
|
||
def main():
|
||
if sys.version_info < (3, 6):
|
||
print("错误：需要Python 3.6或更高版本")
|
||
sys.exit(1)
|
||
|
||
try:
|
||
app = UNIZLanguageGUI()
|
||
app.run()
|
||
except Exception as e:
|
||
print(f"启动失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
messagebox.showerror("错误", f"程序启动失败: {e}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|