Files
language-installer/A07/Qiyuan_A07_Multi-lan-installer.py
T
2026-07-07 14:37:04 +08:00

2920 lines
133 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import atexit
import shlex
import sys
import subprocess
import json
import re
import threading
import tkinter as tk
from tkinter import ttk, scrolledtext, filedialog, messagebox, simpledialog
from pathlib import Path
from urllib.request import urlopen, Request
from urllib.error import URLError, HTTPError
from urllib.parse import urlencode
from datetime import datetime
import zipfile
try:
import pyzipper
except ImportError:
pyzipper = None
import shutil
import time
STARTUP_T = {
'zh': {
'python_version_error': '错误:需要Python 3.6或更高版本',
'startup_failed_title': '错误',
'startup_failed_console': '启动失败: {error}',
'startup_failed_dialog': '程序启动失败: {error}',
},
'en': {
'python_version_error': 'Error: Python 3.6 or later is required',
'startup_failed_title': 'Error',
'startup_failed_console': 'Startup failed: {error}',
'startup_failed_dialog': 'Program startup failed: {error}',
}
}
def startup_t(key):
lang = 'en' if os.environ.get('LANGUAGE_INSTALLER_LANG', '').lower().startswith('en') else 'zh'
return STARTUP_T.get(lang, STARTUP_T['zh']).get(key, key)
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)
def set_windows_app_user_model_id():
if sys.platform != 'win32':
return
try:
import ctypes
ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID(
"yibin.keyi.qiyuan.a07.language.installer"
)
except Exception:
pass
class ADKAPKGUI:
A07_SHELL_PASSWORD = "omo75A322@"
SYSTEM_MOUNT_CANDIDATES = ("/system", "/system_root", "/")
def __init__(self):
set_windows_app_user_model_id()
self.root = tk.Tk()
self.root.title("")
self.root.geometry("650x640")
self.root.resizable(True, True)
self.set_window_icon()
# 设置颜色主题
self.colors_dark = {
'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.colors_light = {
'bg_dark': '#f5f5f5',
'bg_light': '#ffffff',
'accent': '#6c5ce7',
'accent_hover': '#5b4bc4',
'success': '#00b894',
'error': '#d63031',
'warning': '#e17055',
'info': '#0984e3',
'text': '#2d3436',
'text_secondary': '#636e72',
'border': '#dfe6e9'
}
self.colors = dict(self.colors_dark)
self.theme = 'dark'
# 多语言
self.lang = 'zh'
self.T = {
'zh': {
'title': '启源A07多语言安装',
'btn_root': '🔓 获取权限',
'btn_push': '📦 刷入语言包',
'btn_install': '📱 安装App',
'btn_language': '🌐 语言设置',
'btn_timezone': '⏰ 时区设置',
'btn_settings': '⚙️ 安卓设置',
'btn_reboot': '🔄 重启设备',
'btn_clear_cache': '🧹 清理缓存',
'btn_clear_log': '🗑 清空日志',
'device_label': '设备:',
'vin_label': 'VIN码:',
'auth_label': '授权:',
'log_title': '📋 运行日志',
'status_ready': '就绪',
'status_connected': '已连接',
'status_disconnected': '未连接',
'status_detecting': '未检测',
'vin_none': '未获取',
'auth_none': '未验证',
'auth_yes': '已授权',
'auth_no': '未授权',
'btn_refresh': '🔄 检查',
'theme_dark': '🌙 暗色',
'theme_light': '☀️ 亮色',
'lang_zh': '中文',
'lang_en': 'English',
'switch_lang': '语言 / Language',
'switch_theme': '切换主题',
'key_query_label': '登录密码获取:',
'vin_placeholder': '输入VIN或者VIN后六位',
'auth_code_placeholder': '输入授权码',
'btn_query_pwd': '获取密码',
'key_need_input': '请输入VIN和授权码',
'key_querying': '正在获取密码...',
'key_query_success': '密码: {password}',
'key_query_failed': '失败: {message}',
'key_request_failed': '请求失败',
'key_dial_tip': '打开车辆拨号App,拨号*#*#666,获取授权码。',
'tip_1': '1. 安装语言过程中请保持车辆和电脑的电量充足,不可中途停止。',
'tip_2': '2. 获取权限以后,车辆自动重启以后再进入语言刷入。',
'tip_3': '3. 部分语言需要重启后生效,可以一切工作完成以后再重启。',
'dialog_warning': '警告',
'dialog_error': '错误',
'dialog_info': '提示',
'dialog_success': '成功',
'dialog_device_not_connected_title': '设备未连接',
'dialog_device_not_connected_body': '请先连接设备并点击「检查」按钮刷新状态!',
'log_cleared': '日志已清空',
'log_device_connected': '设备已连接',
'log_device_disconnected': '设备已断开连接',
'log_device_not_connected': '设备未连接',
'log_current_vin': '当前车辆VIN: {vin}',
'log_vin_unavailable': '无法获取VIN',
'log_refresh_failed': '刷新设备状态失败: {error}',
'log_root_ready': 'A07 system 分区已解锁,可以开始刷入',
'log_root_failed': '获取权限失败',
'log_cache_invalid': '已解压缓存无效: {reason}',
'log_no_available_apk': '未找到可用 APK',
'log_zero_apk': '发现 0KB APK: {preview}{suffix}',
'extract_wrong_password': '解压密码错误,请重新确认 package.bin 密码',
'extract_data_error': '资源包数据错误,可能是密码错误或 package.bin 损坏',
'extract_broken': '资源包损坏或不完整,请检查 package.bin',
'extract_prepare_failed_detail': '资源准备失败: {error}',
'extract_prepare_failed': '资源准备失败,请检查解压密码是否正确',
'progress_resource_loading': '资源加载中...',
'progress_resource_done': '资源加载完成',
'log_package_missing_path': '错误:未找到资源包 ({path})',
'log_extract_password_missing': '错误:解压密码未设置',
'log_7za_missing': '错误:未找到 7za.exe ({path})',
'log_resource_preparing': '正在准备资源包...',
'log_resource_dir_missing': '警告:未找到对应目录',
'log_resource_invalid': '解压后的资源无效: {reason}。已停止刷入,请检查解压密码或资源包。',
'log_resource_ready_debug': '资源准备完成 (app: {app}, priv-app: {priv}, system_ext: {system_ext})',
'log_resource_ready': '资源准备完成',
'log_resource_prepare_exception': '资源准备失败: {error}',
'log_resource_prepare_retry': '资源准备失败,请检查网络连接后重试',
'log_package_missing': '未找到资源包文件',
'log_adb_missing': '未找到adb命令,请将ADB文件放入本目录',
'log_cache_invalid_cleaned': '缓存资源无效,已清理: {reason}',
'log_debug_skip_auth': '调试模式: 跳过授权验证',
'log_auth_checking': '正在验证授权...',
'log_auth_ok': '授权验证通过',
'log_vehicle_name': '车辆名称: {vehicle_name}',
'log_auth_failed': '授权验证失败',
'log_connect_adb_first': '请先连接adb',
'log_prepare_failed_no_vehicle': '数据准备失败: 未获取到车型',
'log_prepare_failed_reason': '数据准备失败: {reason}',
'log_prepare_failed_error': '数据准备失败: {error}',
'err_unknown': '未知错误',
'push_fail': 'push失败: {error}',
'copy_fail': 'cp失败: {error}',
'dialog_need_vin': '请先刷新设备状态并获取VIN码',
'dialog_flash_warning_title': '重要提示',
'dialog_flash_warning_body': '刷入过程中请勿:\n ● 重启车机\n ● 退出本程序\n ● 关闭电脑\n\n否则可能导致车机系统损坏!',
'log_flash_start': '开始刷入语言包,请勿断电或重启电脑和车机。',
'dialog_auth_failed_title': '授权失败',
'dialog_auth_failed_body': '设备未授权',
'dialog_resource_failed_body': '资源准备失败!',
'dialog_resource_dir_missing': '资源目录未找到',
'log_system_unlock_failed': 'system 分区解锁失败',
'log_lang_pkg_missing': '未找到语言包文件',
'log_readonly_system': 'system 分区仍为只读,请重新点击「获取权限」后再试',
'progress_flashing': '正在刷入...',
'progress_flash_done': '刷入完成',
'progress_aborted': '已终止',
'log_flash_done': '刷入完成,共 {total} 个语言包',
'log_flash_reboot_required': '语言包已刷入完成,请务必重启设备,system/priv-app 需要开机扫描后才会显示',
'log_flash_partial': '部分刷入成功({success}/{total}',
'log_flash_scan_required': '已刷入的系统应用需要重启设备后才会显示',
'dialog_select_apk_folder': '选择包含APK文件的文件夹',
'dialog_no_apk_in_folder': '所选文件夹中没有APK文件!',
'dialog_confirm_install_title': '确认安装',
'dialog_confirm_install_folder': '找到 {count} 个APK文件\n\n是否开始批量安装?',
'log_batch_install_start': '开始批量安装 {count} 个APK...',
'progress_installing': '安装中...',
'progress_installing_apk': '安装中 ({apk})',
'progress_install_done': '安装完成',
'log_install_all_success': '安装完成:全部 {count} 个成功',
'dialog_install_done_title': '安装完成',
'dialog_install_all_success': '成功安装 {count} 个APK',
'log_install_partial': '安装完成:{success}/{total} 成功',
'dialog_install_partial_title': '部分成功',
'dialog_install_partial_body': '成功: {success}\n失败: {failed}',
'log_install_failed': '安装失败',
'dialog_install_failed_title': '安装失败',
'dialog_install_all_failed': '所有APK安装失败!',
'log_install_exception': '安装过程异常: {error}',
'dialog_install_exception': '安装过程异常:{error}',
'dialog_select_apk_file': '选择APK文件',
'filetype_apk': 'APK文件',
'filetype_all': '所有文件',
'progress_done': '完成',
'log_single_install_success': '✓ 安装成功',
'log_single_install_failed': '✗ 安装失败',
'quick_lang_title': '快捷语言设置',
'quick_lang_header': '选择目标语言',
'quick_lang_hint': '点击按钮即可将系统语言切换为对应语言,重启后生效',
'quick_lang_system_button': '⚙️ 打开系统语言设置(手动选择)',
'quick_lang_zh': '🇨🇳 中文',
'quick_lang_en': '🇺🇸 英语',
'quick_lang_ru': '🇷🇺 俄语',
'quick_lang_fr': '🇫🇷 法语',
'quick_lang_es': '🇪🇸 西班牙语',
'quick_lang_pt': '🇵🇹 葡萄牙语',
'quick_lang_it': '🇮🇹 意大利语',
'quick_lang_ar': '🇸🇦 阿拉伯语',
'log_setting_language': '正在设置系统语言为: {language} ({locale})',
'log_language_set_success': '✓ 语言已设置为 {language}',
'dialog_language_success_title': '设置成功',
'dialog_language_success_body': '系统语言已设置为 {language}\n\n⚠️ 请重启设备使其生效。',
'log_language_set_failed': '✗ 语言设置失败: {error}',
'dialog_language_failed_title': '设置失败',
'dialog_language_failed_body': '语言设置失败!\n\n{error}',
'dialog_confirm_reboot_title': '确认重启',
'dialog_confirm_reboot_body': '确定要重启设备吗?',
'log_rebooting': '设备正在重启...',
'log_reboot_failed': '重启失败: {error}',
'dialog_clear_cache_title': '确认清理缓存',
'dialog_clear_cache_body': '将删除本地解压缓存目录:\n{cache_dir}\n\n下次刷入会重新解压 package.bin,是否继续?',
'log_clear_cache_cancelled': '已取消清理缓存',
'log_cache_cleared': '解压缓存已清理',
'dialog_cache_cleared_title': '清理完成',
'dialog_cache_cleared_body': '解压缓存已清理。',
'log_cache_clear_failed': '清理缓存失败: {error}',
'dialog_cache_clear_failed_title': '清理失败',
'dialog_cache_clear_failed_body': '清理缓存失败:{error}',
'log_key_success': '登录密码获取成功',
'log_key_failed': '登录密码获取失败: {message}',
'log_key_request_failed_detail': '登录密码请求失败: {error}',
'debug_title': '调试模式',
'debug_prompt': '请输入调试密码:',
'debug_status': '🔧 调试模式',
'log_debug_disabled': '调试模式已关闭',
'log_debug_enabled': '🔧 调试模式已开启 - 跳过授权和设备校验,显示详细ADB日志',
'debug_password_wrong': '密码错误',
'debug_extract_need_enable': '请先按 Ctrl+Shift+D 开启调试模式',
'debug_extract_title': '测试解压',
'debug_extract_prompt': '请输入 package.bin 解压密码:',
'debug_extract_success_title': '测试解压成功',
'debug_extract_success_body': '资源已解压到:\n{path}',
'debug_extract_failed_title': '测试解压失败',
'debug_extract_failed_body': '请查看日志中的 7za 输出',
'log_debug_extract_success': '测试解压成功',
'log_debug_extract_failed': '测试解压失败',
'log_apkpure_permission_ok': '已授予 APKPure 安装应用权限',
'log_apkpure_permission_partial': 'APKPure 安装应用权限部分失败: {details}',
'log_language_switched': '语言已切换为 {language}',
'language_name_zh': '中文',
'language_name_en': 'English',
'log_shell_login_try': '正在尝试交互式 adb shell 登录...',
'log_selinux_permissive': 'SELinux 已切换为宽松模式',
'log_system_rw_debug': '系统分区已重新挂载为可写: {mount_point}',
'log_builtin_cleaning': '正在删除预置应用目录...',
'log_builtin_deleted': '已删除: {path}',
'log_builtin_delete_failed': '删除失败: {path} ({error})',
'log_builtin_done_partial': '预置应用清理完成,{count} 个目录未成功',
'log_builtin_done': '预置应用清理完成',
'log_selinux_restored': 'SELinux 已恢复为 Enforcing',
'log_selinux_restore_failed': 'SELinux 恢复失败: {error}',
'startup_python_version_error': '错误:需要Python 3.6或更高版本',
'startup_failed_title': '错误',
'startup_failed_console': '启动失败: {error}',
'startup_failed_dialog': '程序启动失败: {error}',
},
'en': {
'title': 'Qiyuan A07 Multi-Language',
'btn_root': '🔓 Get Root',
'btn_push': '📦 Flash Lang Pkg',
'btn_install': '📱 Install App',
'btn_language': '🌐 Language',
'btn_timezone': '⏰ Timezone',
'btn_settings': '⚙️ Settings',
'btn_reboot': '🔄 Reboot',
'btn_clear_cache': '🧹 Clear Cache',
'btn_clear_log': '🗑 Clear Log',
'device_label': 'Device:',
'vin_label': 'VIN:',
'auth_label': 'Auth:',
'log_title': '📋 Log',
'status_ready': 'Ready',
'status_connected': 'Connected',
'status_disconnected': 'Disconnected',
'status_detecting': 'Detecting',
'vin_none': 'None',
'auth_none': 'Unknown',
'auth_yes': 'Authorized',
'auth_no': 'Unauthorized',
'btn_refresh': '🔄 Check',
'theme_dark': '🌙 Dark',
'theme_light': '☀️ Light',
'lang_zh': '中文',
'lang_en': 'English',
'switch_lang': 'Language',
'switch_theme': 'Theme',
'key_query_label': 'Login password:',
'vin_placeholder': 'Enter VIN or last 6 digits',
'auth_code_placeholder': 'Enter auth code',
'btn_query_pwd': 'Get Password',
'key_need_input': 'Enter VIN and auth code',
'key_querying': 'Getting password...',
'key_query_success': 'Password: {password}',
'key_query_failed': 'Failed: {message}',
'key_request_failed': 'Request failed',
'key_dial_tip': 'Open the vehicle Dialer app and dial *#*#666 to get the auth code.',
'tip_1': '1. Keep the vehicle and computer powered during installation. Do not stop midway.',
'tip_2': '2. After getting permission, wait for the vehicle system to reboot before flashing.',
'tip_3': '3. Some languages take effect after reboot. You can reboot after all tasks are complete.',
'dialog_warning': 'Warning',
'dialog_error': 'Error',
'dialog_info': 'Notice',
'dialog_success': 'Success',
'dialog_device_not_connected_title': 'Device Not Connected',
'dialog_device_not_connected_body': 'Connect the device, then click "Check" to refresh status.',
'log_cleared': 'Log cleared',
'log_device_connected': 'Device connected',
'log_device_disconnected': 'Device disconnected',
'log_device_not_connected': 'Device not connected',
'log_current_vin': 'Current VIN: {vin}',
'log_vin_unavailable': 'Unable to get VIN',
'log_refresh_failed': 'Failed to refresh device status: {error}',
'log_root_ready': 'A07 system partition is ready. You can start flashing.',
'log_root_failed': 'Failed to get permission',
'log_cache_invalid': 'Extracted cache is invalid: {reason}',
'log_no_available_apk': 'No available APK found',
'log_zero_apk': 'Found 0KB APK: {preview}{suffix}',
'extract_wrong_password': 'Wrong extraction password. Please confirm the package.bin password.',
'extract_data_error': 'Resource data error. The password may be wrong or package.bin may be damaged.',
'extract_broken': 'Resource package is damaged or incomplete. Please check package.bin.',
'extract_prepare_failed_detail': 'Resource preparation failed: {error}',
'extract_prepare_failed': 'Resource preparation failed. Please check the extraction password.',
'progress_resource_loading': 'Preparing resources...',
'progress_resource_done': 'Resources ready',
'log_package_missing_path': 'Error: resource package not found ({path})',
'log_extract_password_missing': 'Error: extraction password is not set',
'log_7za_missing': 'Error: 7za.exe not found ({path})',
'log_resource_preparing': 'Preparing resources...',
'log_resource_dir_missing': 'Warning: required resource folder not found',
'log_resource_invalid': 'Extracted resources are invalid: {reason}. Flashing stopped. Please check the password or package.',
'log_resource_ready_debug': 'Resources ready (app: {app}, priv-app: {priv}, system_ext: {system_ext})',
'log_resource_ready': 'Resources ready',
'log_resource_prepare_exception': 'Resource preparation failed: {error}',
'log_resource_prepare_retry': 'Resource preparation failed. Check the network connection and try again.',
'log_package_missing': 'Resource package file not found',
'log_adb_missing': 'ADB command not found. Put ADB files in this folder.',
'log_cache_invalid_cleaned': 'Invalid cached resources removed: {reason}',
'log_debug_skip_auth': 'Debug mode: skipped authorization check',
'log_auth_checking': 'Checking authorization...',
'log_auth_ok': 'Authorization verified',
'log_vehicle_name': 'Vehicle name: {vehicle_name}',
'log_auth_failed': 'Authorization failed',
'log_connect_adb_first': 'Please connect ADB first.',
'log_prepare_failed_no_vehicle': 'Data preparation failed: vehicle model not found',
'log_prepare_failed_reason': 'Data preparation failed: {reason}',
'log_prepare_failed_error': 'Data preparation failed: {error}',
'err_unknown': 'Unknown error',
'push_fail': 'Push failed: {error}',
'copy_fail': 'Copy failed: {error}',
'dialog_need_vin': 'Please refresh device status and get VIN first.',
'dialog_flash_warning_title': 'Important Notice',
'dialog_flash_warning_body': 'During flashing, do not:\n - reboot the vehicle system\n - exit this program\n - shut down the computer\n\nOtherwise the vehicle system may be damaged.',
'log_flash_start': 'Starting language installation. Do not power off or reboot the computer or vehicle system.',
'dialog_auth_failed_title': 'Authorization Failed',
'dialog_auth_failed_body': 'Device is not authorized',
'dialog_resource_failed_body': 'Resource preparation failed!',
'dialog_resource_dir_missing': 'Resource folder not found',
'log_system_unlock_failed': 'System partition unlock failed',
'log_lang_pkg_missing': 'Language package file not found',
'log_readonly_system': 'System partition is still read-only. Click "Get Root" and try again.',
'progress_flashing': 'Flashing...',
'progress_flash_done': 'Flash complete',
'progress_aborted': 'Aborted',
'log_flash_done': 'Flash complete. {total} language packages processed.',
'log_flash_reboot_required': 'Language packages have been flashed. Please reboot the device so system apps can be scanned.',
'log_flash_partial': 'Partially flashed ({success}/{total})',
'log_flash_scan_required': 'Flashed system apps need a reboot before they appear.',
'dialog_select_apk_folder': 'Select Folder Containing APK Files',
'dialog_no_apk_in_folder': 'No APK files found in the selected folder!',
'dialog_confirm_install_title': 'Confirm Install',
'dialog_confirm_install_folder': 'Found {count} APK files.\n\nStart batch installation?',
'log_batch_install_start': 'Starting batch install for {count} APKs...',
'progress_installing': 'Installing...',
'progress_installing_apk': 'Installing ({apk})',
'progress_install_done': 'Install complete',
'log_install_all_success': 'Install complete: all {count} succeeded',
'dialog_install_done_title': 'Install Complete',
'dialog_install_all_success': 'Successfully installed {count} APKs!',
'log_install_partial': 'Install complete: {success}/{total} succeeded',
'dialog_install_partial_title': 'Partial Success',
'dialog_install_partial_body': 'Success: {success}\nFailed: {failed}',
'log_install_failed': 'Install failed',
'dialog_install_failed_title': 'Install Failed',
'dialog_install_all_failed': 'All APK installations failed!',
'log_install_exception': 'Install error: {error}',
'dialog_install_exception': 'Install error: {error}',
'dialog_select_apk_file': 'Select APK File',
'filetype_apk': 'APK files',
'filetype_all': 'All files',
'progress_done': 'Done',
'log_single_install_success': '✓ Install succeeded',
'log_single_install_failed': '✗ Install failed',
'quick_lang_title': 'Quick Language Settings',
'quick_lang_header': 'Select Target Language',
'quick_lang_hint': 'Click a button to switch the system language. Reboot to apply.',
'quick_lang_system_button': '⚙️ Open System Language Settings',
'quick_lang_zh': '🇨🇳 Chinese',
'quick_lang_en': '🇺🇸 English',
'quick_lang_ru': '🇷🇺 Russian',
'quick_lang_fr': '🇫🇷 French',
'quick_lang_es': '🇪🇸 Spanish',
'quick_lang_pt': '🇵🇹 Portuguese',
'quick_lang_it': '🇮🇹 Italian',
'quick_lang_ar': '🇸🇦 Arabic',
'log_setting_language': 'Setting system language to: {language} ({locale})',
'log_language_set_success': '✓ Language set to {language}',
'dialog_language_success_title': 'Success',
'dialog_language_success_body': 'System language has been set to {language}.\n\n⚠️ Please reboot the device to apply it.',
'log_language_set_failed': '✗ Failed to set language: {error}',
'dialog_language_failed_title': 'Failed',
'dialog_language_failed_body': 'Language setting failed!\n\n{error}',
'dialog_confirm_reboot_title': 'Confirm Reboot',
'dialog_confirm_reboot_body': 'Are you sure you want to reboot the device?',
'log_rebooting': 'Device is rebooting...',
'log_reboot_failed': 'Reboot failed: {error}',
'dialog_clear_cache_title': 'Confirm Cache Cleanup',
'dialog_clear_cache_body': 'This will delete the local extracted cache folder:\n{cache_dir}\n\nNext flash will extract package.bin again. Continue?',
'log_clear_cache_cancelled': 'Cache cleanup cancelled',
'log_cache_cleared': 'Extracted cache cleared',
'dialog_cache_cleared_title': 'Cleanup Complete',
'dialog_cache_cleared_body': 'Extracted cache has been cleared.',
'log_cache_clear_failed': 'Failed to clear cache: {error}',
'dialog_cache_clear_failed_title': 'Cleanup Failed',
'dialog_cache_clear_failed_body': 'Failed to clear cache: {error}',
'log_key_success': 'Login password retrieved',
'log_key_failed': 'Failed to get login password: {message}',
'log_key_request_failed_detail': 'Login password request failed: {error}',
'debug_title': 'Debug Mode',
'debug_prompt': 'Enter debug password:',
'debug_status': '🔧 Debug Mode',
'log_debug_disabled': 'Debug mode disabled',
'log_debug_enabled': '🔧 Debug mode enabled - authorization/device checks are skipped and detailed ADB logs are shown',
'debug_password_wrong': 'Wrong password',
'debug_extract_need_enable': 'Press Ctrl+Shift+D to enable debug mode first.',
'debug_extract_title': 'Extraction Test',
'debug_extract_prompt': 'Enter package.bin extraction password:',
'debug_extract_success_title': 'Extraction Test Succeeded',
'debug_extract_success_body': 'Resources extracted to:\n{path}',
'debug_extract_failed_title': 'Extraction Test Failed',
'debug_extract_failed_body': 'Check the 7za output in the log.',
'log_debug_extract_success': 'Extraction test succeeded',
'log_debug_extract_failed': 'Extraction test failed',
'log_apkpure_permission_ok': 'APKPure install permission granted',
'log_apkpure_permission_partial': 'APKPure install permission partly failed: {details}',
'log_language_switched': 'Language switched to {language}',
'language_name_zh': '中文',
'language_name_en': 'English',
'log_shell_login_try': 'Trying interactive adb shell login...',
'log_selinux_permissive': 'SELinux switched to permissive mode',
'log_system_rw_debug': 'System partition remounted writable: {mount_point}',
'log_builtin_cleaning': 'Cleaning preinstalled app entries...',
'log_builtin_deleted': 'Deleted: {path}',
'log_builtin_delete_failed': 'Delete failed: {path} ({error})',
'log_builtin_done_partial': 'Preinstalled app cleanup completed, {count} entries failed',
'log_builtin_done': 'Preinstalled app cleanup completed',
'log_selinux_restored': 'SELinux restored to Enforcing',
'log_selinux_restore_failed': 'Failed to restore SELinux: {error}',
'startup_python_version_error': 'Error: Python 3.6 or later is required',
'startup_failed_title': 'Error',
'startup_failed_console': 'Startup failed: {error}',
'startup_failed_dialog': 'Program startup failed: {error}',
}
}
self.root.title(self.t('title'))
# 从 exe/py 所在目录查找资源文件
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.extract_password = None
self.apps_dir = None
self.priv_apps_dir = None
self.system_ext_dir = None
self.temp_dir = None
self.api_url = "https://api.changan.softwindy.cn/api/authorizations/auth-check"
self.debug_password_api_url = "https://api.changan.softwindy.cn/api/authorizations/verify-debug-mode-password"
self.vin = None
self.vehicle_name = ""
self.device_connected = False
self._refreshing = False # 防止并发刷新
self.debug_mode = False # 调试模式
self.shell_password = self.A07_SHELL_PASSWORD
self.system_mount_point = None
self.selinux_restore_mode = None
self.shell_login_verified = False
self._pwd_result_state = None
# 设置样式
self.setup_styles()
self.setup_ui()
self.root.after(200, self.set_window_icon)
self.center_window()
# 检查环境
self.check_environment()
# 启动设备状态监控
self.start_device_monitor()
self.root.protocol("WM_DELETE_WINDOW", self.on_close)
atexit.register(self.cleanup_cache_on_exit)
def set_window_icon(self):
"""Set runtime Tk titlebar/taskbar icon; PyInstaller --icon only sets the exe file icon."""
try:
icon_path = find_resource("app.ico")
if icon_path.exists():
self.root.iconbitmap(str(icon_path))
self._set_windows_hwnd_icon(icon_path)
except Exception:
pass
def _set_windows_hwnd_icon(self, icon_path):
if sys.platform != 'win32':
return
try:
import ctypes
user32 = ctypes.windll.user32
hwnd = self.root.winfo_id()
image_icon = 1
lr_loadfromfile = 0x00000010
wm_seticon = 0x0080
icon_small = 0
icon_big = 1
path = str(icon_path)
small = user32.LoadImageW(None, path, image_icon, 16, 16, lr_loadfromfile)
big = user32.LoadImageW(None, path, image_icon, 32, 32, lr_loadfromfile)
if small:
user32.SendMessageW(hwnd, wm_seticon, icon_small, small)
if big:
user32.SendMessageW(hwnd, wm_seticon, icon_big, big)
except Exception:
pass
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('TLabelframe', background=self.colors['bg_dark'], foreground=self.colors['text'])
style.configure('TLabelframe.Label', background=self.colors['bg_dark'], foreground=self.colors['accent'])
# 配置进度条
style.configure('TProgressbar',
background=self.colors['accent'],
troughcolor=self.colors['bg_light'],
borderwidth=0)
def setup_ui(self):
"""设置UI界面"""
# 配置根窗口
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=45)
title_frame.pack(fill=tk.X, pady=(0, 10))
title_frame.pack_propagate(False)
title_frame.grid_columnconfigure(0, weight=1)
title_frame.grid_columnconfigure(1, weight=0)
title_frame.grid_columnconfigure(2, weight=1)
# 标题
self.title_label = tk.Label(title_frame,
text=f"🚀 {self.t('title')}",
font=('Microsoft YaHei', 18, 'bold'),
fg=self.colors['accent'],
bg=self.colors['bg_dark'])
self.title_label.grid(row=0, column=1, pady=(3, 0))
self.btn_lang_switch = tk.Button(title_frame,
text=self.t('lang_en') if self.lang == 'zh' else self.t('lang_zh'),
command=self.toggle_lang,
font=('Microsoft YaHei', 10, 'bold'),
fg='white',
bg=self.colors['accent'],
activeforeground='white',
activebackground=self.colors['accent_hover'],
relief=tk.FLAT,
cursor='hand2',
width=9,
height=1)
self.btn_lang_switch.grid(row=0, column=2, sticky='e', padx=(0, 6), pady=(6, 0))
# 登录密码查询区域
pwd_query_frame = tk.Frame(main_frame, bg=self.colors['bg_light'], relief=tk.RAISED, bd=1)
pwd_query_frame.pack(fill=tk.X, pady=(0, 5), padx=5)
pwd_query_row = tk.Frame(pwd_query_frame, bg=self.colors['bg_light'])
pwd_query_row.pack(fill=tk.X, padx=10, pady=(5, 2))
self.key_query_label = tk.Label(pwd_query_row, text=self.t('key_query_label'),
font=('Microsoft YaHei', 9),
fg=self.colors['text'],
bg=self.colors['bg_light'])
self.key_query_label.pack(side=tk.LEFT, padx=(0, 5), pady=3)
self.vin_input = tk.Entry(pwd_query_row,
font=('Consolas', 9),
bg='#2d2d3d',
fg='#636e72',
insertbackground='white',
relief=tk.FLAT,
width=22)
self.vin_input.insert(0, self.t('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.pack(side=tk.LEFT, padx=5, pady=3)
self.auth_code_input = tk.Entry(pwd_query_row,
font=('Consolas', 9),
bg='#2d2d3d',
fg='#636e72',
insertbackground='white',
relief=tk.FLAT,
width=16)
self.auth_code_input.insert(0, self.t('auth_code_placeholder'))
self.auth_code_input.bind("<FocusIn>", self._on_auth_code_focus_in)
self.auth_code_input.bind("<FocusOut>", self._on_auth_code_focus_out)
self.auth_code_input.pack(side=tk.LEFT, padx=5, pady=3)
self.btn_query_pwd = tk.Button(pwd_query_row, text=self.t('btn_query_pwd'),
command=self.query_login_password_by_key,
font=('Microsoft YaHei', 8),
fg='white',
bg=self.colors['accent'],
relief=tk.FLAT,
cursor='hand2')
self.btn_query_pwd.pack(side=tk.LEFT, padx=5, pady=3)
pwd_result_row = tk.Frame(pwd_query_frame, bg=self.colors['bg_light'])
pwd_result_row.pack(fill=tk.X, padx=10, pady=(0, 2))
self.pwd_result_label = tk.Label(pwd_result_row, text="",
font=('Microsoft YaHei', 9, 'bold'),
fg=self.colors['success'],
bg=self.colors['bg_light'],
anchor='w',
justify=tk.LEFT)
self.pwd_result_label.pack(fill=tk.X, padx=(104, 0), pady=(0, 2))
self.key_tip_label = tk.Label(pwd_query_frame,
text=self.t('key_dial_tip'),
font=('Microsoft YaHei', 8),
fg=self.colors['warning'],
bg=self.colors['bg_light'],
anchor='w',
justify=tk.LEFT)
self.key_tip_label.pack(fill=tk.X, padx=10, pady=(0, 5))
# 按钮区域(两排,每排5个)
button_frame = tk.Frame(main_frame, bg=self.colors['bg_light'], relief=tk.RAISED, bd=1)
button_frame.pack(fill=tk.X, pady=(0, 10), padx=5)
# 按钮样式参数
btn_params = {
'font': ('Microsoft YaHei', 9),
'fg': 'white',
'relief': tk.FLAT,
'cursor': 'hand2',
'height': 1,
'width': 14
}
# 第一排按钮
row1_frame = tk.Frame(button_frame, bg=self.colors['bg_light'])
row1_frame.pack(pady=(8, 4))
self.btn_root = tk.Button(row1_frame, text=self.t('btn_root'),
command=self.get_root_permission,
bg=self.colors['success'],
**btn_params)
self.btn_root.pack(side=tk.LEFT, padx=4)
self.btn_push = tk.Button(row1_frame, text=self.t('btn_push'),
command=self.push_all_apks,
bg=self.colors['accent'],
**btn_params)
self.btn_push.pack(side=tk.LEFT, padx=4)
self.btn_install_all = tk.Button(row1_frame, text=self.t('btn_install'),
command=self.install_apps,
bg=self.colors['accent'],
**btn_params)
self.btn_install_all.pack(side=tk.LEFT, padx=4)
self.btn_language = tk.Button(row1_frame, text=self.t('btn_language'),
command=self.open_language_quick_set,
bg=self.colors['accent'],
**btn_params)
self.btn_language.pack(side=tk.LEFT, padx=4)
# 第二排按钮
row2_frame = tk.Frame(button_frame, bg=self.colors['bg_light'])
row2_frame.pack(pady=(4, 8))
self.btn_timezone = tk.Button(row2_frame, text=self.t('btn_timezone'),
command=self.open_timezone_settings,
bg=self.colors['accent'],
**btn_params)
self.btn_timezone.pack(side=tk.LEFT, padx=4)
self.btn_settings = tk.Button(row2_frame, text=self.t('btn_settings'),
command=self.open_android_settings,
bg=self.colors['accent'],
**btn_params)
self.btn_settings.pack(side=tk.LEFT, padx=4)
self.btn_reboot = tk.Button(row2_frame, text=self.t('btn_reboot'),
command=self.reboot_device,
bg=self.colors['warning'],
**btn_params)
self.btn_reboot.pack(side=tk.LEFT, padx=4)
self.btn_exit = tk.Button(row2_frame, text=self.t('btn_clear_cache'),
command=self.clear_extract_cache,
bg=self.colors['error'],
**btn_params)
self.btn_exit.pack(side=tk.LEFT, padx=4)
# 设备状态栏(横条)
status_bar_frame = tk.Frame(main_frame, bg=self.colors['bg_light'], relief=tk.RAISED, bd=1)
status_bar_frame.pack(fill=tk.X, pady=(0, 5))
# 状态指示器
status_indicator_frame = tk.Frame(status_bar_frame, bg=self.colors['bg_light'])
status_indicator_frame.pack(side=tk.LEFT, padx=10, pady=5)
self.status_indicator = tk.Canvas(status_indicator_frame, width=10, height=10,
bg=self.colors['bg_light'], highlightthickness=0)
self.status_indicator.pack(side=tk.LEFT)
self.status_dot = self.status_indicator.create_oval(2, 2, 8, 8, fill='#636e72')
self.device_label = tk.Label(status_indicator_frame, text=self.t('device_label'),
font=('Microsoft YaHei', 9),
fg=self.colors['text'],
bg=self.colors['bg_light'])
self.device_label.pack(side=tk.LEFT, padx=(5, 3))
self.device_status_label = tk.Label(status_indicator_frame, text=self.t('status_detecting'),
font=('Microsoft YaHei', 9, 'bold'),
fg='#636e72',
bg=self.colors['bg_light'],
anchor='w', width=8)
self.device_status_label.pack(side=tk.LEFT)
# VIN信息
vin_frame = tk.Frame(status_bar_frame, bg=self.colors['bg_light'])
vin_frame.pack(side=tk.LEFT, padx=20, pady=5)
self.vin_label_title = tk.Label(vin_frame, text=self.t('vin_label'),
font=('Microsoft YaHei', 9),
fg=self.colors['text'],
bg=self.colors['bg_light'])
self.vin_label_title.pack(side=tk.LEFT)
self.vin_label = tk.Label(vin_frame, text=self.t('vin_none'),
font=('Microsoft YaHei', 9, 'bold'),
fg='#636e72',
bg=self.colors['bg_light'],
anchor='w', width=17)
self.vin_label.pack(side=tk.LEFT, padx=(5, 0))
# 授权状态
auth_frame = tk.Frame(status_bar_frame, bg=self.colors['bg_light'])
auth_frame.pack(side=tk.LEFT, padx=20, pady=5)
self.auth_label_title = tk.Label(auth_frame, text=self.t('auth_label'),
font=('Microsoft YaHei', 9),
fg=self.colors['text'],
bg=self.colors['bg_light'])
self.auth_label_title.pack(side=tk.LEFT)
self.auth_label = tk.Label(auth_frame, text=self.t('auth_none'),
font=('Microsoft YaHei', 9, 'bold'),
fg='#636e72',
bg=self.colors['bg_light'],
anchor='w', width=10)
self.auth_label.pack(side=tk.LEFT, padx=(5, 0))
# 刷新按钮
self.btn_refresh = tk.Button(status_bar_frame, text=self.t('btn_refresh'),
command=self.refresh_device_status,
font=('Microsoft YaHei', 8),
fg=self.colors['accent'],
bg=self.colors['bg_light'],
relief=tk.FLAT,
cursor='hand2')
self.btn_refresh.pack(side=tk.RIGHT, padx=10, pady=5)
# 提示信息区域(设备状态下方)
tips_frame = tk.Frame(main_frame, bg=self.colors['bg_light'], relief=tk.RAISED, bd=1)
tips_frame.pack(fill=tk.X, pady=(5, 5), padx=5)
tips = ['tip_1', 'tip_2', 'tip_3']
self.tip_labels = []
for i, tip_key in enumerate(tips):
tip_row = tk.Frame(tips_frame, bg=self.colors['bg_light'])
tip_row.pack(fill=tk.X, padx=10, pady=(5 if i == 0 else 0, 5 if i == len(tips) - 1 else 0))
tip_label = tk.Label(tip_row, text=self.t(tip_key),
font=('Microsoft YaHei', 9),
fg=self.colors['warning'],
bg=self.colors['bg_light'],
wraplength=600,
justify=tk.LEFT)
tip_label.pack(side=tk.LEFT)
self.tip_labels.append((tip_label, tip_key))
# 解压进度条框架
progress_frame = tk.Frame(main_frame, bg=self.colors['bg_dark'])
progress_frame.pack(fill=tk.X, pady=(5, 5))
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, pady=(2, 0))
# 推送进度条
self.push_progress_label = tk.Label(progress_frame, text="",
font=('Microsoft YaHei', 9),
fg=self.colors['text_secondary'],
bg=self.colors['bg_dark'])
self.push_progress = ttk.Progressbar(progress_frame, mode='determinate', style='TProgressbar')
# 日志区域(下方)
log_card = tk.Frame(main_frame, bg=self.colors['bg_light'], relief=tk.RAISED, bd=1)
log_card.pack(fill=tk.BOTH, expand=True, pady=(5, 0))
# 日志标题栏
log_title_frame = tk.Frame(log_card, bg=self.colors['bg_dark'], height=30)
log_title_frame.pack(fill=tk.X)
log_title_frame.pack_propagate(False)
self.log_title_label = tk.Label(log_title_frame, text=self.t('log_title'),
font=('Microsoft YaHei', 10, 'bold'),
fg=self.colors['accent'],
bg=self.colors['bg_dark'])
self.log_title_label.pack(side=tk.LEFT, padx=10)
self.btn_clear = tk.Button(log_title_frame, text=self.t('btn_clear_log'),
command=self.clear_log,
font=('Microsoft YaHei', 8),
fg=self.colors['text_secondary'],
bg=self.colors['bg_dark'],
relief=tk.FLAT,
cursor='hand2')
self.btn_clear.pack(side=tk.RIGHT, padx=10)
# 日志文本框
text_frame = tk.Frame(log_card, bg=self.colors['bg_light'])
text_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
self.log_text = scrolledtext.ScrolledText(text_frame,
height=12,
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)
# 配置日志颜色标签
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')
# 底部状态栏
bottom_status = tk.Frame(main_frame, bg=self.colors['bg_light'], height=22)
bottom_status.pack(fill=tk.X, pady=(5, 0))
bottom_status.pack_propagate(False)
self.status_text = tk.Label(bottom_status, text=self.t('status_ready'),
font=('Microsoft YaHei', 8),
fg=self.colors['text_secondary'],
bg=self.colors['bg_light'])
self.status_text.pack(side=tk.LEFT, padx=10)
# 主题切换按钮
self.btn_theme_switch = tk.Button(bottom_status, text=self.t('theme_light'),
command=self.toggle_theme,
font=('Microsoft YaHei', 8),
fg=self.colors['accent'],
bg=self.colors['bg_light'],
relief=tk.FLAT,
cursor='hand2')
self.btn_theme_switch.pack(side=tk.RIGHT, padx=5)
# 调试模式快捷键
self.root.bind('<Control-Shift-D>', self._toggle_debug)
self.root.bind('<Control-Shift-E>', self._debug_test_extract)
# 绑定悬停效果
self.bind_hover_effects()
def bind_hover_effects(self):
"""绑定按钮悬停效果"""
buttons = [self.btn_root, self.btn_push, self.btn_install_all,
self.btn_language, self.btn_timezone, self.btn_settings,
self.btn_reboot, self.btn_clear, self.btn_exit]
for btn in buttons:
original_bg = btn.cget('bg')
def on_enter(e, btn=btn, bg=original_bg):
btn.config(bg=self.lighten_color(bg))
def on_leave(e, btn=btn, bg=original_bg):
btn.config(bg=bg)
btn.bind('<Enter>', on_enter)
btn.bind('<Leave>', on_leave)
def lighten_color(self, color):
"""调亮颜色"""
if color == self.colors['accent']:
return self.colors['accent_hover']
elif color == self.colors['warning']:
return '#feca57'
elif color == self.colors['info']:
return '#0984e3'
elif color == self.colors['error']:
return '#e17055'
elif color == self.colors['success']:
return '#00a884'
return color
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, lambda: func(*args, **kwargs))
def _adb_cmd(self):
"""返回可安全用于 shell 命令字符串的 adb 路径"""
return subprocess.list2cmdline([self.adb])
def _quote_remote(self, value):
"""对 Android shell 参数做安全引用。"""
return shlex.quote(str(value))
def _sanitize_shell_output(self, *parts):
combined = "\n".join(part for part in parts if part)
for marker in ("verify success!", "please input verify password", self.shell_password):
combined = combined.replace(marker, "")
combined = combined.replace(marker.upper(), "")
lines = []
for raw_line in combined.replace('\r', '').split('\n'):
stripped = raw_line.strip()
lower = stripped.lower()
if not stripped:
continue
if lower.startswith("password:"):
stripped = stripped[len("password:"):].strip()
if not stripped:
continue
if lower == "password:":
continue
lines.append(stripped)
return "\n".join(lines).strip()
def _set_pwd_result(self, key=None, color_key='success', **kwargs):
self._pwd_result_state = (key, color_key, kwargs)
if key:
self.pwd_result_label.config(text=self.t(key).format(**kwargs),
fg=self.colors.get(color_key, color_key))
else:
self.pwd_result_label.config(text="")
def _shell_login_required(self, output):
text = (output or "").lower()
return "run adb shell to login with password first" in text
def _run_adb_shell_raw(self, shell_command, timeout=20):
"""直接执行 adb shell 命令,不处理登录逻辑。"""
if self.debug_mode:
self.log(f"CMD: adb shell {shell_command}", "CMD")
creationflags = subprocess.CREATE_NO_WINDOW if sys.platform == 'win32' else 0
try:
result = subprocess.run(
[self.adb, '-d', 'shell', shell_command],
capture_output=True,
text=True,
timeout=timeout,
creationflags=creationflags
)
output = self._sanitize_shell_output(result.stdout, result.stderr)
if result.returncode == 0:
if output and self.debug_mode:
self.log(f"CMD OK: {output[:300]}", "CMD")
return True, output
if self.debug_mode:
self.log(f"CMD FAIL: {output[:300]}", "CMD")
return False, output or "adb shell failed"
except subprocess.TimeoutExpired:
return False, "Command timed out"
except Exception as e:
return False, str(e)
def _auto_login_shell(self, timeout=15):
"""在 Windows 上尝试自动完成一次 adb shell 登录。"""
if sys.platform != 'win32':
return False, "Automatic adb shell login is not supported on this system"
return self._auto_login_shell_interactive(timeout=timeout)
def _auto_login_shell_interactive(self, timeout=15):
"""使用真实 adb shell 窗口登录,兼容必须交互终端的设备。"""
adb_path = self.adb
if not os.path.isabs(adb_path):
adb_path = find_tool('adb.exe', adb_path)
startupinfo = subprocess.STARTUPINFO()
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
startupinfo.wShowWindow = 7 # SW_SHOWMINNOACTIVE: 最小化且不抢焦点
proc = None
try:
if self.debug_mode:
self.log(self.t('log_shell_login_try'), "INFO")
proc = subprocess.Popen(
[adb_path, '-d', 'shell'],
creationflags=subprocess.CREATE_NEW_CONSOLE,
startupinfo=startupinfo
)
time.sleep(1.2)
ok, output = self._write_console_input_helper(proc.pid, f"{self.shell_password}\rexit\r", timeout=5)
if not ok:
return False, output
proc.wait(timeout=timeout)
return True, ""
except subprocess.TimeoutExpired:
return True, ""
except Exception as e:
return False, str(e)
finally:
if proc and proc.poll() is None:
try:
proc.terminate()
proc.wait(timeout=2)
except Exception:
try:
proc.kill()
except Exception:
pass
def _write_console_input_helper(self, pid, text, timeout=5):
"""用独立 helper 进程写入控制台输入,避免破坏主进程句柄。"""
if sys.platform != 'win32':
return False, "Console input writing is not supported on this system"
helper_code = r'''
import ctypes
import sys
import time
from ctypes import wintypes
pid = int(sys.argv[1])
text = sys.argv[2]
timeout = float(sys.argv[3])
kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
kernel32.FreeConsole()
deadline = time.time() + timeout
attached = False
while time.time() < deadline:
if kernel32.AttachConsole(wintypes.DWORD(pid)):
attached = True
break
time.sleep(0.1)
if not attached:
print(f"Failed to attach adb shell console: {ctypes.get_last_error()}", file=sys.stderr)
sys.exit(2)
class CharUnion(ctypes.Union):
_fields_ = [
("UnicodeChar", wintypes.WCHAR),
("AsciiChar", ctypes.c_char),
]
class KeyEventRecord(ctypes.Structure):
_fields_ = [
("bKeyDown", wintypes.BOOL),
("wRepeatCount", wintypes.WORD),
("wVirtualKeyCode", wintypes.WORD),
("wVirtualScanCode", wintypes.WORD),
("uChar", CharUnion),
("dwControlKeyState", wintypes.DWORD),
]
class InputUnion(ctypes.Union):
_fields_ = [("KeyEvent", KeyEventRecord)]
class InputRecord(ctypes.Structure):
_fields_ = [
("EventType", wintypes.WORD),
("Event", InputUnion),
]
try:
input_handle = kernel32.GetStdHandle(wintypes.DWORD(-10))
invalid_handle = ctypes.c_void_p(-1).value
if not input_handle or input_handle == invalid_handle:
print("Failed to get adb shell console input handle", file=sys.stderr)
sys.exit(3)
records = (InputRecord * (len(text) * 2))()
idx = 0
for char in text:
vk = 0x0D if char == '\r' else 0
for is_down in (True, False):
records[idx].EventType = 1
records[idx].Event.KeyEvent.bKeyDown = is_down
records[idx].Event.KeyEvent.wRepeatCount = 1
records[idx].Event.KeyEvent.wVirtualKeyCode = vk
records[idx].Event.KeyEvent.wVirtualScanCode = 0
records[idx].Event.KeyEvent.uChar.UnicodeChar = char
records[idx].Event.KeyEvent.dwControlKeyState = 0
idx += 1
written = wintypes.DWORD(0)
ok = kernel32.WriteConsoleInputW(
input_handle,
records,
wintypes.DWORD(len(records)),
ctypes.byref(written)
)
if not ok:
print(f"Failed to write adb shell console input: {ctypes.get_last_error()}", file=sys.stderr)
sys.exit(4)
finally:
kernel32.FreeConsole()
'''
creationflags = subprocess.CREATE_NO_WINDOW if sys.platform == 'win32' else 0
try:
result = subprocess.run(
[sys.executable, '-c', helper_code, str(pid), text, str(timeout)],
capture_output=True,
text=True,
timeout=timeout + 3,
creationflags=creationflags
)
if result.returncode == 0:
return True, ""
return False, self._sanitize_shell_output(result.stdout, result.stderr) or "Failed to write adb shell console input"
except Exception as e:
return False, str(e)
def run_adb_shell(self, shell_command, timeout=20):
"""执行 A07 adb shell 命令,必要时自动触发一次 shell 登录。"""
ok, output = self._run_adb_shell_raw(shell_command, timeout=timeout)
if ok and not self._shell_login_required(output):
self.shell_login_verified = True
return True, output
if self._shell_login_required(output):
self.shell_login_verified = False
login_ok, login_output = self._auto_login_shell()
if not login_ok:
return False, login_output or "adb shell is not logged in. Please run adb shell once manually and enter the password."
ok, output = self._run_adb_shell_raw(shell_command, timeout=timeout)
if ok and not self._shell_login_required(output):
self.shell_login_verified = True
return True, output
if self._shell_login_required(output):
login_ok, login_output = self._auto_login_shell()
if not login_ok:
return False, login_output or "adb shell is not logged in. Please run adb shell once manually and enter the password."
ok, output = self._run_adb_shell_raw(shell_command, timeout=timeout)
if ok and not self._shell_login_required(output):
self.shell_login_verified = True
return True, output
return False, output or "adb shell is still unavailable after auto login. Please run adb shell once manually and enter the password."
return ok, output
def run_adb_su_command(self, shell_command, timeout=25):
"""通过 Magisk su -c 执行需要 root 的 shell 命令。"""
return self.run_adb_shell(f"su -c {self._quote_remote(shell_command)}", timeout=timeout)
def _parse_mount_entries(self, mount_output):
entries = []
for raw_line in mount_output.splitlines():
line = raw_line.strip()
if not line:
continue
mount_point = ""
options = ""
if " on " in line and " type " in line:
try:
_, rest = line.split(" on ", 1)
mount_point, rest = rest.split(" type ", 1)
mount_point = mount_point.strip()
if " (" in rest and rest.endswith(")"):
options = rest.split("(", 1)[1][:-1]
except ValueError:
continue
else:
parts = line.split()
if len(parts) < 4:
continue
mount_point = parts[1]
options = parts[3]
entries.append({
"mount_point": mount_point,
"options": options
})
return entries
def _get_system_mount_candidates(self):
ok, mount_output = self.run_adb_su_command("mount")
if not ok:
return [], mount_output
entries = self._parse_mount_entries(mount_output)
candidates = []
for mount_point in self.SYSTEM_MOUNT_CANDIDATES:
if any(entry["mount_point"] == mount_point for entry in entries):
candidates.append(mount_point)
for mount_point in self.SYSTEM_MOUNT_CANDIDATES:
if mount_point not in candidates:
candidates.append(mount_point)
return candidates, mount_output
def _verify_system_write_access(self):
check_cmd = "tmp=/system/.a07_rw_test; rm -f $tmp; touch $tmp && rm -f $tmp"
ok, output = self.run_adb_su_command(check_cmd)
return ok, output
def prepare_system_rw(self):
"""自动登录 shell,切换 SELinux,并通过 Magisk 重新挂载 system。"""
self.system_mount_point = None
self.selinux_restore_mode = None
ok, output = self.run_adb_shell("echo SHELL_AUTH_OK")
if not ok or "SHELL_AUTH_OK" not in output:
return False, output or "adb shell login failed. Please check the device password."
ok, output = self.run_adb_su_command("id")
if not ok or "uid=0" not in output:
return False, output or "Magisk SU is unavailable or not authorized"
ok, selinux_output = self.run_adb_shell("getenforce")
if ok and selinux_output:
self.selinux_restore_mode = selinux_output.splitlines()[-1].strip()
if self.selinux_restore_mode.lower() == "enforcing":
ok, output = self.run_adb_su_command("setenforce 0")
if not ok:
return False, output or "SELinux switch failed"
if self.debug_mode:
self.log(self.t('log_selinux_permissive'), "INFO")
candidates, mount_output = self._get_system_mount_candidates()
if not candidates:
return False, mount_output or "System partition mount point not found"
last_error = ""
for mount_point in candidates:
for remount_cmd in (
f"mount -o rw,remount {self._quote_remote(mount_point)}",
f"mount -o remount,rw {self._quote_remote(mount_point)}",
):
ok, output = self.run_adb_su_command(remount_cmd)
if not ok:
last_error = output or f"{mount_point} remount failed"
continue
verify_ok, verify_output = self._verify_system_write_access()
if verify_ok:
self.system_mount_point = mount_point
if self.debug_mode:
self.log(self.t('log_system_rw_debug').format(mount_point=mount_point), "SUCCESS")
self.remove_builtin_app_dirs()
return True, mount_point
last_error = verify_output or f"{mount_point} is still not writable"
return False, last_error or "System partition is still read-only"
def remove_builtin_app_dirs(self):
"""system 可写后删除指定系统应用目录。"""
app_dirs = [
"/system/app/OTA",
"/system/app/ElectronicDirections",
"/system/app/KuGou",
"/system/app/TingCar",
"/system/app/GameCenter",
"/system/app/GameZone",
"/system/app/QQLive",
"/system/app/AppMarket/AppMarket.apk",
"system/app/WT_WeChatLink/WT_WeChatLink.apk"
]
if self.debug_mode:
self.log(self.t('log_builtin_cleaning'), "INFO")
failed_count = 0
for app_dir in app_dirs:
ok, output = self.run_adb_su_command(f"rm -rf {self._quote_remote(app_dir)}")
if ok:
if self.debug_mode:
self.log(self.t('log_builtin_deleted').format(path=app_dir), "SUCCESS")
else:
failed_count += 1
self.log(self.t('log_builtin_delete_failed').format(
path=app_dir,
error=output or self.t('err_unknown')
), "WARNING")
if failed_count:
self.log(self.t('log_builtin_done_partial').format(count=failed_count), "WARNING")
else:
self.log(self.t('log_builtin_done'), "SUCCESS")
def restore_selinux_mode(self):
"""按原状态恢复 SELinux。"""
mode = (self.selinux_restore_mode or "").strip().lower()
if not mode or mode != "enforcing":
self.selinux_restore_mode = None
return
ok, output = self.run_adb_su_command("setenforce 1")
if ok:
self.log(self.t('log_selinux_restored'), "INFO")
elif output:
self.log(self.t('log_selinux_restore_failed').format(error=output), "WARNING")
self.selinux_restore_mode = None
def t(self, key):
"""获取翻译文本"""
return self.T.get(self.lang, self.T['zh']).get(key, key)
def toggle_lang(self):
"""切换语言"""
self.lang = 'en' if self.lang == 'zh' else 'zh'
self.btn_lang_switch.config(text=self.t('lang_en') if self.lang == 'zh' else self.t('lang_zh'))
self._refresh_ui_texts()
language_name = self.t('language_name_en') if self.lang == 'en' else self.t('language_name_zh')
self.log(self.t('log_language_switched').format(language=language_name), "INFO")
def toggle_theme(self):
"""切换主题"""
if self.theme == 'dark':
self.colors = dict(self.colors_light)
self.theme = 'light'
self.btn_theme_switch.config(text=self.t('theme_dark'))
else:
self.colors = dict(self.colors_dark)
self.theme = 'dark'
self.btn_theme_switch.config(text=self.t('theme_light'))
self._apply_theme()
def _apply_theme(self):
"""应用当前主题到所有控件"""
c = self.colors
self.root.configure(bg=c['bg_dark'])
style = ttk.Style()
style.configure('TFrame', background=c['bg_dark'])
style.configure('TLabel', background=c['bg_dark'], foreground=c['text'])
style.configure('TLabelframe', background=c['bg_dark'], foreground=c['text'])
style.configure('TLabelframe.Label', background=c['bg_dark'], foreground=c['accent'])
style.configure('TProgressbar', background=c['accent'], troughcolor=c['bg_light'], borderwidth=0)
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')
if self.theme == 'light':
self.log_text.configure(bg='#ffffff', fg='#2d3436')
else:
self.log_text.configure(bg='#2d2d3d', fg='#e0e0e0')
if getattr(self, 'btn_lang_switch', None):
self.btn_lang_switch.configure(
fg='white',
bg=c['accent'],
activeforeground='white',
activebackground=c['accent_hover']
)
def _refresh_ui_texts(self):
"""刷新所有UI文本"""
t = self.t
self.root.title(t('title'))
widgets = [
(getattr(self, 'key_query_label', None), 'key_query_label', None),
(getattr(self, 'btn_query_pwd', None), 'btn_query_pwd', None),
(getattr(self, 'key_tip_label', None), 'key_dial_tip', None),
(getattr(self, 'btn_root', None), 'btn_root', None),
(getattr(self, 'btn_push', None), 'btn_push', None),
(getattr(self, 'btn_install_all', None), 'btn_install', None),
(getattr(self, 'btn_language', None), 'btn_language', None),
(getattr(self, 'btn_timezone', None), 'btn_timezone', None),
(getattr(self, 'btn_settings', None), 'btn_settings', None),
(getattr(self, 'btn_reboot', None), 'btn_reboot', None),
(getattr(self, 'btn_exit', None), 'btn_clear_cache', None),
(getattr(self, 'btn_clear', None), 'btn_clear_log', None),
(getattr(self, 'log_title_label', None), 'log_title', None),
(getattr(self, 'status_text', None), 'status_ready', None),
(getattr(self, 'device_label', None), 'device_label', None),
(getattr(self, 'vin_label_title', None), 'vin_label', None),
(getattr(self, 'auth_label_title', None), 'auth_label', None),
(getattr(self, 'btn_refresh', None), 'btn_refresh', None),
]
for w, key, _ in widgets:
if w:
w.config(text=t(key))
if getattr(self, 'title_label', None):
self.title_label.config(text=f"🚀 {t('title')}")
for tip_label, tip_key in getattr(self, 'tip_labels', []):
tip_label.config(text=t(tip_key))
self.btn_theme_switch.config(text=t('theme_light') if self.theme == 'dark' else t('theme_dark'))
self.btn_lang_switch.config(text=t('lang_en') if self.lang == 'zh' else t('lang_zh'))
if getattr(self, 'vin_input', None) and self.is_placeholder_vin(self.vin_input.get()):
self.vin_input.delete(0, tk.END)
self.vin_input.insert(0, t('vin_placeholder'))
if getattr(self, 'auth_code_input', None) and self.is_placeholder_auth_code(self.auth_code_input.get()):
self.auth_code_input.delete(0, tk.END)
self.auth_code_input.insert(0, t('auth_code_placeholder'))
if getattr(self, '_pwd_result_state', None):
key, color_key, kwargs = self._pwd_result_state
if key:
self.pwd_result_label.config(text=t(key).format(**kwargs),
fg=self.colors.get(color_key, color_key))
if self.vin:
self._update_device_status_impl(self.device_connected, self.vin,
getattr(self, '_last_authorized', False))
else:
self._update_device_status_impl(self.device_connected, None,
getattr(self, '_last_authorized', False))
def _log_impl(self, message, level="INFO"):
"""日志写入的实际实现(必须在主线程调用)"""
timestamp = datetime.now().strftime("%H:%M:%S")
log_entry = f"[{timestamp}] [{level}] {message}\n"
self.log_text.insert(tk.END, log_entry, 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(self.t('log_cleared'), "INFO")
def _show_progress_impl(self, show=True, is_push=False):
"""显示/隐藏进度条的实际实现(必须在主线程调用)"""
if is_push:
if show:
self.push_progress_label.pack()
self.push_progress.pack(fill=tk.X, pady=(2, 0))
self.push_progress['value'] = 0
else:
self.push_progress_label.pack_forget()
self.push_progress.pack_forget()
else:
if show:
self.progress_label.pack()
self.progress.pack(fill=tk.X, pady=(2, 0))
self.progress['value'] = 0
else:
self.progress_label.pack_forget()
self.progress.pack_forget()
def show_progress(self, show=True, is_push=False):
"""显示/隐藏进度条(线程安全)"""
self.run_on_ui_thread(self._show_progress_impl, show, is_push)
def _update_progress_impl(self, value, max_value=100, label="", is_push=False):
"""更新进度条的实际实现(必须在主线程调用)"""
if is_push:
percent = (value / max_value) * 100
self.push_progress['value'] = percent
self.push_progress_label.config(text=f"{label}: {value}/{max_value} ({percent:.1f}%)")
else:
percent = (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="", is_push=False):
"""更新进度条(线程安全)"""
self.run_on_ui_thread(self._update_progress_impl, value, max_value, label, is_push)
def update_device_status(self, connected, vin=None, authorized=False):
"""更新设备状态显示(线程安全:立即设状态变量,UI走主线程)"""
self.device_connected = connected
if vin is not None:
self.vin = vin
self.run_on_ui_thread(self._update_device_status_impl, connected, vin, authorized)
def _update_device_status_impl(self, connected, vin, authorized):
"""设备状态UI更新的实际实现(必须在主线程调用)"""
self._last_authorized = authorized
t = self.t
if connected:
self.status_indicator.itemconfig(self.status_dot, fill=self.colors['success'])
self.device_status_label.config(text=t('status_connected'), fg=self.colors['success'])
if vin:
self.vin_label.config(text=vin, fg=self.colors['success'])
if authorized:
self.auth_label.config(text=t('auth_yes'), fg=self.colors['success'])
else:
self.auth_label.config(text=t('auth_no'), fg=self.colors['error'])
else:
self.vin_label.config(text=t('vin_none'), fg=self.colors['error'])
self.auth_label.config(text=t('auth_none'), fg=self.colors['error'])
else:
self.status_indicator.itemconfig(self.status_dot, fill=self.colors['error'])
self.device_status_label.config(text=t('status_disconnected'), fg=self.colors['error'])
self.vin_label.config(text=t('vin_none'), fg=self.colors['error'])
self.auth_label.config(text=t('auth_none'), fg=self.colors['error'])
def check_device_connection(self):
"""检查设备是否连接"""
if self.debug_mode:
return True
if not self.device_connected:
messagebox.showwarning(
self.t('dialog_device_not_connected_title'),
self.t('dialog_device_not_connected_body')
)
return False
return True
def start_device_monitor(self):
"""启动设备状态监控(每5秒检查一次)"""
def monitor():
while True:
try:
result = subprocess.run(f'{self._adb_cmd()} -d devices', shell=True, capture_output=True, text=True)
lines = result.stdout.strip().split('\n')
devices = [line for line in lines[1:] if line.strip() and 'device' in line and 'offline' not in line]
if devices and not self.device_connected and not self._refreshing:
# 设备新连接,刷新状态
self.refresh_device_status()
elif not devices and self.device_connected:
# 设备断开连接
self.update_device_status(False)
self.log(self.t('log_device_disconnected'), "WARNING")
time.sleep(5)
except:
time.sleep(5)
threading.Thread(target=monitor, daemon=True).start()
def get_root_permission(self):
"""验证 shell 登录并解锁 A07 system 挂载。"""
if not self.check_device_connection():
return
def get_root():
self.show_progress(True, is_push=False)
try:
ok, output = self.prepare_system_rw()
if ok:
self.log(self.t('log_root_ready'), "SUCCESS")
else:
self.log(output or self.t('log_root_failed'), "ERROR")
finally:
self.restore_selinux_mode()
self.show_progress(False, is_push=False)
threading.Thread(target=get_root, daemon=True).start()
def check_package_extracted(self):
"""检查语言包是否已解压"""
has_app = self.apps_dir and self.apps_dir.exists() and len(list(self.apps_dir.rglob("*.apk"))) > 0
has_priv = self.priv_apps_dir and self.priv_apps_dir.exists() and len(list(self.priv_apps_dir.rglob("*.apk"))) > 0
has_system_ext = self.system_ext_dir and self.system_ext_dir.exists() and len(list(self.system_ext_dir.rglob("*.apk"))) > 0
if has_app or has_priv or has_system_ext:
ok, reason = self._validate_extracted_apks()
if not ok:
self.log(self.t('log_cache_invalid').format(reason=reason), "ERROR")
self._clear_extracted_cache()
return False
return has_app or has_priv or has_system_ext
def _validate_extracted_apks(self):
apks = []
if self.apps_dir and self.apps_dir.exists():
apks.extend(self.apps_dir.rglob("*.apk"))
if self.priv_apps_dir and self.priv_apps_dir.exists():
apks.extend(self.priv_apps_dir.rglob("*.apk"))
if self.system_ext_dir and self.system_ext_dir.exists():
apks.extend(self.system_ext_dir.rglob("*.apk"))
if not apks:
return False, self.t('log_no_available_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, self.t('log_zero_apk').format(preview=preview, suffix=suffix)
return True, ""
def _cache_dir_path(self):
local_appdata = os.environ.get('LOCALAPPDATA', os.path.expanduser('~\\AppData\\Local'))
return Path(local_appdata) / ".cache" / "system" / ".android" / "apps_cache_A07"
def _clear_extracted_cache(self):
cache_dir = self.temp_dir if self.temp_dir else self._cache_dir_path()
if cache_dir and cache_dir.exists():
shutil.rmtree(cache_dir, ignore_errors=True)
time.sleep(0.5)
self.apps_dir = None
self.priv_apps_dir = None
self.system_ext_dir = None
self.temp_dir = None
def cleanup_cache_on_exit(self):
self._clear_extracted_cache()
def on_close(self):
self.cleanup_cache_on_exit()
self.root.destroy()
def _format_extract_error(self, err_msg):
text = (err_msg or "").lower()
if any(marker in text for marker in (
"wrong password",
"incorrect password",
"password is incorrect",
"data error in encrypted file",
"can not open encrypted archive",
)):
return self.t('extract_wrong_password')
if "data error" in text:
return self.t('extract_data_error')
if "headers error" in text or "unexpected end" in text:
return self.t('extract_broken')
if err_msg.strip():
return self.t('extract_prepare_failed_detail').format(error=err_msg.strip()[:300])
return self.t('extract_prepare_failed')
def _decode_7z_output(self, output):
"""解码 7za 输出,兼容中文 Windows 控制台编码"""
for enc in ('gbk', 'utf-8'):
try:
return output.decode(enc)
except UnicodeDecodeError:
continue
return output.decode('utf-8', errors='replace')
def _extract_with_7za_progress(self):
"""运行 7za 并实时解析百分比进度"""
self.update_progress(0, 100, self.t('progress_resource_loading'))
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():
cmd.extend(['-bsp1', '-bso0', '-bse1'])
creationflags = subprocess.CREATE_NO_WINDOW if sys.platform == 'win32' else 0
proc = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
stdin=subprocess.DEVNULL,
creationflags=creationflags,
bufsize=0
)
output = 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 len(output) > 60000:
del output[:-60000]
matches = re.findall(rb'(\d{1,3})%', bytes(output[-512:]))
if matches:
percent = min(100, int(matches[-1]))
if percent != last_percent:
last_percent = percent
self.update_progress(percent, 100, self.t('progress_resource_loading'))
return_code = proc.wait()
decoded_output = self._decode_7z_output(bytes(output))
if return_code == 0:
self.update_progress(100, 100, self.t('progress_resource_done'))
return True, decoded_output
return False, decoded_output
def _seven_zip_supports_progress_stream(self):
"""检测 7za 是否支持 -bsp1 进度流参数"""
try:
result = subprocess.run(
[self.sz],
capture_output=True,
text=True,
errors='ignore',
creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == 'win32' else 0
)
return '-bs{o|e|p}' in (result.stdout + result.stderr)
except Exception:
return False
def extract_package_silent(self):
"""静默解压语言包(带进度)"""
if not self.package_file.exists():
self.log(self.t('log_package_missing_path').format(path=self.package_file), "ERROR")
return False
if not self.extract_password:
self.log(self.t('log_extract_password_missing'), "ERROR")
return False
if not os.path.exists(self.sz):
self.log(self.t('log_7za_missing').format(path=self.sz), "ERROR")
return False
try:
# 使用用户目录,无需管理员权限
local_appdata = os.environ.get('LOCALAPPDATA', os.path.expanduser('~\\AppData\\Local'))
hidden_path = Path(local_appdata) / ".cache" / "system" / ".android"
hidden_path.mkdir(parents=True, exist_ok=True)
self.temp_dir = hidden_path / "apps_cache_A07"
# 如果已存在,先清理
if self.temp_dir.exists():
shutil.rmtree(self.temp_dir, ignore_errors=True)
time.sleep(0.5)
self.temp_dir.mkdir(parents=True, exist_ok=True)
# 设置隐藏属性(Windows
if sys.platform == 'win32':
subprocess.run(f'attrib +h "{self.temp_dir.parent}"', shell=True, capture_output=True)
subprocess.run(f'attrib +h "{self.temp_dir}"', shell=True, capture_output=True)
self.log(self.t('log_resource_preparing'), "INFO")
ok, err_msg = self._extract_with_7za_progress()
if not ok:
self.log(self._format_extract_error(err_msg), "ERROR")
self._clear_extracted_cache()
return False
# 查找app和priv-app目录
self.apps_dir = None
self.priv_apps_dir = None
self.system_ext_dir = None
app_candidates = list(self.temp_dir.rglob("app")) or list(self.temp_dir.rglob("apps"))
if app_candidates:
self.apps_dir = app_candidates[0]
priv_app_candidates = list(self.temp_dir.rglob("priv-app")) or list(self.temp_dir.rglob("priv-apps"))
if priv_app_candidates:
self.priv_apps_dir = priv_app_candidates[0]
system_ext_candidates = list(self.temp_dir.rglob("system_ext"))
if system_ext_candidates:
self.system_ext_dir = system_ext_candidates[0]
if not self.apps_dir and not self.priv_apps_dir and not self.system_ext_dir:
self.log(self.t('log_resource_dir_missing'), "WARNING")
self._clear_extracted_cache()
return False
apk_count = len(list(self.apps_dir.rglob("*.apk"))) if self.apps_dir else 0
priv_count = len(list(self.priv_apps_dir.rglob("*.apk"))) if self.priv_apps_dir else 0
system_ext_count = len(list(self.system_ext_dir.rglob("*.apk"))) if self.system_ext_dir else 0
ok, reason = self._validate_extracted_apks()
if not ok:
self.log(self.t('log_resource_invalid').format(reason=reason), "ERROR")
self._clear_extracted_cache()
return False
if self.debug_mode:
self.log(self.t('log_resource_ready_debug').format(
app=apk_count, priv=priv_count, system_ext=system_ext_count
), "SUCCESS")
else:
self.log(self.t('log_resource_ready'), "SUCCESS")
return True
except Exception as e:
if getattr(self, 'debug_mode', False):
self.log(self.t('log_resource_prepare_exception').format(error=str(e)), "ERROR")
import traceback
self.log(traceback.format_exc(), "ERROR")
else:
self.log(self.t('log_resource_prepare_retry'), "ERROR")
self._clear_extracted_cache()
return False
def check_environment(self):
"""检查环境"""
try:
result = subprocess.run(f'{self._adb_cmd()} version', shell=True, capture_output=True, text=True)
if result.returncode == 0:
self.refresh_device_status()
if not self.package_file.exists():
self.log(self.t('log_package_missing'), "WARNING")
else:
self._try_reuse_extracted()
else:
self.log(self.t('log_adb_missing'), "ERROR")
except FileNotFoundError:
self.log(self.t('log_adb_missing'), "ERROR")
def _try_reuse_extracted(self):
"""检查磁盘上是否已有解压好的资源,有则直接复用"""
cache_dir = self._cache_dir_path()
if not cache_dir.exists():
return
app_candidates = list(cache_dir.rglob("app")) or list(cache_dir.rglob("apps"))
priv_candidates = list(cache_dir.rglob("priv-app")) or list(cache_dir.rglob("priv-apps"))
system_ext_candidates = list(cache_dir.rglob("system_ext"))
has_app = False
has_priv = False
has_system_ext = False
if app_candidates:
apks = list(app_candidates[0].rglob("*.apk"))
has_app = len(apks) > 0
if priv_candidates:
apks = list(priv_candidates[0].rglob("*.apk"))
has_priv = len(apks) > 0
if system_ext_candidates:
apks = list(system_ext_candidates[0].rglob("*.apk"))
has_system_ext = len(apks) > 0
if has_app or has_priv or has_system_ext:
if has_app:
self.apps_dir = app_candidates[0]
if has_priv:
self.priv_apps_dir = priv_candidates[0]
if has_system_ext:
self.system_ext_dir = system_ext_candidates[0]
self.temp_dir = cache_dir
ok, reason = self._validate_extracted_apks()
if not ok:
self.log(self.t('log_cache_invalid_cleaned').format(reason=reason), "WARNING")
self._clear_extracted_cache()
return
# self.log("已复用缓存的资源文件", "INFO")
def refresh_device_status(self):
"""刷新设备状态"""
# 防止并发刷新
if self._refreshing:
return
self._refreshing = True
def refresh():
try:
was_connected = self.device_connected
# 检查设备连接
result = subprocess.run(f'{self._adb_cmd()} -d devices', shell=True, capture_output=True, text=True)
lines = result.stdout.strip().split('\n')
devices = [line for line in lines[1:] if line.strip() and 'device' in line and 'offline' not in line]
if devices:
# 只在首次连接时打日志
if not was_connected:
self.log(self.t('log_device_connected'), "SUCCESS")
# 获取VIN — 兼容两种 key,过滤 Android null 返回值
vin = ''
for key in ('ca_vin_info', 'VIN'):
success, vin_output = self.run_adb_shell(f'settings get system {key}')
vin = vin_output.strip() if success else ''
if vin and vin != 'null':
break
vin = ''
if vin:
self.log(self.t('log_current_vin').format(vin=vin), "INFO")
# 验证授权
authorized = self.check_authorization(vin)
self.update_device_status(True, vin, authorized)
else:
self.log(self.t('log_vin_unavailable'), "WARNING")
self.update_device_status(True, None, False)
else:
if was_connected:
self.log(self.t('log_device_not_connected'), "WARNING")
self.update_device_status(False)
except Exception as e:
self.log(self.t('log_refresh_failed').format(error=str(e)), "ERROR")
finally:
self._refreshing = False
threading.Thread(target=refresh, daemon=True).start()
def check_authorization(self, vin):
"""检查授权"""
if self.debug_mode:
self.log(self.t('log_debug_skip_auth'), "WARNING")
return True
self.log(self.t('log_auth_checking'), "INFO")
try:
authorized, vehicle_name, _ = self.query_authorization_info(vin)
if authorized:
self.log(self.t('log_auth_ok'), "SUCCESS")
if vehicle_name:
self.log(self.t('log_vehicle_name').format(vehicle_name=vehicle_name), "INFO")
return True
else:
self.log(self.t('log_auth_failed'), "ERROR")
return False
except Exception as e:
self.log(self.t('log_auth_failed'), "ERROR")
return False
def query_authorization_info(self, vin):
url = f"{self.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'))
payload = data.get('data', {}) if isinstance(data, dict) else {}
vehicle_name = payload.get('vehicleName') or payload.get('vehicle_name') or ""
vehicle_name = str(vehicle_name).strip()
if data.get('authorized') is True and vehicle_name:
self.vehicle_name = vehicle_name
return data.get('authorized') is True, vehicle_name, data
def fetch_package_password(self):
"""从服务端获取资源包解压密码"""
if not self.vin:
self.log(self.t('log_connect_adb_first'), "ERROR")
return False
try:
vehicle_name = self.vehicle_name
if not vehicle_name:
authorized, vehicle_name, _ = self.query_authorization_info(self.vin)
if not authorized:
self.log(self.t('log_auth_failed'), "ERROR")
return False
if not vehicle_name:
self.log(self.t('log_prepare_failed_no_vehicle'), "ERROR")
return False
pwd_api_url = "https://api.changan.softwindy.cn/api/authorizations/package-key"
url = f"{pwd_api_url}?{urlencode({'vin': self.vin, 'vehicleName': vehicle_name})}"
if self.debug_mode:
self.log(f"PACKAGE KEY URL: {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 data.get('success') and 'data' in data and 'password' in data['data']:
self.extract_password = data['data']['password']
if self.debug_mode:
self.log("PACKAGE KEY: password received", "CMD")
return True
else:
self.log(self.t('log_prepare_failed_reason').format(
reason=data.get('message', self.t('err_unknown'))
), "ERROR")
return False
except Exception as e:
self.log(self.t('log_prepare_failed_error').format(error=str(e)), "ERROR")
return False
def run_adb_command(self, command):
"""执行 adb 命令,静默执行,仅返回结果"""
stripped = command.strip()
for prefix in ('adb -d shell ', 'adb shell '):
if stripped.startswith(prefix):
return self.run_adb_shell(stripped[len(prefix):])
command = command.replace('adb', self._adb_cmd(), 1)
if self.debug_mode:
self.log(f"CMD: {command}", "CMD")
try:
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
encoding='utf-8',
creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == 'win32' else 0
)
if self.debug_mode:
out = result.stdout.strip()
err = result.stderr.strip()
if out:
self.log(f" -> {out[:300]}", "CMD")
if err:
self.log(f" !! {err[:300]}", "ERROR")
if result.returncode == 0:
return True, result.stdout.strip()
else:
return False, result.stderr.strip()
except Exception as e:
return False, str(e)
def _get_apk_relative_path(self, apk_path, target_type):
if target_type == "system_ext":
root_dir = self.system_ext_dir
elif target_type == "priv-app":
root_dir = self.priv_apps_dir
else:
root_dir = self.apps_dir
try:
relative_path = Path(apk_path).relative_to(root_dir)
except Exception:
relative_path = Path(apk_path).name
return Path(relative_path)
def _get_target_paths(self, apk_path, target_type):
if target_type == "system_ext":
base_dir = "/system/system_ext/priv-app"
elif target_type == "priv-app":
base_dir = "/system/priv-app"
else:
base_dir = "/system/app"
relative_path = self._get_apk_relative_path(apk_path, target_type)
parts = [part for part in relative_path.parts if part not in ("", ".", "..")]
if len(parts) >= 2:
target_subdir = "/".join(parts[:-1])
apk_filename = parts[-1]
else:
target_subdir = Path(apk_path).stem
apk_filename = f"{Path(apk_path).stem}.apk"
target_dir = f"{base_dir}/{target_subdir}"
target_apk_path = f"{target_dir}/{apk_filename}"
return target_dir, target_apk_path
def _refresh_package_scan_after_system_push(self):
self.run_adb_su_command("sync")
self.run_adb_su_command("am force-stop android")
def push_single_apk(self, apk_path, apk_name, target_type="app"):
"""推送单个APK到系统分区,返回 (成功, 错误信息)"""
temp_apk_path = f"/data/local/tmp/a07_{target_type}_{apk_name}.apk"
target_dir, target_apk_path = self._get_target_paths(apk_path, target_type)
ok, err = self.run_adb_command(f'adb -d push "{apk_path}" {temp_apk_path}')
if not ok:
return False, self.t('push_fail').format(error=err)
self.run_adb_su_command(f"mkdir -p {self._quote_remote(target_dir)}")
ok, err = self.run_adb_su_command(
f"cp -f {self._quote_remote(temp_apk_path)} {self._quote_remote(target_apk_path)}"
)
if ok:
self.run_adb_su_command(f"rm -rf {self._quote_remote(f'{target_dir}/oat')}")
self.run_adb_su_command(f"rm -f {self._quote_remote(temp_apk_path)}")
if not ok:
return False, self.t('copy_fail').format(error=err)
return True, ""
def push_all_apks(self):
"""推送APK到系统分区(支持app和priv-app"""
if not self.check_device_connection():
return
if not self.vin:
messagebox.showwarning(self.t('dialog_warning'), self.t('dialog_need_vin'))
return
messagebox.showwarning(
self.t('dialog_flash_warning_title'),
self.t('dialog_flash_warning_body')
)
def do_push_all():
try:
self.log(self.t('log_flash_start'), "WARNING")
if not self.check_authorization(self.vin):
self.run_on_ui_thread(
messagebox.showerror,
self.t('dialog_auth_failed_title'),
self.t('dialog_auth_failed_body')
)
return
if not self.extract_password:
if not self.fetch_package_password():
self.run_on_ui_thread(
messagebox.showerror,
self.t('dialog_error'),
self.t('dialog_resource_failed_body')
)
return
if not self.check_package_extracted():
self.show_progress(True, is_push=False)
if not self.extract_package_silent():
self.show_progress(False, is_push=False)
self.run_on_ui_thread(
messagebox.showerror,
self.t('dialog_error'),
self.t('dialog_resource_failed_body')
)
return
self.show_progress(False, is_push=False)
if (not self.apps_dir or not self.apps_dir.exists()) and \
(not self.priv_apps_dir or not self.priv_apps_dir.exists()) and \
(not self.system_ext_dir or not self.system_ext_dir.exists()):
self.run_on_ui_thread(
messagebox.showerror,
self.t('dialog_error'),
self.t('dialog_resource_dir_missing')
)
return
self.show_progress(True, is_push=True)
ok, output = self.prepare_system_rw()
if not ok:
self.log(output or self.t('log_system_unlock_failed'), "ERROR")
return
self.run_adb_su_command('mkdir -p /data/local/tmp')
all_apks = []
if self.apps_dir and self.apps_dir.exists():
for apk in self.apps_dir.rglob("*.apk"):
all_apks.append((apk, "app"))
if self.priv_apps_dir and self.priv_apps_dir.exists():
for apk in self.priv_apps_dir.rglob("*.apk"):
all_apks.append((apk, "priv-app"))
if self.system_ext_dir and self.system_ext_dir.exists():
for apk in self.system_ext_dir.rglob("*.apk"):
all_apks.append((apk, "system_ext"))
if not all_apks:
# 缓存可能过期,强制重新解压
self.apps_dir = None
self.priv_apps_dir = None
self.system_ext_dir = None
self.temp_dir = None
if not self.fetch_package_password() or not self.extract_package_silent():
self.log(self.t('log_lang_pkg_missing'), "WARNING")
return
# 重新收集
all_apks = []
if self.apps_dir and self.apps_dir.exists():
for apk in self.apps_dir.rglob("*.apk"):
all_apks.append((apk, "app"))
if self.priv_apps_dir and self.priv_apps_dir.exists():
for apk in self.priv_apps_dir.rglob("*.apk"):
all_apks.append((apk, "priv-app"))
if self.system_ext_dir and self.system_ext_dir.exists():
for apk in self.system_ext_dir.rglob("*.apk"):
all_apks.append((apk, "system_ext"))
if not all_apks:
self.log(self.t('log_lang_pkg_missing'), "WARNING")
return
total = len(all_apks)
success_count = 0
aborted = False
for i, (apk_path, apk_type) in enumerate(all_apks, 1):
apk_name = apk_path.stem
ok, err = self.push_single_apk(apk_path, apk_name, apk_type)
if ok:
success_count += 1
else:
if "Read-only file system" in err or "read-only" in err.lower() or "只读" in err:
self.log(self.t('log_readonly_system'), "ERROR")
aborted = True
break
self.update_progress(i, total, self.t('progress_flashing'), is_push=True)
self.update_progress(total, total,
self.t('progress_flash_done') if not aborted else self.t('progress_aborted'),
is_push=True)
if success_count == total:
self.log(self.t('log_flash_done').format(total=total), "SUCCESS")
self._refresh_package_scan_after_system_push()
self.log(self.t('log_flash_reboot_required'), "WARNING")
elif success_count > 0:
self.log(self.t('log_flash_partial').format(success=success_count, total=total), "WARNING")
if not aborted:
self._refresh_package_scan_after_system_push()
self.log(self.t('log_flash_scan_required'), "WARNING")
finally:
self.restore_selinux_mode()
self.show_progress(False, is_push=True)
threading.Thread(target=do_push_all, daemon=True).start()
def install_all_apks(self):
"""批量安装APK — 手动选择文件夹"""
if not self.check_device_connection():
return
apk_dir = filedialog.askdirectory(title=self.t('dialog_select_apk_folder'))
if not apk_dir:
return
apk_files = list(Path(apk_dir).glob("*.apk"))
if not apk_files:
messagebox.showerror(self.t('dialog_error'), self.t('dialog_no_apk_in_folder'))
return
result = messagebox.askyesno(
self.t('dialog_confirm_install_title'),
self.t('dialog_confirm_install_folder').format(count=len(apk_files))
)
if not result:
return
def install():
self.show_progress(True, is_push=True)
total = len(apk_files)
self.log(self.t('log_batch_install_start').format(count=total), "INFO")
success_count = 0
try:
self.run_adb_command('adb -d shell setprop vecentek.model 1')
for i, apk_path in enumerate(apk_files, 1):
self.update_progress(i, total, self.t('progress_installing'), is_push=True)
success, _ = self.run_adb_command(f'adb -d install -r "{apk_path}"')
if success:
success_count += 1
self.update_progress(total, total, self.t('progress_install_done'), is_push=True)
self.grant_apkpure_install_permission_if_present()
if success_count == total:
self.log(self.t('log_install_all_success').format(count=total), "SUCCESS")
self.run_on_ui_thread(
messagebox.showinfo,
self.t('dialog_install_done_title'),
self.t('dialog_install_all_success').format(count=total)
)
elif success_count > 0:
self.log(self.t('log_install_partial').format(success=success_count, total=total), "WARNING")
self.run_on_ui_thread(
messagebox.showwarning,
self.t('dialog_install_partial_title'),
self.t('dialog_install_partial_body').format(
success=success_count,
failed=total - success_count
)
)
else:
self.log(self.t('log_install_failed'), "ERROR")
self.run_on_ui_thread(
messagebox.showerror,
self.t('dialog_install_failed_title'),
self.t('dialog_install_all_failed')
)
except Exception as e:
self.log(self.t('log_install_exception').format(error=str(e)), "ERROR")
self.run_on_ui_thread(
messagebox.showerror,
self.t('dialog_install_failed_title'),
self.t('dialog_install_exception').format(error=str(e))
)
finally:
self.run_adb_command('adb -d shell setprop vecentek.model 0')
self.show_progress(False, is_push=True)
threading.Thread(target=install, daemon=True).start()
def install_single_apk(self):
"""安装单个APK"""
# 检查设备连接
if not self.check_device_connection():
return
file_path = filedialog.askopenfilename(
title=self.t('dialog_select_apk_file'),
filetypes=[(self.t('filetype_apk'), "*.apk"), (self.t('filetype_all'), "*.*")]
)
if not file_path:
return
def install():
self.show_progress(True, is_push=True)
self.update_progress(50, 100, self.t('progress_installing'), is_push=True)
try:
self.run_adb_command('adb -d shell setprop vecentek.model 1')
success, _ = self.run_adb_command(f'adb -d install -r "{file_path}"')
self.update_progress(100, 100, self.t('progress_done'), is_push=True)
self.grant_apkpure_install_permission_if_present()
if success:
self.log(self.t('log_single_install_success'), "SUCCESS")
else:
self.log(self.t('log_single_install_failed'), "ERROR")
except Exception as e:
self.log(self.t('log_install_exception').format(error=str(e)), "ERROR")
finally:
self.run_adb_command('adb -d shell setprop vecentek.model 0')
self.show_progress(False, is_push=True)
threading.Thread(target=install, daemon=True).start()
def open_language_settings(self):
"""打开系统语言设置"""
if not self.check_device_connection():
return
self.run_adb_command('adb -d shell am start -a android.settings.LOCALE_SETTINGS')
def open_language_quick_set(self):
"""打开快捷语言设置弹窗"""
# 检查设备连接
if not self.check_device_connection():
return
# 创建弹窗
popup = tk.Toplevel(self.root)
popup.title(self.t('quick_lang_title'))
popup.geometry("520x320")
popup.configure(bg=self.colors['bg_dark'])
popup.resizable(False, False)
# 居中显示
popup.update_idletasks()
x = self.root.winfo_x() + (self.root.winfo_width() - 520) // 2
y = self.root.winfo_y() + (self.root.winfo_height() - 320) // 2
popup.geometry(f"+{x}+{y}")
popup.transient(self.root)
popup.grab_set()
# 标题
header = tk.Label(popup, text=self.t('quick_lang_header'),
font=('Microsoft YaHei', 13, 'bold'),
fg=self.colors['accent'],
bg=self.colors['bg_dark'])
header.pack(pady=(15, 10))
hint = tk.Label(popup, text=self.t('quick_lang_hint'),
font=('Microsoft YaHei', 9),
fg=self.colors['text_secondary'],
bg=self.colors['bg_dark'])
hint.pack(pady=(0, 12))
# 语言列表:(显示名, locale_code)
languages = [
(self.t('quick_lang_zh'), "zh-CN"),
(self.t('quick_lang_en'), "en-US"),
(self.t('quick_lang_ru'), "ru-RU"),
(self.t('quick_lang_fr'), "fr-FR"),
(self.t('quick_lang_es'), "es-ES"),
(self.t('quick_lang_pt'), "pt-BR"),
(self.t('quick_lang_it'), "it-IT"),
(self.t('quick_lang_ar'), "ar-SA"),
]
# 创建按钮容器
btn_frame = tk.Frame(popup, bg=self.colors['bg_dark'])
btn_frame.pack(pady=(0, 10))
btn_colors = [
self.colors['accent'], self.colors['info'],
self.colors['success'], self.colors['warning'],
'#e17055', '#00b894',
'#6c5ce7', '#0984e3',
]
for i, (label, locale) in enumerate(languages):
row = i // 4
col = i % 4
def make_cmd(loc=locale, lbl=label):
return lambda: self._quick_set_language(loc, lbl, popup)
btn = tk.Button(btn_frame, text=label,
command=make_cmd(),
font=('Microsoft YaHei', 10),
fg='white',
bg=btn_colors[i],
relief=tk.FLAT,
cursor='hand2',
width=12, height=2)
btn.grid(row=row, column=col, padx=5, pady=5)
# 底部分隔 + 打开系统设置入口
sep = tk.Frame(popup, bg=self.colors['border'], height=1)
sep.pack(fill=tk.X, padx=20, pady=(8, 6))
sys_btn = tk.Button(popup, text=self.t('quick_lang_system_button'),
command=lambda: self._open_sys_and_close(popup),
font=('Microsoft YaHei', 9),
fg=self.colors['text_secondary'],
bg=self.colors['bg_light'],
relief=tk.FLAT,
cursor='hand2')
sys_btn.pack(pady=(0, 10))
def _quick_set_language(self, locale_code, language_name, popup):
"""执行快捷语言设置"""
popup.destroy()
def do_set():
self.log(self.t('log_setting_language').format(language=language_name, locale=locale_code), "INFO")
success, output = self.run_adb_command(
f'adb -d shell settings put system system_locales {locale_code}'
)
if success:
self.log(self.t('log_language_set_success').format(language=language_name), "SUCCESS")
self.run_on_ui_thread(
messagebox.showinfo,
self.t('dialog_language_success_title'),
self.t('dialog_language_success_body').format(language=language_name)
)
else:
self.log(self.t('log_language_set_failed').format(error=output), "ERROR")
self.run_on_ui_thread(
messagebox.showerror,
self.t('dialog_language_failed_title'),
self.t('dialog_language_failed_body').format(error=output)
)
threading.Thread(target=do_set, daemon=True).start()
def _open_sys_and_close(self, popup):
"""关闭弹窗并打开系统语言设置"""
popup.destroy()
self.open_language_settings()
def open_timezone_settings(self):
"""打开时区设置"""
if not self.check_device_connection():
return
self.run_adb_command('adb -d shell am start -a android.settings.TIMEZONE_SETTINGS')
def open_android_settings(self):
"""打开安卓原生设置"""
if not self.check_device_connection():
return
self.run_adb_command('adb -d shell am start -a android.settings.SETTINGS')
def reboot_device(self):
"""重启设备"""
if not self.check_device_connection():
return
if messagebox.askyesno(self.t('dialog_confirm_reboot_title'), self.t('dialog_confirm_reboot_body')):
def do_reboot():
ok, output = self.run_adb_shell('reboot')
if ok:
self.log(self.t('log_rebooting'), "INFO")
self.update_device_status(False)
elif output:
self.log(self.t('log_reboot_failed').format(error=output), "ERROR")
threading.Thread(target=do_reboot, daemon=True).start()
def clear_extract_cache(self):
"""清理解压缓存目录。"""
cache_dir = self._cache_dir_path()
result = messagebox.askyesno(
self.t('dialog_clear_cache_title'),
self.t('dialog_clear_cache_body').format(cache_dir=cache_dir)
)
if not result:
self.log(self.t('log_clear_cache_cancelled'), "INFO")
return
def clear_cache():
self.show_progress(True, is_push=False)
try:
if cache_dir.exists():
shutil.rmtree(cache_dir, ignore_errors=True)
time.sleep(0.5)
self.apps_dir = None
self.priv_apps_dir = None
self.system_ext_dir = None
self.temp_dir = None
self.log(self.t('log_cache_cleared'), "SUCCESS")
self.run_on_ui_thread(
messagebox.showinfo,
self.t('dialog_cache_cleared_title'),
self.t('dialog_cache_cleared_body')
)
except Exception as e:
self.log(self.t('log_cache_clear_failed').format(error=e), "ERROR")
self.run_on_ui_thread(
messagebox.showerror,
self.t('dialog_cache_clear_failed_title'),
self.t('dialog_cache_clear_failed_body').format(error=e)
)
finally:
self.show_progress(False, is_push=False)
threading.Thread(target=clear_cache, daemon=True).start()
def is_placeholder_vin(self, value):
return value in (self.T['zh']['vin_placeholder'], self.T['en']['vin_placeholder'])
def is_placeholder_auth_code(self, value):
return value in (self.T['zh']['auth_code_placeholder'], self.T['en']['auth_code_placeholder'])
def _on_vin_input_focus_in(self, event):
"""输入框获得焦点时清除占位符"""
if self.is_placeholder_vin(self.vin_input.get()):
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():
self.vin_input.insert(0, self.t('vin_placeholder'))
self.vin_input.config(fg='#636e72')
def _on_auth_code_focus_in(self, event):
"""输入框获得焦点时清除占位符"""
if self.is_placeholder_auth_code(self.auth_code_input.get()):
self.auth_code_input.delete(0, tk.END)
self.auth_code_input.config(fg='#e0e0e0')
def _on_auth_code_focus_out(self, event):
"""输入框失去焦点时恢复占位符"""
if not self.auth_code_input.get():
self.auth_code_input.insert(0, self.t('auth_code_placeholder'))
self.auth_code_input.config(fg='#636e72')
def query_login_password_by_key(self):
"""通过 VIN 和授权码查询登录密码。"""
vin_text = self.vin_input.get().strip()
auth_code = self.auth_code_input.get().strip()
if (not vin_text or self.is_placeholder_vin(vin_text) or
not auth_code or self.is_placeholder_auth_code(auth_code)):
messagebox.showwarning(self.t('dialog_info'), self.t('key_need_input'))
return
self._set_pwd_result('key_querying', 'warning')
def do_query():
try:
api_url = "https://api.changan.softwindy.cn/api/authorizations/a07-dial-password"
url = f"{api_url}?{urlencode({'vin': vin_text, 'display_code': auth_code})}"
if self.debug_mode:
self.log(f"A07 DIAL PASSWORD URL: {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'))
def update_ui():
if data.get('success'):
password = data.get('data', {}).get('password')
if password:
self._set_pwd_result('key_query_success', 'success', password=password)
self.log(self.t('log_key_success'), "SUCCESS")
return
message = data.get('message') or self.t('key_request_failed')
self._set_pwd_result('key_query_failed', 'error', message=message)
self.log(self.t('log_key_failed').format(message=message), "ERROR")
self.run_on_ui_thread(update_ui)
except HTTPError as e:
try:
error_body = e.read().decode('utf-8', errors='replace')
error_data = json.loads(error_body) if error_body else {}
message = error_data.get('message') or self.t('key_request_failed')
except Exception:
message = self.t('key_request_failed')
def update_ui_http_error():
self._set_pwd_result('key_query_failed', 'error', message=message)
self.log(self.t('log_key_failed').format(message=message), "ERROR")
self.run_on_ui_thread(update_ui_http_error)
except Exception as e:
def update_ui_error():
self._set_pwd_result('key_request_failed', 'error')
if self.debug_mode:
self.log(self.t('log_key_request_failed_detail').format(error=str(e)), "ERROR")
else:
self.log(self.t('key_request_failed'), "ERROR")
self.run_on_ui_thread(update_ui_error)
threading.Thread(target=do_query, daemon=True).start()
def _toggle_debug(self, event=None):
"""切换调试模式(隐藏入口,Ctrl+Shift+D"""
if self.debug_mode:
self.debug_mode = False
self.log(self.t('log_debug_disabled'), "WARNING")
self.status_text.config(text=self.t('status_ready'))
self.refresh_device_status()
return
pwd = simpledialog.askstring(self.t('debug_title'), self.t('debug_prompt'), show='*', parent=self.root)
if not pwd:
return
def verify():
valid, message = self.verify_debug_mode_password(pwd)
def apply_result():
if valid:
self.debug_mode = True
self.update_device_status(True, "", True)
self.log(self.t('log_debug_enabled'), "WARNING")
self.status_text.config(text=self.t('debug_status'))
else:
messagebox.showwarning(self.t('dialog_error'), message or self.t('debug_password_wrong'))
self.run_on_ui_thread(apply_result)
threading.Thread(target=verify, daemon=True).start()
def _post_json(self, url, payload, timeout=10):
body = json.dumps(payload, ensure_ascii=False).encode('utf-8')
req = Request(
url,
data=body,
method='POST',
headers={
'Content-Type': 'application/json',
'User-Agent': 'Mozilla/5.0',
}
)
with urlopen(req, timeout=timeout) as response:
return json.loads(response.read().decode('utf-8'))
def verify_debug_mode_password(self, password):
try:
data = self._post_json(self.debug_password_api_url, {"password": password})
if data.get('success') is True:
return True, ""
return False, data.get('message') or self.t('debug_password_wrong')
except Exception as e:
return False, str(e)
def _debug_test_extract(self, event=None):
"""调试模式下仅测试资源包解压,不检查设备和授权"""
if not self.debug_mode:
messagebox.showwarning(self.t('debug_title'), self.t('debug_extract_need_enable'))
return
pwd = simpledialog.askstring(self.t('debug_extract_title'), self.t('debug_extract_prompt'),
show='*', parent=self.root)
if not pwd:
return
def do_extract():
old_password = self.extract_password
old_apps_dir = self.apps_dir
old_priv_apps_dir = self.priv_apps_dir
old_temp_dir = self.temp_dir
self.extract_password = pwd
try:
self.show_progress(True, is_push=False)
if self.extract_package_silent():
self.log(self.t('log_debug_extract_success'), "SUCCESS")
self.run_on_ui_thread(
messagebox.showinfo,
self.t('debug_extract_success_title'),
self.t('debug_extract_success_body').format(path=self.temp_dir)
)
else:
self.log(self.t('log_debug_extract_failed'), "ERROR")
self.run_on_ui_thread(
messagebox.showerror,
self.t('debug_extract_failed_title'),
self.t('debug_extract_failed_body')
)
finally:
self.show_progress(False, is_push=False)
self.extract_password = old_password
self.apps_dir = old_apps_dir
self.priv_apps_dir = old_priv_apps_dir
self.temp_dir = old_temp_dir
threading.Thread(target=do_extract, daemon=True).start()
def install_apps(self):
"""安装App — 支持单选或多选APK文件"""
if not self.check_device_connection():
return
file_paths = filedialog.askopenfilenames(
title=self.t('dialog_select_apk_file'),
filetypes=[(self.t('filetype_apk'), "*.apk"), (self.t('filetype_all'), "*.*")]
)
if not file_paths:
return
count = len(file_paths)
result = messagebox.askyesno(
self.t('dialog_confirm_install_title'),
self.t('dialog_confirm_install_folder').format(count=count)
)
if not result:
return
def install():
self.show_progress(True, is_push=True)
self.log(self.t('log_batch_install_start').format(count=count), "INFO")
success_count = 0
try:
self.run_adb_command('adb -d shell setprop vecentek.model 1')
for i, file_path in enumerate(file_paths, 1):
apk_name = Path(file_path).stem
self.update_progress(
i, count,
self.t('progress_installing_apk').format(apk=apk_name),
is_push=True
)
success, _ = self.run_adb_command(f'adb -d install -r "{file_path}"')
if success:
self.log(f"✓ {apk_name}.apk", "SUCCESS")
success_count += 1
else:
self.log(f"✗ {apk_name}.apk", "ERROR")
self.update_progress(count, count, self.t('progress_install_done'), is_push=True)
self.grant_apkpure_install_permission_if_present()
if success_count == count:
self.log(self.t('log_install_all_success').format(count=count), "SUCCESS")
self.run_on_ui_thread(
messagebox.showinfo,
self.t('dialog_install_done_title'),
self.t('dialog_install_all_success').format(count=count)
)
elif success_count > 0:
self.log(self.t('log_install_partial').format(success=success_count, total=count), "WARNING")
self.run_on_ui_thread(
messagebox.showwarning,
self.t('dialog_install_partial_title'),
self.t('dialog_install_partial_body').format(
success=success_count,
failed=count - success_count
)
)
else:
self.log(self.t('log_install_failed'), "ERROR")
self.run_on_ui_thread(
messagebox.showerror,
self.t('dialog_install_failed_title'),
self.t('dialog_install_all_failed')
)
except Exception as e:
self.log(self.t('log_install_exception').format(error=str(e)), "ERROR")
self.run_on_ui_thread(
messagebox.showerror,
self.t('dialog_install_failed_title'),
self.t('dialog_install_exception').format(error=str(e))
)
finally:
self.run_adb_command('adb -d shell setprop vecentek.model 0')
self.show_progress(False, is_push=True)
threading.Thread(target=install, daemon=True).start()
def grant_apkpure_install_permission_if_present(self):
"""如果已安装 APKPure,则允许它安装未知来源应用。"""
package_name = "com.apkpure.aegon"
ok, output = self.run_adb_command(f"adb -d shell pm path {package_name}")
if not ok or "package:" not in (output or ""):
return
failed = []
for user_id in ("0", "10"):
ok, output = self.run_adb_command(
f"adb -d shell appops set --user {user_id} {package_name} REQUEST_INSTALL_PACKAGES allow"
)
if not ok:
failed.append(f"user {user_id}: {output}")
if not failed:
self.log(self.t('log_apkpure_permission_ok'), "SUCCESS")
else:
self.log(self.t('log_apkpure_permission_partial').format(details='; '.join(failed)), "WARNING")
def run(self):
"""运行程序"""
self.root.mainloop()
def main():
"""主函数"""
if sys.version_info < (3, 6):
print(startup_t('python_version_error'))
sys.exit(1)
try:
app = ADKAPKGUI()
app.run()
except Exception as e:
print(startup_t('startup_failed_console').format(error=e))
import traceback
traceback.print_exc()
messagebox.showerror(
startup_t('startup_failed_title'),
startup_t('startup_failed_dialog').format(error=e)
)
if __name__ == "__main__":
main()