Files
2026-07-07 14:37:04 +08:00

3402 lines
157 KiB
Python
Raw Permalink 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 sys
import subprocess
import json
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
import base64
import hashlib
import tempfile
import zlib
import struct
import atexit
import re
try:
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
except ImportError:
AESGCM = None
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 resource_dir_candidates(dir_name):
base_dir = get_app_dir()
candidates = []
if getattr(sys, 'frozen', False):
candidates.append(Path(getattr(sys, '_MEIPASS', base_dir)) / dir_name)
candidates.extend([
base_dir / dir_name,
base_dir / 'tools' / dir_name,
base_dir / 'shared' / dir_name,
base_dir.parent / dir_name,
base_dir.parent / 'tools' / dir_name,
base_dir.parent / 'shared' / dir_name,
])
unique = []
for candidate in candidates:
if candidate not in unique:
unique.append(candidate)
return unique
def find_resource_dir(dir_name):
candidates = resource_dir_candidates(dir_name)
for candidate in candidates:
if candidate.exists() and candidate.is_dir():
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_id():
if sys.platform != 'win32':
return
try:
import ctypes
app_id = 'YibinKeyi.Q05Lidar.Installer.1.0'
ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID(app_id)
except Exception:
pass
class ADKAPKGUI:
def __init__(self):
set_windows_app_id()
self.root = tk.Tk()
self.root.title("启源Q05Ultra+激光雷达")
self.root.geometry("900x660")
self.root.resizable(True, True)
self.set_window_icon()
self.root.after(200, self.set_window_icon)
self.default_window_size = (900, 660)
self.right_panel_width = 280
# 设置颜色主题
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 = dict(self.colors_dark)
# 多语言
self.lang = 'zh'
self.T = {
'zh': {
'title': '启源Q05Ultra+激光雷达',
'btn_permission': '🔓 获取权限',
'btn_push': '📦 刷入语言包',
'btn_install': '📱 安装App',
'btn_language': '🌐 语言设置',
'btn_timezone': '⏰ 时区设置',
'btn_settings': '⚙️ 安卓设置',
'btn_reboot': '🔄 重启设备',
'btn_disable_upgrade': '❌ 禁用升级',
'btn_clear_log': '🗑 清空日志',
'btn_query_pwd': '查询密码',
'btn_debug_fingerprint': '指纹测试',
'btn_debug_decrypt': '解密测试',
'btn_debug_extract': '解压测试',
'pwd_query_label': '工程密码查询:',
'vin_placeholder': '请输入VIN',
'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': '🔄 检查',
'btn_install_driver': '🧩 安装驱动',
'hint_factory': '🔧 关闭车辆WI-FI和4G网络,拨号获取的密码进入工程模式',
'hotspot_title': '📶 电脑热点',
'hotspot_start': '🔧 打开热点设置',
'hint_title': '💡 使用提示',
'lang_zh': '中',
'lang_en': 'EN',
'hotspot_name_detecting': '名称: 检测中...',
'hotspot_name_unset': '名称: 未配置',
'hotspot_name_value': '名称: {ssid}',
'hotspot_pwd_default': '密码: changan2024',
'hotspot_pwd_value': '密码: {password}',
'hotspot_status_off': '状态: 未启动',
'hotspot_status_value': '状态: {status}',
'hotspot_started': '已启动',
'hint_lines': [
'1. 确保电脑已开启热点',
'2. 拨号进入工厂模式,点击调试工具',
'3. 需要云端认证时,点击车机状态栏',
'Wi-Fi图标,连接上方显示的热点',
'4. 连接后点击车机“云端认证”按钮',
'5. 打开ADB后即可正常刷入语言包',
],
'msg_device_not_connected_title': '设备未连接',
'msg_device_not_connected': '请先连接设备并点击「检查」按钮刷新状态!',
'msg_warn_title': '警告',
'msg_error_title': '错误',
'msg_auth_failed_title': '授权失败',
'msg_need_vin': '请先刷新设备状态并获取VIN码',
'msg_device_unauthorized': '设备未授权',
'msg_done_title': '完成',
'msg_permission_done': '获取成功,设备正在重启',
'msg_permission_failed': '获取失败',
'msg_confirm_permission_title': '确认获取权限',
'msg_confirm_permission': '此操作将为当前设备获取权限。\n过程中请不要断电或拔线。\n\n是否继续?',
'msg_driver_missing_title': '驱动环境',
'msg_driver_missing': '驱动缺失,即将自动安装驱动。',
'msg_driver_install_confirm': '即将安装驱动环境,需要管理员权限。',
'msg_driver_install_done': '驱动安装完成。如设备仍无法识别,请重新插拔 USB 线缆。',
'msg_driver_install_failed': '驱动安装失败: {error}',
'status_debug': '🔧 调试模式',
'log_cleared': '日志已清空',
'log_lang_changed': '语言已切换为中文',
'log_debug_on': '🔧 调试模式已开启',
'log_debug_off': '调试模式已关闭',
'log_driver_found': '已检测到驱动环境',
'log_driver_missing': '未检测到驱动环境',
'log_driver_install_start': '正在安装驱动环境...',
'log_driver_install_success': '驱动环境安装完成',
'log_driver_install_failed': '驱动环境安装失败: {error}',
'log_permission_success': '获取成功',
'log_push_success': '语言包刷入完成,重启设备后生效',
'log_push_failed': '语言包刷入失败',
'msg_data_prepare_failed': '数据准备失败!',
'msg_resource_prepare_failed': '资源准备失败!',
'msg_resource_dir_missing': '资源目录未找到',
'msg_apk_folder_missing': '未找到apks文件夹!\n请在程序目录下创建apks文件夹并放入APK文件。',
'msg_apk_folder_empty': 'apks文件夹中没有找到APK文件!',
'msg_install_title': '确认安装',
'msg_install_selected': '已选择 {count} 个APK文件\n\n是否开始安装?',
'msg_install_found': '找到 {count} 个APK文件\n\n是否开始批量安装?',
'msg_install_done_title': '安装完成',
'msg_install_done': '成功安装 {count} 个APK',
'msg_partial_success_title': '部分成功',
'msg_partial_success': '成功: {success}\n失败: {failed}',
'msg_install_failed_title': '安装失败',
'msg_install_failed_all': '所有APK安装失败!',
'file_select_apk': '选择APK文件',
'file_type_apk': 'APK文件',
'file_type_all': '所有文件',
'progress_installing': '安装中',
'progress_done': '完成',
'progress_flashing': '正在刷入',
'progress_flash_done': '刷入完成',
'progress_extracting_percent': '资源准备中 {percent}%',
'progress_install_module': '安装资源',
'progress_push_files': '推送文件 {current}/{total}',
'quick_lang_title': '快捷语言设置',
'quick_lang_header': '选择目标语言',
'quick_lang_hint': '点击按钮即可将系统语言切换为对应语言,重启后生效',
'quick_lang_system': '⚙️ 打开系统语言设置(手动选择)',
'quick_lang_success_title': '设置成功',
'quick_lang_success': '系统语言已设置为 {language}\n\n⚠️ 请重启设备使其生效。',
'quick_lang_failed_title': '设置失败',
'quick_lang_failed': '语言设置失败!\n\n{output}',
'reboot_confirm_title': '确认重启',
'reboot_confirm': '确定要重启设备吗?',
'disable_confirm_title': '确认禁用升级',
'disable_confirm': '⚠️ 警告:禁用系统升级后,将无法接收系统更新!\n\n是否确定要禁用系统升级应用?',
'disable_success_title': '成功',
'disable_success': '系统升级已成功禁用!',
'disable_failed': '禁用失败:{output}',
'msg_prompt_title': '提示',
'msg_enter_vin': '请输入VIN码',
'pwd_result': '密码: *#{pwd}#*',
'pwd_failed': '失败: {message}',
'pwd_request_failed': '请求失败',
'debug_wrong_password': '密码错误',
'debug_verify_failed': '调试模式密码校验失败: {message}',
'debug_need_enable': '请先按 Ctrl+Shift+D 进入调试模式',
'debug_password_prompt': '请输入调试密码:',
'debug_password_verifying': '正在校验调试模式密码...',
'debug_decrypt_title': '解密测试',
'debug_decrypt_prompt': '未检测到VIN/设备。\n请输入 BOOT_KEY 或 boot-key 返回的 sessionKey:',
'debug_key_len_error': '密钥长度错误,应为32字节AES密钥',
'debug_key_format_error': '密钥格式错误: {error}',
'debug_need_vin': '请先刷新设备VIN,或在工程密码输入框填入VIN',
'log_decrypt_success': '解密测试成功: size={size}, sha256={sha}',
'log_package_extract_success': 'Q05_Lidar-package.bin 解压测试成功: APK数量={count}',
'log_package_extract_modules_success': 'Q05_Lidar-package.bin 解压测试成功: 资源数量={count}',
'hotspot_stopped': '未启动',
'log_device_disconnected': '设备已断开连接',
'log_cache_invalid': '已解压缓存无效: {reason}',
'err_missing_apps_dir': '缺少 apps 目录',
'err_no_apks': 'apps 目录没有 APK',
'err_zero_kb_apk': '发现 0KB APK: {files}',
'err_extract_wrong_password': '解压密码错误,请重新确认 Q05_Lidar-package.bin 密码',
'err_package_data': '资源包数据错误,可能是密码错误或 Q05_Lidar-package.bin 损坏',
'err_package_corrupt': '资源包损坏或不完整,请检查 Q05_Lidar-package.bin',
'err_extract_failed': '解压失败: {error}',
'err_extract_failed_code': '解压失败 (返回码 {code}),请检查密码是否正确',
'log_package_missing': '错误:未找到资源包 ({path})',
'log_extract_password_missing': '错误:解压密码未设置',
'log_7za_missing': '错误:未找到 7za.exe ({path})',
'log_extracting_package': '资源准备中...',
'progress_resource_loading': '资源准备中...',
'progress_resource_done': '资源准备完成',
'log_apps_dir_missing': '警告:未找到 apps 目录',
'log_extracted_resource_invalid': '解压后的资源无效: {reason}。已停止刷入,请检查解压密码或资源包。',
'log_resource_ready': '资源准备完成 (app: {count})',
'log_module_resource_ready': '资源准备完成',
'err_module_prop_missing': '资源包缺少资源配置',
'err_module_id_missing': '资源配置缺少 ID 字段',
'err_module_zip_invalid': '资源文件无效或缺少配置: {file}',
'err_duplicate_module_id': '资源包存在重复资源 ID',
'log_resource_prepare_failed_detail': '资源准备失败: {error}',
'log_resource_prepare_network': '资源准备失败,请检查网络连接后重试',
'log_package_file_missing': '未找到资源包文件',
'log_adb_missing': '未找到adb命令,请将ADB文件放入本目录',
'log_device_connected': '设备已连接',
'log_vin_unavailable': '无法获取VIN,请确认设备已进入工厂模式',
'log_device_disconnected_short': '设备未连接',
'log_auth_skip_debug': '调试模式: 跳过授权验证',
'log_auth_checking': '正在验证授权状态...',
'log_auth_ok': '授权验证通过!',
'log_vehicle_name': '车辆名称: {name}',
'log_auth_failed': '授权验证失败',
'log_query_vin': '当前VIN: {vin}',
'log_adb_required': '请先连接adb',
'unknown_error': '未知错误',
'log_data_prepare_failed_detail': '数据准备失败: {error}',
'log_boot_challenge_failed': '获取失败',
'log_boot_key_failed': '获取失败',
'log_boot_key_len_error': '获取失败',
'log_boot_key_success': '正在获取权限中',
'log_runtime_missing': '获取失败',
'log_base_apk_invalid': '获取失败',
'log_runtime_ready': '正在获取权限中',
'log_crypto_missing': '获取失败',
'log_resource_missing': '获取失败',
'log_resource_format_unsupported': '获取失败',
'log_resource_algorithm_unsupported': '获取失败',
'log_resource_decrypt_auth_failed': '获取失败',
'log_resource_decrypt_ready': '正在获取权限中',
'log_resource_decrypt_failed_detail': '获取失败',
'log_temp_img_deleted': '临时文件已清理',
'log_temp_img_delete_failed': '临时文件清理失败',
'progress_prepare_runtime': '正在获取权限中',
'progress_install_runtime': '正在获取权限中',
'progress_fetch_boot_key': '正在获取权限中',
'progress_reboot_fastboot': '正在获取权限中',
'progress_decrypt_init_boot': '正在获取权限中',
'progress_flash_init_boot': '正在获取权限中',
'progress_reboot_device': '正在获取权限中',
'log_runtime_install_failed': '获取失败',
'log_runtime_install_success': '正在获取权限中',
'log_fastboot_enter_failed': '获取失败',
'log_fastboot_wait': '正在获取权限,请不要关闭程序或断开数据连接!',
'log_fastboot_missing': '获取失败',
'log_init_boot_flash_failed': '获取失败',
'log_init_boot_flash_success': '正在获取权限中',
'log_fastboot_reboot_failed': '获取失败',
'log_push_failed_detail': 'push失败: {error}',
'log_install_failed_detail': 'install失败: {error}',
'log_prepare_resource': '正在准备资源...',
'log_flash_start': '开始刷入语言包...',
'log_open_magisk': '请在弹窗中点击“允许”授予权限',
'log_root_checking': '正在检测权限...',
'log_root_ok': '权限已授予',
'log_root_failed': '权限获取失败,请手动授予权限',
'msg_root_failed': '请手动授予权限后重试。',
'log_module_install_start': '开始安装资源',
'log_module_install_done': '资源安装完成',
'log_module_install_failed': '资源安装失败: {error}',
'log_push_file_failed': '文件推送失败: {file}: {error}',
'log_root_cmd_failed': '权限操作失败: {error}',
'log_no_lang_pkg': '未找到语言包文件',
'log_install_item_ok': '安装成功: {name}.apk',
'log_install_item_failed': '安装失败: {name}.apk',
'progress_installing_name': '安装中 ({name})',
'log_batch_install_start': '开始批量安装 {count} 个APK...',
'log_install_start': '开始安装 {count} 个APK...',
'log_install_ok': '安装成功',
'log_install_failed': '安装失败',
'log_install_done_all': '安装完成:全部 {count} 个成功',
'log_install_done_partial': '安装完成:{success}/{count} 成功',
'quick_lang_names': ['🇨🇳 中文', '英 English', '俄 Русский', '法 Français', '西 Español', '葡 Português', '意 Italiano', '阿 العربية'],
'log_quick_lang_setting': '正在设置系统语言为: {language} ({locale})',
'log_quick_lang_success': '语言已设置为 {language}',
'log_quick_lang_failed': '语言设置失败: {output}',
'log_rebooting': '设备正在重启...',
'log_disable_cancelled': '已取消禁用升级操作',
'log_disable_success': '系统升级已禁用',
'log_disable_failed': '禁用系统升级失败',
'pwd_unknown': '未知',
'pwd_query_failed_default': '查询失败',
'log_pwd_query_success': '密码查询成功 VIN={vin}',
'log_pwd_query_failed': '密码查询失败: {message}',
'log_pwd_query_request_failed': '密码查询请求失败: {error}',
'log_debug_fingerprint_start': '开始设备指纹测试...',
'log_debug_fingerprint_empty': '未采集到设备指纹信息,请检查ADB连接',
'log_debug_fingerprint_hash': '本地指纹摘要: {hash}',
'log_debug_decrypt_start': '开始 resource.dat 解密测试...',
'log_debug_decrypt_failed': 'resource.dat 解密测试失败',
'log_package_key_failed': 'package-key 获取失败',
'log_package_extract_failed': 'Q05_Lidar-package.bin 解压测试失败',
'log_hosts_admin_required': '需要管理员权限,请以管理员身份运行',
'log_cloud_prepare_failed': '云端文件准备失败: {error}',
'err_decode_failed': '解码失败',
'err_command_timeout': '命令超时',
'err_fastboot_timeout': '获取失败',
'log_hotspot_opening': '已打开热点设置,正在检测热点...',
'log_hotspot_detected': '检测到热点: {ssid} / {password}',
'log_hotspot_not_detected': '未检测到热点,请确认已开启',
'app_start_error_title': '错误',
'app_start_error': '程序启动失败: {error}',
},
'en': {
'title': 'Qiyuan Q05-Ultra+ Lidar',
'btn_permission': '🔓 Unlock',
'btn_push': '📦 Flash Lang Pkg',
'btn_install': '📱 Install App',
'btn_language': '🌐 Language',
'btn_timezone': '⏰ Timezone',
'btn_settings': '⚙️ Settings',
'btn_reboot': '🔄 Reboot',
'btn_disable_upgrade': '❌ Disable OTA',
'btn_clear_log': '🗑 Clear Log',
'btn_query_pwd': 'Query Pwd',
'btn_debug_fingerprint': 'Fingerprint',
'btn_debug_decrypt': 'Decrypt',
'btn_debug_extract': 'Extract',
'pwd_query_label': 'Password query:',
'vin_placeholder': 'Enter VIN',
'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',
'btn_install_driver': '🧩 Driver',
'hint_factory': '🔧 Turn off the vehicle WiFi & 4G, enter factory mode with dial code',
'hotspot_title': '📶 Hotspot',
'hotspot_start': '🔧 Open Hotspot Settings',
'hint_title': '💡 Tips',
'lang_zh': '中',
'lang_en': 'EN',
'hotspot_name_detecting': 'Name: detecting...',
'hotspot_name_unset': 'Name: not configured',
'hotspot_name_value': 'Name: {ssid}',
'hotspot_pwd_default': 'Password: changan2024',
'hotspot_pwd_value': 'Password: {password}',
'hotspot_status_off': 'Status: stopped',
'hotspot_status_value': 'Status: {status}',
'hotspot_started': 'started',
'hint_lines': [
'1. Turn on the PC hotspot',
'2. Enter factory mode from the dialer',
'then click the button named "调试工具"',
'3. Connect the head unit to the hotspot',
'shown above when cloud auth is needed',
'4. Tap the button named "云端认证", after',
'vehicle Wi-Fi connect to laptop',
'5. Enable ADB, then run the installer',
],
'msg_device_not_connected_title': 'Device disconnected',
'msg_device_not_connected': 'Connect the device and click Check first.',
'msg_warn_title': 'Warning',
'msg_error_title': 'Error',
'msg_auth_failed_title': 'Authorization failed',
'msg_need_vin': 'Refresh device status and get VIN first.',
'msg_device_unauthorized': 'Device is not authorized.',
'msg_done_title': 'Done',
'msg_permission_done': 'Permission granted. The device is rebooting.',
'msg_permission_failed': 'Permission failed',
'msg_confirm_permission_title': 'Confirm Unlock',
'msg_confirm_permission': 'This will grant permission for the current device.\nDo not disconnect power or USB during the process.\n\nContinue?',
'msg_driver_missing_title': 'Driver Environment',
'msg_driver_missing': 'Driver environment is missing. Driver installation will start automatically.',
'msg_driver_install_confirm': 'Driver environment installation requires administrator permission.',
'msg_driver_install_done': 'Driver installation completed. If the device is still not recognized, reconnect USB cable.',
'msg_driver_install_failed': 'Driver installation failed: {error}',
'status_debug': '🔧 Debug mode',
'log_cleared': 'Log cleared',
'log_lang_changed': 'Language switched to English',
'log_debug_on': '🔧 Debug mode enabled',
'log_debug_off': 'Debug mode disabled',
'log_driver_found': 'Driver environment detected',
'log_driver_missing': 'Driver environment not detected',
'log_driver_install_start': 'Installing driver environment...',
'log_driver_install_success': 'Driver environment installed',
'log_driver_install_failed': 'Driver environment installation failed: {error}',
'log_permission_success': 'Permission granted',
'log_push_success': 'Language package flashed. Reboot to apply changes.',
'log_push_failed': 'Language package flash failed',
'msg_data_prepare_failed': 'Data preparation failed.',
'msg_resource_prepare_failed': 'Resource preparation failed.',
'msg_resource_dir_missing': 'Resource directory not found.',
'msg_apk_folder_missing': 'APK folder not found.\nCreate an apks folder next to the app and put APK files in it.',
'msg_apk_folder_empty': 'No APK files were found in the apks folder.',
'msg_install_title': 'Confirm Install',
'msg_install_selected': '{count} APK file(s) selected.\n\nStart installation?',
'msg_install_found': '{count} APK file(s) found.\n\nStart batch installation?',
'msg_install_done_title': 'Install Complete',
'msg_install_done': 'Successfully installed {count} APK(s).',
'msg_partial_success_title': 'Partially Complete',
'msg_partial_success': 'Success: {success}\nFailed: {failed}',
'msg_install_failed_title': 'Install Failed',
'msg_install_failed_all': 'All APK installations failed.',
'file_select_apk': 'Select APK file',
'file_type_apk': 'APK files',
'file_type_all': 'All files',
'progress_installing': 'Installing',
'progress_done': 'Done',
'progress_flashing': 'Flashing',
'progress_flash_done': 'Flash complete',
'progress_extracting_percent': 'Preparing resources {percent}%',
'progress_install_module': 'Installing resources',
'progress_push_files': 'Pushing files {current}/{total}',
'quick_lang_title': 'Quick Language',
'quick_lang_header': 'Choose target language',
'quick_lang_hint': 'Tap a button to switch system language. Reboot to apply.',
'quick_lang_system': '⚙️ Open system language settings',
'quick_lang_success_title': 'Language Set',
'quick_lang_success': 'System language was set to {language}.\n\n⚠️ Reboot the device to apply it.',
'quick_lang_failed_title': 'Language Failed',
'quick_lang_failed': 'Language setting failed.\n\n{output}',
'reboot_confirm_title': 'Confirm Reboot',
'reboot_confirm': 'Reboot the device now?',
'disable_confirm_title': 'Confirm Disable OTA',
'disable_confirm': '⚠️ Warning: disabling system upgrade prevents system updates.\n\nDisable the system upgrade app?',
'disable_success_title': 'Success',
'disable_success': 'System upgrade app disabled.',
'disable_failed': 'Disable failed: {output}',
'msg_prompt_title': 'Prompt',
'msg_enter_vin': 'Enter VIN.',
'pwd_result': 'Password: *#{pwd}#*',
'pwd_failed': 'Failed: {message}',
'pwd_request_failed': 'Request failed',
'debug_wrong_password': 'Wrong password',
'debug_verify_failed': 'Debug mode password verification failed: {message}',
'debug_need_enable': 'Press Ctrl+Shift+D first.',
'debug_password_prompt': 'Debug password:',
'debug_password_verifying': 'Verifying debug mode password...',
'debug_decrypt_title': 'Decrypt Test',
'debug_decrypt_prompt': 'No VIN/device detected.\nEnter BOOT_KEY or sessionKey returned by boot-key:',
'debug_key_len_error': 'Invalid key length. Expected a 32-byte AES key.',
'debug_key_format_error': 'Invalid key format: {error}',
'debug_need_vin': 'Refresh device VIN first, or enter a VIN in the password query field.',
'log_decrypt_success': 'Decrypt test passed: size={size}, sha256={sha}',
'log_package_extract_success': 'Q05_Lidar-package.bin extract test passed: APK count={count}',
'log_package_extract_modules_success': 'Q05_Lidar-package.bin extract test passed: resource count={count}',
'hotspot_stopped': 'stopped',
'log_device_disconnected': 'Device disconnected',
'log_cache_invalid': 'Extract cache invalid: {reason}',
'err_missing_apps_dir': 'Missing apps directory',
'err_no_apks': 'No APK files in apps directory',
'err_zero_kb_apk': 'Found 0 KB APK: {files}',
'err_extract_wrong_password': 'Incorrect extraction password. Check the Q05_Lidar-package.bin password.',
'err_package_data': 'Package data error. The password may be wrong or Q05_Lidar-package.bin is damaged.',
'err_package_corrupt': 'Package is damaged or incomplete. Check Q05_Lidar-package.bin.',
'err_extract_failed': 'Extraction failed: {error}',
'err_extract_failed_code': 'Extraction failed (exit code {code}). Check whether the password is correct.',
'log_package_missing': 'Package not found: {path}',
'log_extract_password_missing': 'Extraction password is not set',
'log_7za_missing': '7za.exe not found: {path}',
'log_extracting_package': 'Preparing resources...',
'progress_resource_loading': 'Preparing resources...',
'progress_resource_done': 'Resources ready',
'log_apps_dir_missing': 'apps directory not found',
'log_extracted_resource_invalid': 'Extracted resources are invalid: {reason}. Flashing stopped. Check the password or package.',
'log_resource_ready': 'Resources ready (app: {count})',
'log_module_resource_ready': 'Resources ready',
'err_module_prop_missing': 'Resource config is missing from the package',
'err_module_id_missing': 'Resource config is missing the id field',
'err_module_zip_invalid': 'Resource file is invalid or missing config: {file}',
'err_duplicate_module_id': 'Duplicate resource ID found in package',
'log_resource_prepare_failed_detail': 'Resource preparation failed: {error}',
'log_resource_prepare_network': 'Resource preparation failed. Check the network connection and retry.',
'log_package_file_missing': 'Package file not found',
'log_adb_missing': 'adb not found. Put ADB files in this directory.',
'log_device_connected': 'Device connected',
'log_vin_unavailable': 'Unable to get VIN. Make sure the device is in factory mode.',
'log_device_disconnected_short': 'Device disconnected',
'log_auth_skip_debug': 'Debug mode: authorization skipped',
'log_auth_checking': 'Checking authorization...',
'log_auth_ok': 'Authorization verified.',
'log_vehicle_name': 'Vehicle name: {name}',
'log_auth_failed': 'Authorization failed',
'log_query_vin': 'Current VIN: {vin}',
'log_adb_required': 'Connect adb first.',
'unknown_error': 'Unknown error',
'log_data_prepare_failed_detail': 'Data preparation failed: {error}',
'log_boot_challenge_failed': 'Permission failed',
'log_boot_key_failed': 'Permission failed',
'log_boot_key_len_error': 'Permission failed',
'log_boot_key_success': 'Getting permission',
'log_runtime_missing': 'Permission failed',
'log_base_apk_invalid': 'Permission failed',
'log_runtime_ready': 'Getting permission',
'log_crypto_missing': 'Permission failed',
'log_resource_missing': 'Permission failed',
'log_resource_format_unsupported': 'Permission failed',
'log_resource_algorithm_unsupported': 'Permission failed',
'log_resource_decrypt_auth_failed': 'Permission failed',
'log_resource_decrypt_ready': 'Getting permission',
'log_resource_decrypt_failed_detail': 'Permission failed',
'log_temp_img_deleted': 'Temporary files cleaned',
'log_temp_img_delete_failed': 'Temporary file cleanup failed',
'progress_prepare_runtime': 'Getting permission',
'progress_install_runtime': 'Getting permission',
'progress_fetch_boot_key': 'Getting permission',
'progress_reboot_fastboot': 'Getting permission',
'progress_decrypt_init_boot': 'Getting permission',
'progress_flash_init_boot': 'Getting permission',
'progress_reboot_device': 'Getting permission',
'log_runtime_install_failed': 'Permission failed',
'log_runtime_install_success': 'Getting permission',
'log_fastboot_enter_failed': 'Permission failed',
'log_fastboot_wait': 'Getting permission. Do not close the program or disconnect the data cable.',
'log_fastboot_missing': 'Permission failed',
'log_init_boot_flash_failed': 'Permission failed',
'log_init_boot_flash_success': 'Getting permission',
'log_fastboot_reboot_failed': 'Permission failed',
'log_push_failed_detail': 'push failed: {error}',
'log_install_failed_detail': 'install failed: {error}',
'log_prepare_resource': 'Preparing resources...',
'log_flash_start': 'Starting language package flash...',
'log_open_magisk': 'Tap "允许" to grant permission when prompted.',
'log_root_checking': 'Checking permission...',
'log_root_ok': 'Permission granted',
'log_root_failed': 'Permission grant failed. Grant permission manually.',
'msg_root_failed': 'Grant permission manually, then retry.',
'log_module_install_start': 'Installing resources',
'log_module_install_done': 'Resources installed',
'log_module_install_failed': 'Resource installation failed: {error}',
'log_push_file_failed': 'File push failed: {file}: {error}',
'log_root_cmd_failed': 'Permission operation failed: {error}',
'log_no_lang_pkg': 'No language package files found',
'log_install_item_ok': 'Installed: {name}.apk',
'log_install_item_failed': 'Install failed: {name}.apk',
'progress_installing_name': 'Installing ({name})',
'log_batch_install_start': 'Starting batch install of {count} APK(s)...',
'log_install_start': 'Starting install of {count} APK(s)...',
'log_install_ok': 'Install succeeded',
'log_install_failed': 'Install failed',
'log_install_done_all': 'Install complete: all {count} succeeded',
'log_install_done_partial': 'Install complete: {success}/{count} succeeded',
'quick_lang_names': ['Chinese', 'English', 'Russian', 'French', 'Spanish', 'Portuguese', 'Italian', 'Arabic'],
'log_quick_lang_setting': 'Setting system language to {language} ({locale})',
'log_quick_lang_success': 'Language set to {language}',
'log_quick_lang_failed': 'Language setting failed: {output}',
'log_rebooting': 'Device is rebooting...',
'log_disable_cancelled': 'Disable OTA cancelled',
'log_disable_success': 'System upgrade app disabled',
'log_disable_failed': 'Failed to disable system upgrade app',
'pwd_unknown': 'Unknown',
'pwd_query_failed_default': 'Query failed',
'log_pwd_query_success': 'Password query succeeded VIN={vin}',
'log_pwd_query_failed': 'Password query failed: {message}',
'log_pwd_query_request_failed': 'Password query request failed: {error}',
'log_debug_fingerprint_start': 'Starting device fingerprint test...',
'log_debug_fingerprint_empty': 'No device fingerprint info collected. Check ADB connection.',
'log_debug_fingerprint_hash': 'Local fingerprint digest: {hash}',
'log_debug_decrypt_start': 'Starting resource.dat decrypt test...',
'log_debug_decrypt_failed': 'resource.dat decrypt test failed',
'log_package_key_failed': 'package-key fetch failed',
'log_package_extract_failed': 'Q05_Lidar-package.bin extract test failed',
'log_hosts_admin_required': 'Administrator permission required. Run as administrator.',
'log_cloud_prepare_failed': 'Cloud file preparation failed: {error}',
'err_decode_failed': 'Decode failed',
'err_command_timeout': 'Command timed out',
'err_fastboot_timeout': 'Permission failed',
'log_hotspot_opening': 'Opened hotspot settings. Detecting hotspot...',
'log_hotspot_detected': 'Hotspot detected: {ssid} / {password}',
'log_hotspot_not_detected': 'Hotspot not detected. Make sure it is turned on.',
'app_start_error_title': 'Error',
'app_start_error': 'Failed to start app: {error}',
}
}
self.base_dir = get_app_dir()
self.adb = self._find_tool('adb.exe', 'adb')
self.sz = self._find_tool('7za.exe', str(find_resource('7za.exe')))
self.fastboot = self._find_tool('fastboot.exe', 'fastboot')
self.driver_dir = find_resource_dir("usb_driver")
self.driver_inf = self.driver_dir / "android_winusb.inf"
self.package_file = find_resource("Q05_Lidar-package.bin")
self.runtime_file = self._find_data_file("runtime.dat", prefer_embedded=True)
self.resource_file = self._find_data_file("resource.dat", prefer_embedded=True)
self.extract_password = None
self.runtime_password = None
self.resource_key = None
self.apps_dir = None
self.temp_dir = None
self.module_zips = []
self.runtime_cache_dir = None
self.api_url = "https://api.changan.softwindy.cn/api/authorizations/auth-check"
self.boot_challenge_api_url = "https://api.changan.softwindy.cn/api/authorizations/boot-challenge"
self.boot_key_api_url = "https://api.changan.softwindy.cn/api/authorizations/boot-key"
self.debug_password_api_url = "https://api.changan.softwindy.cn/api/authorizations/verify-debug-mode-password"
self.tool_version = "Q05-Lidar_Installer/1.0.0"
self.vin = None
self.vehicle_name = ""
self.device_connected = False
self._refreshing = False # 防止并发刷新
self.debug_mode = False # 调试模式
self.driver_prompted = False
atexit.register(self.cleanup_cache_on_exit)
# 设置样式
self.setup_styles()
self.setup_ui()
self.root.protocol("WM_DELETE_WINDOW", self.on_close)
self.center_window()
# 检查环境
self.check_environment()
# 启动设备状态监控
self.start_device_monitor()
def set_window_icon(self):
"""Set the Tk window/taskbar icon at runtime; 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))
except Exception:
pass
def _find_tool(self, exe_name, fallback):
candidates = resource_candidates(exe_name)
candidates.extend([
self.base_dir / "tools" / exe_name,
self.base_dir.parent / "tools" / exe_name,
Path(r"D:\UserData\adb-fastboot") / exe_name,
])
for candidate in candidates:
if candidate.exists():
return str(candidate)
return fallback
def _find_data_file(self, file_name, prefer_embedded=False):
candidates = []
if prefer_embedded and getattr(sys, 'frozen', False):
candidates.append(Path(sys._MEIPASS) / file_name)
candidates.extend([
self.base_dir / file_name,
self.base_dir / "payload" / file_name,
self.base_dir.parent / file_name,
self.base_dir.parent / "payload" / file_name,
])
if not prefer_embedded and getattr(sys, 'frozen', False):
candidates.append(Path(sys._MEIPASS) / file_name)
for candidate in candidates:
if candidate.exists():
return candidate
return candidates[0] if candidates else self.base_dir / file_name
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)
# 左侧内容区
left_frame = tk.Frame(main_frame, bg=self.colors['bg_dark'])
left_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
# 右侧提示面板
right_frame = tk.Frame(main_frame, bg=self.colors['bg_light'], relief=tk.RAISED, bd=1, width=self.right_panel_width)
right_frame.pack(side=tk.RIGHT, fill=tk.Y, padx=(10, 0))
right_frame.pack_propagate(False)
# 顶部标题栏
title_frame = tk.Frame(left_frame, bg=self.colors['bg_dark'], height=65)
title_frame.pack(fill=tk.X, pady=(0, 10))
title_frame.pack_propagate(False)
# 标题
self.title_label = tk.Label(title_frame,
text="🚀 启源Q05Ultra+激光雷达",
font=('Microsoft YaHei', 18, 'bold'),
fg=self.colors['accent'],
bg=self.colors['bg_dark'])
self.title_label.pack(side=tk.LEFT, padx=(0, 10), expand=True)
self.btn_lang_switch = tk.Button(title_frame, text="EN",
command=self.toggle_lang,
font=('Microsoft YaHei', 8, 'bold'),
fg=self.colors['accent'],
bg=self.colors['bg_dark'],
relief=tk.FLAT, cursor='hand2')
self.btn_lang_switch.pack(side=tk.RIGHT, padx=5, pady=(4, 0))
# 工程密码查询区域
pwd_query_frame = tk.Frame(left_frame, bg=self.colors['bg_light'], relief=tk.RAISED, bd=1)
pwd_query_frame.pack(fill=tk.X, pady=(0, 5), padx=5)
self.pwd_query_label = tk.Label(pwd_query_frame, text=self.t('pwd_query_label'),
font=('Microsoft YaHei', 9),
fg=self.colors['text'],
bg=self.colors['bg_light'])
self.pwd_query_label.pack(side=tk.LEFT, padx=(10, 5), pady=5)
self.vin_input = tk.Entry(pwd_query_frame,
font=('Consolas', 9),
bg='#2d2d3d',
fg='#636e72',
insertbackground='white',
relief=tk.FLAT,
width=20)
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=5)
self.btn_query_pwd = tk.Button(pwd_query_frame, text=self.t('btn_query_pwd'),
command=self.query_password_by_vin,
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=5)
self.pwd_result_label = tk.Label(pwd_query_frame, text="",
font=('Microsoft YaHei', 9, 'bold'),
fg=self.colors['success'],
bg=self.colors['bg_light'])
self.pwd_result_label.pack(side=tk.LEFT, padx=10, pady=5)
# 工厂模式提示
factory_hint_frame = tk.Frame(left_frame, bg=self.colors['bg_dark'])
factory_hint_frame.pack(fill=tk.X, pady=(0, 3))
self.hint_label = tk.Label(factory_hint_frame, text=self.t('hint_factory'),
font=('Microsoft YaHei', 8),
fg=self.colors['warning'],
bg=self.colors['bg_dark'])
self.hint_label.pack(side=tk.LEFT, padx=2)
# 按钮区域(两排)
button_frame = tk.Frame(left_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': 13
}
# 第一排按钮
row1_frame = tk.Frame(button_frame, bg=self.colors['bg_light'])
row1_frame.pack(pady=(8, 4))
self.btn_permission = tk.Button(row1_frame, text=self.t('btn_permission'),
command=self.prepare_lidar_permission,
bg=self.colors['warning'],
**btn_params)
self.btn_permission.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_disable_upgrade'),
command=self.on_disable_upgrade,
bg=self.colors['error'],
**btn_params)
self.btn_exit.pack(side=tk.LEFT, padx=4)
self.debug_button_frame = tk.Frame(button_frame, bg=self.colors['bg_light'])
self.btn_debug_fingerprint = tk.Button(self.debug_button_frame, text=self.t('btn_debug_fingerprint'),
command=self.debug_test_device_fingerprint,
bg=self.colors['info'],
**btn_params)
self.btn_debug_fingerprint.pack(side=tk.LEFT, padx=4)
self.btn_debug_decrypt = tk.Button(self.debug_button_frame, text=self.t('btn_debug_decrypt'),
command=self.debug_test_resource_decrypt,
bg=self.colors['info'],
**btn_params)
self.btn_debug_decrypt.pack(side=tk.LEFT, padx=4)
self.btn_debug_extract = tk.Button(self.debug_button_frame, text=self.t('btn_debug_extract'),
command=self.debug_test_package_extract,
bg=self.colors['info'],
**btn_params)
self.btn_debug_extract.pack(side=tk.LEFT, padx=4)
# 设备状态栏(横条)
status_bar_frame = tk.Frame(left_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'])
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'])
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'])
self.auth_label.pack(side=tk.LEFT, padx=(5, 0))
# 操作按钮:检查靠近设备状态,安装驱动在检查右侧
status_actions_frame = tk.Frame(status_bar_frame, bg=self.colors['bg_light'])
status_actions_frame.pack(side=tk.RIGHT, padx=10, pady=5)
self.btn_refresh = tk.Button(status_actions_frame, text=self.t('btn_refresh'),
command=lambda: self.refresh_device_status(force=True),
font=('Microsoft YaHei', 8),
fg=self.colors['accent'],
bg=self.colors['bg_light'],
relief=tk.FLAT,
cursor='hand2')
self.btn_refresh.pack(side=tk.LEFT, padx=(0, 8))
self.btn_install_driver = tk.Button(status_actions_frame, text=self.t('btn_install_driver'),
command=self.install_fastboot_driver,
font=('Microsoft YaHei', 8),
fg=self.colors['warning'],
bg=self.colors['bg_light'],
relief=tk.FLAT,
cursor='hand2')
self.btn_install_driver.pack(side=tk.LEFT)
# 解压进度条框架
progress_frame = tk.Frame(left_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(left_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('STATUS', 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(left_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.root.bind('<Control-Shift-D>', self._toggle_debug)
# ========== 右侧提示面板 ==========
# 热点信息卡片
hotspot_card = tk.Frame(right_frame, bg=self.colors['bg_dark'], relief=tk.RAISED, bd=1)
hotspot_card.pack(fill=tk.X, padx=5, pady=(10, 5))
hotspot_title_row = tk.Frame(hotspot_card, bg=self.colors['bg_dark'])
hotspot_title_row.pack(pady=(8, 5))
self.hotspot_title_icon_label = tk.Label(hotspot_title_row, text='',
font=('Segoe UI Emoji', 11),
fg=self.colors['accent'],
bg=self.colors['bg_dark'])
self.hotspot_title_icon_label.pack(side=tk.LEFT, padx=(0, 5))
self.hotspot_title_text_label = tk.Label(hotspot_title_row, text='',
font=('Microsoft YaHei', 11, 'bold'),
fg=self.colors['accent'],
bg=self.colors['bg_dark'])
self.hotspot_title_text_label.pack(side=tk.LEFT)
self.set_icon_title(self.hotspot_title_icon_label, self.hotspot_title_text_label, 'hotspot_title')
self.hotspot_ssid_label = tk.Label(hotspot_card, text=self.t('hotspot_name_detecting'),
font=('Microsoft YaHei', 9),
fg=self.colors['text'],
bg=self.colors['bg_dark'],
wraplength=self.right_panel_width - 34,
justify=tk.LEFT,
anchor='w')
self.hotspot_ssid_label.pack(anchor='w', fill=tk.X, padx=10, pady=2)
self.hotspot_pwd_label = tk.Label(hotspot_card, text=self.t('hotspot_pwd_default'),
font=('Microsoft YaHei', 9),
fg=self.colors['text'],
bg=self.colors['bg_dark'],
wraplength=self.right_panel_width - 34,
justify=tk.LEFT,
anchor='w')
self.hotspot_pwd_label.pack(anchor='w', fill=tk.X, padx=10, pady=2)
self.hotspot_status_label = tk.Label(hotspot_card, text=self.t('hotspot_status_off'),
font=('Microsoft YaHei', 9),
fg=self.colors['warning'],
bg=self.colors['bg_dark'],
wraplength=self.right_panel_width - 34,
justify=tk.LEFT,
anchor='w')
self.hotspot_status_label.pack(anchor='w', fill=tk.X, padx=10, pady=2)
self.btn_hotspot = tk.Button(hotspot_card, text=self.t('hotspot_start'),
command=self.start_hotspot_action,
font=('Microsoft YaHei', 8),
fg='white',
bg=self.colors['accent'],
relief=tk.FLAT,
cursor='hand2')
self.btn_hotspot.pack(pady=8, padx=10, fill=tk.X)
# 分隔线
tk.Frame(right_frame, bg=self.colors['border'], height=1).pack(fill=tk.X, padx=8, pady=5)
# 使用提示卡片
self.hint_card = tk.Frame(right_frame, bg=self.colors['bg_dark'], relief=tk.RAISED, bd=1)
self.hint_card.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
hint_title_row = tk.Frame(self.hint_card, bg=self.colors['bg_dark'])
hint_title_row.pack(pady=(8, 5))
self.hint_title_icon_label = tk.Label(hint_title_row, text='',
font=('Segoe UI Emoji', 11),
fg=self.colors['warning'],
bg=self.colors['bg_dark'])
self.hint_title_icon_label.pack(side=tk.LEFT, padx=(0, 5))
self.hint_title_text_label = tk.Label(hint_title_row, text='',
font=('Microsoft YaHei', 11, 'bold'),
fg=self.colors['warning'],
bg=self.colors['bg_dark'])
self.hint_title_text_label.pack(side=tk.LEFT)
self.set_icon_title(self.hint_title_icon_label, self.hint_title_text_label, 'hint_title')
self.hint_line_labels = []
self._render_hint_lines()
# 绑定悬停效果
self.bind_hover_effects()
def bind_hover_effects(self):
"""绑定按钮悬停效果"""
buttons = [self.btn_permission, 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, self.btn_query_pwd,
self.btn_hotspot, self.btn_debug_fingerprint,
self.btn_debug_decrypt, self.btn_debug_extract]
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, win_h = self.default_window_size
x = (screen_w - win_w) // 2
y = (screen_h - win_h) // 2
self.root.geometry(f"{win_w}x{win_h}+{max(0, x)}+{max(0, y)}")
def run_on_ui_thread(self, func, *args, **kwargs):
"""将函数调度到主线程执行,确保线程安全"""
self.root.after(0, lambda: func(*args, **kwargs))
def t(self, key):
return self.T.get(self.lang, self.T['zh']).get(key, key)
def tf(self, key, **kwargs):
return str(self.t(key)).format(**kwargs)
def split_icon_text(self, text):
text = str(text)
if ' ' in text:
icon, label = text.split(' ', 1)
if any(ord(ch) > 0x2600 for ch in icon):
return icon, label
return '', text
def set_icon_title(self, icon_label, text_label, key):
icon, label = self.split_icon_text(self.t(key))
if icon_label:
icon_label.config(text=icon)
if text_label:
text_label.config(text=label)
def is_placeholder_vin(self, text):
return text in (self.T['zh']['vin_placeholder'], self.T['en']['vin_placeholder'])
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()
self.log(self.t('log_lang_changed'), "INFO")
def _render_hint_lines(self):
if not getattr(self, 'hint_card', None):
return
for label in getattr(self, 'hint_line_labels', []):
label.destroy()
self.hint_line_labels = []
for line in self.t('hint_lines'):
label = tk.Label(
self.hint_card,
text=line,
font=('Microsoft YaHei', 8),
fg=self.colors['text_secondary'],
bg=self.colors['bg_dark'],
justify=tk.LEFT,
anchor='w',
wraplength=self.right_panel_width - 34
)
label.pack(anchor='w', fill=tk.X, padx=10, pady=1)
self.hint_line_labels.append(label)
def _refresh_ui_texts(self):
t = self.t
widgets = [
(getattr(self, 'title_label', None), 'title', None),
(getattr(self, 'btn_permission', None), 'btn_permission', 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_disable_upgrade', None),
(getattr(self, 'btn_clear', None), 'btn_clear_log', None),
(getattr(self, 'btn_query_pwd', None), 'btn_query_pwd', None),
(getattr(self, 'btn_debug_fingerprint', None), 'btn_debug_fingerprint', None),
(getattr(self, 'btn_debug_decrypt', None), 'btn_debug_decrypt', None),
(getattr(self, 'btn_debug_extract', None), 'btn_debug_extract', None),
(getattr(self, 'log_title_label', None), 'log_title', 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),
(getattr(self, 'btn_install_driver', None), 'btn_install_driver', None),
(getattr(self, 'hint_label', None), 'hint_factory', None),
(getattr(self, 'pwd_query_label', None), 'pwd_query_label', None),
(getattr(self, 'btn_hotspot', None), 'hotspot_start', None),
]
for w, key, _ in widgets:
if w: w.config(text=t(key))
self.set_icon_title(getattr(self, 'hotspot_title_icon_label', None),
getattr(self, 'hotspot_title_text_label', None),
'hotspot_title')
self.set_icon_title(getattr(self, 'hint_title_icon_label', None),
getattr(self, 'hint_title_text_label', None),
'hint_title')
self.btn_lang_switch.config(text=t('lang_en') if self.lang == 'zh' else t('lang_zh'))
if getattr(self, 'status_text', None):
self.status_text.config(text=t('status_debug') if self.debug_mode else t('status_ready'))
if self.is_placeholder_vin(self.vin_input.get()):
self.vin_input.delete(0, tk.END)
self.vin_input.insert(0, t('vin_placeholder'))
self._render_hint_lines()
self.refresh_hotspot_display()
if getattr(self, 'device_status_label', None):
self._update_device_status_impl(self.device_connected, self.vin,
getattr(self, '_last_authorized', False))
def _sanitize_user_log_message(self, message):
"""Hide low-level device command details from normal operator logs."""
text = str(message)
replacements = [
(r'adb(?:\.exe)?(?:\s+-d)?\s+shell(?:\s+root)?', '权限获取'),
(r'/debug_ramdisk/su(?:\s+-c)?', '权限获取'),
(r'\bsu\s+-c\b', '权限获取'),
(r'Shell 权限', '权限'),
(r'\bShell permission\b', 'permission'),
(r'\b[Rr]oot command\b', 'permission operation'),
(r'\b[Rr]oot\b', 'permission'),
]
for pattern, replacement in replacements:
text = re.sub(pattern, replacement, text)
return text
def _log_impl(self, message, level="INFO"):
"""日志写入的实际实现(必须在主线程调用)"""
if not self.debug_mode and level in ("INFO", "CMD"):
return
if not self.debug_mode:
message = self._sanitize_user_log_message(message)
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 set_debug_buttons_visible(self, visible):
def apply():
if visible:
self.debug_button_frame.pack(pady=(4, 8))
else:
self.debug_button_frame.pack_forget()
self.run_on_ui_thread(apply)
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('msg_device_not_connected_title'), self.t('msg_device_not_connected'))
return False
return True
def start_device_monitor(self):
"""启动设备状态监控(每5秒检查一次)"""
def monitor():
while True:
try:
result = subprocess.run(f'{self.adb} -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 run_adb_shell(self, shell_command, timeout=15):
"""执行设备权限命令,自动静默输入设备密码 adb36987。
静默执行,不显示 adb 原始输出,仅返回结果。"""
if self.debug_mode:
self.log(f"CMD: adb shell {shell_command}", "CMD")
try:
proc = subprocess.Popen(
f'{self.adb} -d shell {shell_command}',
shell=True,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
stdout, stderr = proc.communicate(input='adb36987\n', timeout=timeout)
# 过滤密码提示行
output_lines = []
for line in stdout.split('\n'):
stripped = line.strip()
if 'please input verify password' in stripped.lower():
continue
if stripped == 'verify success!':
continue
output_lines.append(line)
output = '\n'.join(output_lines).strip()
if proc.returncode == 0:
if self.debug_mode:
self.log(f"CMD OK: {output[:200]}", "CMD")
return True, output
else:
if self.debug_mode:
self.log(f"CMD FAIL: {stderr.strip()[:200]}", "CMD")
return False, stderr.strip()
except subprocess.TimeoutExpired:
proc.kill()
proc.communicate()
return False, self.t('err_command_timeout')
except Exception as e:
return False, str(e)
# ============================================================
# 原始 adb 命令(用于 adb push / adb install 等不需要 shell 的操作)
# ============================================================
def run_adb_command(self, command):
"""执行原始 adb 命令(adb push / adb install 等,无需 shell 密码)。
静默执行,不显示 adb 原始输出,仅返回结果。"""
command = command.replace('adb', self.adb, 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')
if result.returncode == 0:
if self.debug_mode:
self.log(f"CMD OK: {result.stdout.strip()[:200]}", "CMD")
return True, result.stdout.strip()
else:
if self.debug_mode:
self.log(f"CMD FAIL: {result.stderr.strip()[:200]}", "CMD")
return False, result.stderr.strip()
except Exception as e:
if self.debug_mode:
self.log(f"CMD ERROR: {str(e)}", "CMD")
return False, str(e)
def shell_quote(self, value):
return "'" + str(value).replace("'", "'\"'\"'") + "'"
def run_root_command(self, command, timeout=60):
return self.run_adb_shell(
f'/debug_ramdisk/su -c {self.shell_quote(command)}',
timeout=timeout
)
def open_magisk_and_check_root(self):
self.log(self.t('log_open_magisk'), "WARNING")
self.run_adb_shell('monkey -p com.topjohnwu.magisk -c android.intent.category.LAUNCHER 1', timeout=20)
time.sleep(3)
self.log(self.t('log_root_checking'), "STATUS")
ok, output = self.run_root_command("id", timeout=20)
if ok and "uid=0" in output:
self.log(self.t('log_root_ok'), "STATUS")
self.run_adb_shell('pm uninstall --user 0 com.topjohnwu.magisk', timeout=30)
return True
self.log(self.t('log_root_failed'), "ERROR")
return False
def disable_and_uninstall_preinstalled_packages(self):
self.run_adb_shell('pm disable-user com.incall.apps.softmanager', timeout=30)
package_names = [
"com.carcontrolhome.app",
"com.tinnove.netease.music",
"com.wtcl.electronicdirections",
]
for package_name in package_names:
self.run_adb_shell(f'pm disable-user {package_name}', timeout=30)
self.run_adb_shell(f'pm uninstall --user 0 {package_name}', timeout=30)
def install_magisk_module_zip(self, zip_path):
zip_path = Path(zip_path)
mod_id = self.read_module_id_from_zip(zip_path)
if not mod_id:
return False, self.t('err_module_id_missing')
if not zip_path.exists():
return False, self.tf('err_extract_failed', error=f"{zip_path} not found")
device_stage_dir = "/data/local/tmp/q05_lidar_modules"
device_zip = f"{device_stage_dir}/{mod_id}.zip"
device_module = f"/data/adb/modules/{mod_id}"
self.log(self.t('log_module_install_start'), "STATUS")
commands = [
f"rm -rf {self.shell_quote(device_module)}",
f"mkdir -p {self.shell_quote(device_stage_dir)} {self.shell_quote(device_module)}",
f"chmod 777 {self.shell_quote(device_stage_dir)}",
]
for command in commands:
ok, output = self.run_root_command(command, timeout=60)
if not ok:
return False, self.tf('log_root_cmd_failed', command=command, error=output)
ok, output = self.run_adb_command(f'adb -d push "{zip_path}" {device_zip}')
if not ok:
return False, self.tf('log_push_file_failed', file=zip_path.name, error=output)
install_commands = [
f"unzip -oq {self.shell_quote(device_zip)} -x 'META-INF/*' -d {self.shell_quote(device_module)}",
f"find {self.shell_quote(device_module)} -type d -exec chmod 755 {{}} \\;",
f"find {self.shell_quote(device_module)} -type f -exec chmod 644 {{}} \\;",
f"find {self.shell_quote(device_module)} -type f -name '*.sh' -exec chmod 755 {{}} \\;",
f"rm -f {self.shell_quote(device_zip)}",
]
for command in install_commands:
ok, output = self.run_root_command(command, timeout=300)
if not ok:
return False, self.tf('log_root_cmd_failed', command=command, error=output)
self.log(self.t('log_module_install_done'), "STATUS")
return True, mod_id
def check_package_extracted(self):
"""检查语言包是否已解压"""
has_module = bool(getattr(self, 'module_zips', None))
if not has_module and self.temp_dir and self.temp_dir.exists():
self.module_zips = self.find_module_zips(self.temp_dir)
has_module = bool(self.module_zips)
if has_module:
ok, reason = self._validate_extracted_modules()
if not ok:
self.log(self.tf('log_cache_invalid', reason=reason), "ERROR")
self._clear_extracted_cache()
return False
return has_module
def _validate_extracted_modules(self):
if not getattr(self, 'module_zips', None):
return False, self.t('err_module_prop_missing')
module_ids = []
for zip_path in self.module_zips:
mod_id = self.read_module_id_from_zip(zip_path)
if not mod_id:
return False, self.tf('err_module_zip_invalid', file=zip_path.name)
module_ids.append(mod_id)
if len(set(module_ids)) != len(module_ids):
return False, self.t('err_duplicate_module_id')
return True, ""
def _clear_extracted_cache(self):
if self.temp_dir and self.temp_dir.exists():
shutil.rmtree(self.temp_dir, ignore_errors=True)
time.sleep(0.5)
self.apps_dir = None
self.module_zips = []
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, return_code):
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('err_extract_wrong_password')
if "data error" in text:
return self.t('err_package_data')
if "headers error" in text or "unexpected end" in text:
return self.t('err_package_corrupt')
if err_msg.strip():
return self.tf('err_extract_failed', error=err_msg.strip()[:300])
return self.tf('err_extract_failed_code', code=return_code)
def _run_7za_extract_with_progress(self, archive_path, output_dir, password=None, progress_cb=None):
base_command = [self.sz, 'x', str(archive_path)]
if password:
base_command.append(f'-p{password}')
base_command.extend([f'-o{output_dir}', '-y'])
progress_command = base_command + ['-bsp1']
ok, output = self._run_7za_streaming(progress_command, progress_cb)
if ok:
return True, output
if "incorrect command line" in output.lower() or "unsupported switch" in output.lower():
return self._run_7za_extract_basic(archive_path, output_dir, password)
return False, output
def _run_7za_streaming(self, command, progress_cb=None):
output_parts = []
try:
proc = subprocess.Popen(
command,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == 'win32' else 0
)
last_percent = -1
progress_text = ""
while True:
chunk = proc.stdout.read(256)
if not chunk:
break
text = self._decode_process_output(chunk)
output_parts.append(text)
progress_text = (progress_text + text)[-2000:]
matches = re.findall(r'(?<!\d)(\d{1,3})\s*%', progress_text)
for match in matches:
percent = max(0, min(100, int(match)))
if percent != last_percent:
if progress_cb:
progress_cb(percent, 100, self.tf('progress_extracting_percent', percent=percent))
last_percent = percent
return_code = proc.wait()
return return_code == 0, ''.join(output_parts)
except Exception as e:
return False, str(e)
def _run_7za_extract_basic(self, archive_path, output_dir, password=None):
command = [self.sz, 'x', str(archive_path)]
if password:
command.append(f'-p{password}')
command.extend([f'-o{output_dir}', '-y'])
result = subprocess.run(
command,
capture_output=True,
creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == 'win32' else 0
)
output = self._decode_process_output((result.stderr or b'') + (result.stdout or b''))
return result.returncode == 0, output
def find_module_zips(self, root_dir):
root_dir = Path(root_dir)
candidates = []
modules_dir = root_dir / "modules"
search_roots = [modules_dir] if modules_dir.exists() else [root_dir]
for search_root in search_roots:
candidates.extend(sorted(search_root.rglob("*.zip")))
module_zips = []
for zip_path in candidates:
if self.read_module_id_from_zip(zip_path):
module_zips.append(zip_path)
return module_zips
def read_module_id_from_zip(self, zip_path):
try:
with zipfile.ZipFile(zip_path, 'r') as zf:
prop_name = self._find_module_prop_in_zip(zf)
if not prop_name:
return ""
raw = zf.read(prop_name)
for line in raw.decode('utf-8', errors='replace').splitlines():
line = line.strip()
if line.startswith("id="):
return line.split("=", 1)[1].strip()
except Exception:
return ""
return ""
def _find_module_prop_in_zip(self, zf):
names = zf.namelist()
normalized = {}
for name in names:
clean = name.replace("\\", "/").lstrip("./")
normalized[clean] = name
if "module.prop" in normalized:
return normalized["module.prop"]
candidates = []
for clean, original in normalized.items():
parts = [part for part in clean.split("/") if part]
if len(parts) == 2 and parts[-1] == "module.prop":
candidates.append((len(parts), original))
if candidates:
candidates.sort()
return candidates[0][1]
return ""
def extract_package_silent(self):
"""静默解压语言包(带进度)—— Q05-Lidar 使用 Magisk 模块结构"""
if not self.package_file.exists():
self.log(self.tf('log_package_missing', 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.tf('log_7za_missing', 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_Q05_Lidar"
# 如果已存在,先清理
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_extracting_package'), "INFO")
self.update_progress(0, 1, self.t('progress_resource_loading'))
ok, err_msg = self._run_7za_extract_with_progress(
self.package_file,
self.temp_dir,
self.extract_password,
self.update_progress
)
if not ok:
self.log(self._format_extract_error(err_msg, 1), "ERROR")
self._clear_extracted_cache()
return False
self.update_progress(1, 1, self.t('progress_resource_done'))
self.module_zips = self.find_module_zips(self.temp_dir)
if not self.module_zips:
self.log(self.t('err_module_prop_missing'), "ERROR")
self._clear_extracted_cache()
return False
ok, reason = self._validate_extracted_modules()
if not ok:
self.log(self.tf('log_extracted_resource_invalid', reason=reason), "ERROR")
self._clear_extracted_cache()
return False
self.log(self.t('log_module_resource_ready'), "INFO")
return True
except Exception as e:
if getattr(self, 'debug_mode', False):
self.log(self.tf('log_resource_prepare_failed_detail', error=str(e)), "ERROR")
import traceback
self.log(traceback.format_exc(), "ERROR")
else:
self.log(self.t('log_resource_prepare_network'), "ERROR")
self._clear_extracted_cache()
return False
def check_environment(self):
"""检查环境"""
# 修改hosts文件
self.modify_hosts()
# 刷新热点显示
self.refresh_hotspot_display()
try:
result = subprocess.run(f'{self.adb} 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_file_missing'), "WARNING")
else:
self.log(self.t('log_adb_missing'), "ERROR")
except FileNotFoundError:
self.log(self.t('log_adb_missing'), "ERROR")
self.root.after(800, self.check_fastboot_driver_on_startup)
def _run_command_capture(self, command, timeout=30):
creationflags = subprocess.CREATE_NO_WINDOW if sys.platform == 'win32' else 0
result = subprocess.run(
command,
capture_output=True,
text=True,
timeout=timeout,
creationflags=creationflags
)
output = (result.stdout or "") + (result.stderr or "")
return result.returncode, output.strip()
def ask_ok_cancel_on_ui_thread(self, title, message):
result = {"value": False}
done = threading.Event()
def prompt():
try:
result["value"] = messagebox.askokcancel(title, message, parent=self.root)
finally:
done.set()
self.run_on_ui_thread(prompt)
done.wait()
return result["value"]
def is_fastboot_driver_installed(self):
if sys.platform != 'win32':
return True, ""
try:
code, output = self._run_command_capture(['pnputil', '/enum-drivers'], timeout=40)
if code != 0:
return False, output or "pnputil enum failed"
normalized = output.lower()
installed = (
'android_winusb.inf'.lower() in normalized
or 'android bootloader interface' in normalized
or 'android adb interface' in normalized
or 'fastboot' in normalized
)
return installed, output
except Exception as e:
return False, str(e)
def _install_driver_with_uac(self, inf_path):
quoted_inf = str(inf_path).replace("'", "''")
ps_command = (
"$proc = Start-Process -FilePath pnputil "
"-ArgumentList @('/add-driver', '{0}', '/install') "
"-Verb RunAs -WindowStyle Hidden -PassThru; "
"if ($null -eq $proc) {{ exit 1 }}; "
"$proc.WaitForExit(); "
"Write-Output $proc.ExitCode"
).format(quoted_inf)
creationflags = subprocess.CREATE_NO_WINDOW if sys.platform == 'win32' else 0
result = subprocess.run(
['powershell', '-NoProfile', '-Command', ps_command],
capture_output=True,
text=True,
creationflags=creationflags
)
output = ((result.stdout or "") + (result.stderr or "")).strip()
exit_code = None
for line in reversed(output.splitlines()):
text = line.strip()
if text.isdigit():
exit_code = int(text)
break
if exit_code is None and result.returncode == 0:
exit_code = 0
return exit_code == 0, output or f"powershell exit={result.returncode}"
def install_fastboot_driver(self, prompt=True):
def worker():
try:
if not self.driver_inf.exists():
error = self.tf('msg_driver_install_failed', error=f"{self.driver_inf} not found")
self.log(self.tf('log_driver_install_failed', error=f"{self.driver_inf} not found"), "ERROR")
self.run_on_ui_thread(lambda: messagebox.showerror(self.t('msg_driver_missing_title'), error))
return
if prompt:
confirmed = self.ask_ok_cancel_on_ui_thread(
self.t('msg_driver_missing_title'),
self.t('msg_driver_install_confirm')
)
if not confirmed:
return
self.log(self.t('log_driver_install_start'), "STATUS")
ok, output = self._install_driver_with_uac(self.driver_inf)
if not ok:
self.log(self.tf('log_driver_install_failed', error=output or self.t('unknown_error')), "ERROR")
self.run_on_ui_thread(
lambda: messagebox.showerror(
self.t('msg_driver_missing_title'),
self.tf('msg_driver_install_failed', error=output or self.t('unknown_error')),
parent=self.root
)
)
return
self.log(self.t('log_driver_install_success'), "SUCCESS")
self.run_on_ui_thread(
lambda: messagebox.showinfo(
self.t('msg_driver_missing_title'),
self.t('msg_driver_install_done'),
parent=self.root
)
)
except Exception as e:
self.log(self.tf('log_driver_install_failed', error=str(e)), "ERROR")
self.run_on_ui_thread(
lambda: messagebox.showerror(
self.t('msg_driver_missing_title'),
self.tf('msg_driver_install_failed', error=str(e)),
parent=self.root
)
)
threading.Thread(target=worker, daemon=True).start()
def check_fastboot_driver_on_startup(self):
if self.driver_prompted:
return
self.driver_prompted = True
def worker():
installed, detail = self.is_fastboot_driver_installed()
if installed:
return
if self.debug_mode and detail:
self.log(detail[:500], "CMD")
def notify_and_install():
messagebox.showinfo(
self.t('msg_driver_missing_title'),
self.t('msg_driver_missing'),
parent=self.root
)
self.install_fastboot_driver(prompt=False)
self.run_on_ui_thread(notify_and_install)
threading.Thread(target=worker, daemon=True).start()
def refresh_device_status(self, force=False):
"""刷新设备状态 —— 逸动版使用 ca.car.vin 获取 VIN"""
# 防止并发刷新(手动点击「检查」时强制忽略锁)
if self._refreshing and not force:
return
self._refreshing = True
def refresh():
was_connected = self.device_connected
# 检查设备连接
result = subprocess.run(f'{self.adb} -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'), "INFO")
# 获取VIN —— 逸动车型使用 ca.car.vin
success, vin_output = self.run_adb_shell(
'settings get system ca.car.vin')
vin = vin_output.strip() if success else ''
if vin:
self.log(f"VIN: {vin}", "STATUS")
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_disconnected_short'), "WARNING")
self.update_device_status(False)
self._refreshing = False
threading.Thread(target=refresh, daemon=True).start()
def check_authorization(self, vin):
"""检查授权"""
if self.debug_mode:
self.log(self.t('log_auth_skip_debug'), "WARNING")
return True
self.log(self.t('log_auth_checking'), "STATUS")
try:
authorized, vehicle_name, _ = self.query_authorization_info(vin)
if authorized:
self.log(self.t('log_auth_ok'), "STATUS")
if vehicle_name:
self.vehicle_name = vehicle_name
self.log(self.tf('log_vehicle_name', name=vehicle_name), "STATUS")
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}?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 ""
if data.get('authorized') is True and str(vehicle_name).strip():
self.vehicle_name = str(vehicle_name).strip()
return data.get('authorized') is True, str(vehicle_name).strip(), data
def fetch_package_password(self):
"""从服务端获取资源包解压密码"""
if not self.vin:
self.log(self.t('log_adb_required'), "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_auth_failed'), "ERROR")
return False
pwd_api_url = "https://api.changan.softwindy.cn/api/authorizations/package-key"
query = urlencode({
"vin": self.vin,
"vehicleName": vehicle_name,
})
url = f"{pwd_api_url}?{query}"
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']
return True
else:
self.log(self.tf('log_data_prepare_failed_detail', error=data.get('message', self.t('unknown_error'))), "ERROR")
return False
except Exception as e:
self.log(self.tf('log_data_prepare_failed_detail', error=str(e)), "ERROR")
return False
def fetch_runtime_password(self):
"""从服务端获取 runtime.dat 解压密码。默认复用 package-key,便于后端保持一个资源密码。"""
if self.runtime_password:
return True
if not self.extract_password:
if not self.fetch_package_password():
return False
self.runtime_password = self.extract_password
return True
def fetch_resource_key(self):
"""通过 boot challenge 协议获取 resource.dat 解密密钥。"""
if self.resource_key:
return True
if not self.vin:
self.log(self.t('log_adb_required'), "ERROR")
return False
try:
device_info = self.collect_boot_device_info()
challenge_data = self._post_json(self.boot_challenge_api_url, {
"vin": self.vin,
"toolVersion": self.tool_version,
"deviceInfo": device_info,
})
challenge_payload = challenge_data.get('data', {}) if isinstance(challenge_data, dict) else {}
challenge_id = str(challenge_payload.get('challengeId') or '').strip()
nonce = str(challenge_payload.get('nonce') or '').strip()
if not challenge_data.get('success') or not challenge_id or not nonce:
self.log(self.tf('log_boot_challenge_failed', error=challenge_data.get('message', self.t('unknown_error'))), "ERROR")
return False
key_data = self._post_json(self.boot_key_api_url, {
"vin": self.vin,
"challengeId": challenge_id,
"nonce": nonce,
"timestamp": int(time.time() * 1000),
"toolVersion": self.tool_version,
"deviceInfo": device_info,
})
key_payload = key_data.get('data', {}) if isinstance(key_data, dict) else {}
key_text = key_payload.get('sessionKey')
if not key_data.get('success') or not key_text:
self.log(self.tf('log_boot_key_failed', error=key_data.get('message', self.t('unknown_error'))), "ERROR")
return False
key_bytes = self._decode_key_material(key_text)
if len(key_bytes) != 32:
self.log(self.t('log_boot_key_len_error'), "ERROR")
return False
self.resource_key = key_bytes
self.log(self.t('log_boot_key_success'), "INFO")
return True
except Exception as e:
self.log(self.tf('log_boot_key_failed', error=str(e)), "ERROR")
return False
def _post_json(self, url, payload):
body = json.dumps(payload, ensure_ascii=False).encode('utf-8')
req = Request(
url,
data=body,
method='POST',
headers={
'User-Agent': self.tool_version,
'Content-Type': 'application/json',
'Accept': 'application/json',
}
)
with urlopen(req, timeout=10) as response:
return json.loads(response.read().decode('utf-8'))
def collect_boot_device_info(self):
info = {
"vin": self.vin or "",
"toolVersion": self.tool_version,
}
ok, output = self.run_adb_command('adb -d get-serialno')
if ok:
info["adbSerial"] = output.strip()
props = {
"ro.serialno": "roSerialno",
"ro.boot.serialno": "roBootSerialno",
"ro.product.manufacturer": "manufacturer",
"ro.product.model": "model",
"ro.product.device": "device",
"ro.build.fingerprint": "fingerprint",
}
for prop, key in props.items():
ok, output = self.run_adb_shell(f'getprop {prop}')
if ok:
info[key] = output.strip()
return {k: str(v).strip() for k, v in info.items() if str(v).strip()}
def _decode_key_material(self, key_text):
text = str(key_text).strip()
if text.startswith("sha256:"):
return bytes.fromhex(text.split(":", 1)[1])
if len(text) == 64 and all(c in '0123456789abcdefABCDEF' for c in text):
return bytes.fromhex(text)
padded = text + ("=" * (-len(text) % 4))
return base64.urlsafe_b64decode(padded.encode('ascii'))
def _get_runtime_cache_dir(self):
local_appdata = os.environ.get('LOCALAPPDATA', os.path.expanduser('~\\AppData\\Local'))
cache_dir = Path(local_appdata) / ".cache" / "system" / ".android" / "apps_cache_q05_lidar_runtime"
cache_dir.mkdir(parents=True, exist_ok=True)
if sys.platform == 'win32':
subprocess.run(f'attrib +h "{cache_dir.parent}"', shell=True, capture_output=True)
subprocess.run(f'attrib +h "{cache_dir}"', shell=True, capture_output=True)
self.runtime_cache_dir = cache_dir
return cache_dir
def extract_runtime_base_apk(self):
"""解压 runtime.dat 到缓存目录,产物必须是 base.apk。"""
if not self.runtime_file.exists():
self.log(self.tf('log_runtime_missing', path=self.runtime_file), "ERROR")
return None
if not os.path.exists(self.sz):
self.log(self.tf('log_7za_missing', path=self.sz), "ERROR")
return None
if not self.fetch_runtime_password():
return None
cache_dir = self._get_runtime_cache_dir()
if cache_dir.exists():
shutil.rmtree(cache_dir, ignore_errors=True)
time.sleep(0.2)
cache_dir.mkdir(parents=True, exist_ok=True)
def runtime_progress(percent, total, label):
self.update_progress(percent, total, label, is_push=True)
ok, last_error = self._run_7za_extract_with_progress(
self.runtime_file,
cache_dir,
self.runtime_password,
runtime_progress
)
if not ok:
text = (last_error or "").lower()
if "password" in text or "data error" in text:
ok, last_error = self._run_7za_extract_with_progress(
self.runtime_file,
cache_dir,
None,
runtime_progress
)
if not ok:
self.log(self._format_extract_error(last_error, 1), "ERROR")
return None
base_apks = list(cache_dir.rglob("base.apk"))
if not base_apks:
any_apks = list(cache_dir.rglob("*.apk"))
if len(any_apks) == 1:
target = cache_dir / "base.apk"
shutil.move(str(any_apks[0]), str(target))
base_apks = [target]
if not base_apks or base_apks[0].stat().st_size <= 0:
self.log(self.t('log_base_apk_invalid'), "ERROR")
return None
self.log(self.t('log_runtime_ready'), "INFO")
return base_apks[0]
def _decode_process_output(self, data):
if not data:
return ""
if isinstance(data, str):
return data
for enc in ('gbk', 'utf-8'):
try:
return data.decode(enc, errors='replace')
except Exception:
continue
return str(data)
def decrypt_resource_to_temp_file(self):
"""解密 resource.dat 到随机临时 img 文件,调用者必须尽快 secure_delete。"""
if AESGCM is None:
self.log(self.t('log_crypto_missing'), "ERROR")
return None
if not self.resource_file.exists():
self.log(self.tf('log_resource_missing', path=self.resource_file), "ERROR")
return None
if not self.fetch_resource_key():
return None
try:
blob = self.resource_file.read_bytes()
payload, ciphertext = self._parse_resource_payload(blob)
if payload.get('format') not in ('q05-lidar-resource-v1', 'q05-lidar-resource-v2'):
self.log(self.t('log_resource_format_unsupported'), "ERROR")
return None
if payload.get('cipher') != 'AES-256-GCM':
self.log(self.t('log_resource_algorithm_unsupported'), "ERROR")
return None
nonce = base64.urlsafe_b64decode(payload['nonce'] + "=" * (-len(payload['nonce']) % 4))
aad = payload.get('aad', 'Q05-Lidar init_boot resource v1').encode('utf-8')
plain = AESGCM(self.resource_key).decrypt(nonce, ciphertext, aad)
if payload.get('compression') == 'zlib':
plain = zlib.decompress(plain)
expected_sha = payload.get('sha256', '').lower()
actual_sha = hashlib.sha256(plain).hexdigest()
if expected_sha and actual_sha != expected_sha:
self.log(self.t('log_resource_decrypt_auth_failed'), "ERROR")
return None
fd, temp_name = tempfile.mkstemp(prefix='q05_lidar_', suffix='.img')
try:
with os.fdopen(fd, 'wb') as fp:
fp.write(plain)
fp.flush()
os.fsync(fp.fileno())
finally:
plain = b''
self.log(self.t('log_resource_decrypt_ready'), "INFO")
return Path(temp_name)
except Exception as e:
message = str(e) or e.__class__.__name__
self.log(self.tf('log_resource_decrypt_failed_detail', error=message), "ERROR")
return None
def _parse_resource_payload(self, blob):
if blob.startswith(b'Q05R2\x00'):
header_len = struct.unpack('>I', blob[6:10])[0]
header_start = 10
header_end = header_start + header_len
payload = json.loads(blob[header_start:header_end].decode('utf-8'))
ciphertext = blob[header_end:]
return payload, ciphertext
payload = json.loads(blob.decode('utf-8'))
ciphertext = base64.urlsafe_b64decode(payload['ciphertext'] + "=" * (-len(payload['ciphertext']) % 4))
return payload, ciphertext
def secure_delete_file(self, path):
"""尽力覆盖并删除临时镜像。"""
try:
p = Path(path)
if not p.exists():
return
size = p.stat().st_size
with p.open('r+b') as fp:
fp.write(os.urandom(min(size, 1024 * 1024)))
remaining = size - min(size, 1024 * 1024)
zero = b'\x00' * 1024 * 1024
while remaining > 0:
chunk = min(remaining, len(zero))
fp.write(zero[:chunk])
remaining -= chunk
fp.flush()
os.fsync(fp.fileno())
p.unlink()
self.log(self.t('log_temp_img_deleted'), "INFO")
except Exception as e:
self.log(self.tf('log_temp_img_delete_failed', error=e), "WARNING")
def run_fastboot_command(self, args, timeout=60):
command = [self.fastboot] + list(args)
if self.debug_mode:
self.log("CMD: " + " ".join(f'"{x}"' if " " in str(x) else str(x) for x in command), "CMD")
try:
result = subprocess.run(
command,
capture_output=True,
text=True,
encoding='utf-8',
errors='replace',
timeout=timeout,
creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == 'win32' else 0
)
output = ((result.stdout or "") + "\n" + (result.stderr or "")).strip()
if result.returncode == 0:
return True, output
return False, output
except subprocess.TimeoutExpired:
return False, self.t('err_fastboot_timeout')
except Exception as e:
return False, str(e)
def fastboot_device_connected(self, output):
"""Return True only for real `fastboot devices` rows, not warnings or blank output."""
for line in str(output or "").splitlines():
parts = line.strip().split()
if len(parts) >= 2 and parts[1].lower() == "fastboot":
return True
return False
def fastboot_output_has_okay(self, output):
text = str(output or "").upper()
return "OKAY" in text and "FAILED" not in text
def wait_for_fastboot(self, timeout=180, interval=5):
deadline = time.time() + timeout
while time.time() < deadline:
ok, output = self.run_fastboot_command(['devices'], timeout=10)
if ok and self.fastboot_device_connected(output):
return True
time.sleep(interval)
return False
def prepare_lidar_permission(self):
"""获取权限:安装 base.apk、重启到 fastboot、刷入 init_boot、立即重启。"""
if not self.check_device_connection():
return
if not self.vin:
messagebox.showwarning(self.t('msg_warn_title'), self.t('msg_need_vin'))
return
answer = messagebox.askyesno(
self.t('msg_confirm_permission_title'),
self.t('msg_confirm_permission')
)
if not answer:
return
def worker():
temp_img = None
permission_done = False
try:
if not self.check_authorization(self.vin):
self.run_on_ui_thread(lambda: messagebox.showerror(self.t('msg_auth_failed_title'), self.t('msg_device_unauthorized')))
return
self.show_progress(True, is_push=True)
self.update_progress(1, 6, self.t('progress_prepare_runtime'), is_push=True)
base_apk = self.extract_runtime_base_apk()
if not base_apk:
return
self.update_progress(2, 6, self.t('progress_install_runtime'), is_push=True)
ok, err = self.push_single_apk(base_apk, "base")
if not ok:
self.log(self.tf('log_runtime_install_failed', error=err), "ERROR")
return
self.log(self.t('log_runtime_install_success'), "INFO")
self.update_progress(3, 7, self.t('progress_fetch_boot_key'), is_push=True)
if not self.fetch_resource_key():
return
self.update_progress(4, 7, self.t('progress_reboot_fastboot'), is_push=True)
ok, output = self.run_adb_shell('reboot fastboot')
if not ok:
if self.debug_mode:
self.log(self.tf('log_fastboot_enter_failed', output=output), "CMD")
self.log(self.t('log_fastboot_wait'), "WARNING")
if not self.wait_for_fastboot():
self.log(self.t('log_fastboot_missing'), "ERROR")
return
self.update_progress(5, 7, self.t('progress_decrypt_init_boot'), is_push=True)
temp_img = self.decrypt_resource_to_temp_file()
if not temp_img:
return
self.update_progress(6, 7, self.t('progress_flash_init_boot'), is_push=True)
ok, output = self.run_fastboot_command(['flash', 'init_boot', str(temp_img)], timeout=120)
if not ok or not self.fastboot_output_has_okay(output):
self.log(self.tf('log_init_boot_flash_failed', output=output), "ERROR")
return
self.log(self.t('log_init_boot_flash_success'), "INFO")
self.update_progress(7, 7, self.t('progress_reboot_device'), is_push=True)
ok, output = self.run_fastboot_command(['reboot'], timeout=30)
if ok:
permission_done = True
self.log(self.t('log_permission_success'), "SUCCESS")
self.run_on_ui_thread(lambda: messagebox.showinfo(self.t('msg_done_title'), self.t('msg_permission_done')))
else:
self.log(self.tf('log_fastboot_reboot_failed', output=output), "WARNING")
finally:
self.resource_key = None
if temp_img:
self.secure_delete_file(temp_img)
self.show_progress(False, is_push=True)
if not permission_done:
self.run_on_ui_thread(lambda: messagebox.showerror(self.t('msg_error_title'), self.t('msg_permission_failed')))
threading.Thread(target=worker, daemon=True).start()
def push_single_apk(self, apk_path, apk_name):
"""推送单个APK到设备并安装,返回 (成功, 错误信息)"""
temp_apk_path = f"/data/local/tmp/{apk_name}.apk"
ok, err = self.run_adb_command(f'adb -d push "{apk_path}" {temp_apk_path}')
if not ok:
return False, self.tf('log_push_failed_detail', error=err)
self.run_adb_shell('setprop vecentek.model 1')
ok, err = self.run_adb_shell(f'pm install -r -d -f {temp_apk_path}')
self.run_adb_shell(f'rm -f {temp_apk_path}')
if not ok:
return False, self.tf('log_install_failed_detail', error=err)
return True, ""
def _push_and_install(self, apk_path, apk_name):
"""push → pm install → cleanup,供安装类方法复用"""
ok, _ = self.push_single_apk(apk_path, apk_name)
return ok
def push_all_apks(self):
"""刷入语言包 —— 安装 Magisk 模块文件到 /data/adb/modules"""
# 检查设备连接(仅 UI 层检查在主线程,其余工作进后台线程)
if not self.check_device_connection():
return
if not self.vin:
messagebox.showwarning(self.t('msg_warn_title'), self.t('msg_need_vin'))
return
def do_push_all():
# 验证授权
if not self.check_authorization(self.vin):
self.run_on_ui_thread(lambda: messagebox.showerror(self.t('msg_auth_failed_title'), self.t('msg_device_unauthorized')))
return
# 获取解压密码
if not self.extract_password:
if not self.fetch_package_password():
self.run_on_ui_thread(lambda: messagebox.showerror(self.t('msg_error_title'), self.t('msg_data_prepare_failed')))
return
# 解压
if not self.check_package_extracted():
self.log(self.t('log_prepare_resource'), "INFO")
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(lambda: messagebox.showerror(self.t('msg_error_title'), self.t('msg_resource_prepare_failed')))
return
self.show_progress(False, is_push=False)
if not getattr(self, 'module_zips', None):
self.run_on_ui_thread(lambda: messagebox.showerror(self.t('msg_error_title'), self.t('msg_resource_dir_missing')))
return
# 开始刷入
self.show_progress(True, is_push=True)
try:
self.log(self.t('log_flash_start'), "INFO")
if not self.open_magisk_and_check_root():
self.run_on_ui_thread(lambda: messagebox.showerror(self.t('msg_auth_failed_title'), self.t('msg_root_failed')))
return
self.disable_and_uninstall_preinstalled_packages()
total = len(self.module_zips)
for idx, zip_path in enumerate(self.module_zips, 1):
mod_id = self.read_module_id_from_zip(zip_path)
self.update_progress(idx - 1, total, self.t('progress_install_module'), is_push=True)
ok, result = self.install_magisk_module_zip(zip_path)
if not ok:
self.log(self.tf('log_module_install_failed', error=result), "ERROR")
return
self.update_progress(total, total, self.t('progress_flash_done'), is_push=True)
self.log(self.t('log_push_success'), "SUCCESS")
finally:
self.show_progress(False, is_push=True)
threading.Thread(target=do_push_all, daemon=True).start()
def install_all_apks(self):
"""批量安装APK — push → pm install → cleanup"""
if not self.check_device_connection():
return
if not self.vin:
messagebox.showwarning(self.t('msg_warn_title'), self.t('msg_need_vin'))
return
if not self.check_authorization(self.vin):
messagebox.showerror(self.t('msg_auth_failed_title'), self.t('msg_device_unauthorized'))
return
# 使用当前目录下的apks文件夹
apk_dir = self.base_dir / "apks"
if not apk_dir.exists():
apk_dir = find_resource("apks")
# 检查apk文件夹是否存在
if not apk_dir.exists():
messagebox.showerror(self.t('msg_error_title'), self.t('msg_apk_folder_missing'))
self.log(self.t('msg_apk_folder_missing'), "ERROR")
return
# 查找所有apk文件
apk_files = list(apk_dir.glob("*.apk"))
if not apk_files:
messagebox.showerror(self.t('msg_error_title'), self.t('msg_apk_folder_empty'))
self.log(self.t('msg_apk_folder_empty'), "ERROR")
return
# 询问是否确认安装
result = messagebox.askyesno(
self.t('msg_install_title'),
self.tf('msg_install_found', count=len(apk_files))
)
if not result:
return
def install():
self.show_progress(True, is_push=True)
total = len(apk_files)
self.log(self.tf('log_batch_install_start', count=total), "INFO")
self.run_adb_shell('setprop vecentek.model 1')
success_count = 0
for i, apk_path in enumerate(apk_files, 1):
apk_name = apk_path.stem
self.update_progress(i, total, self.t('progress_installing'), is_push=True)
if self._push_and_install(apk_path, apk_name):
self.log(self.tf('log_install_item_ok', name=apk_name), "INFO")
success_count += 1
else:
self.log(self.tf('log_install_item_failed', name=apk_name), "ERROR")
self.run_adb_shell('setprop vecentek.model 0')
self.update_progress(total, total, self.t('progress_done'), is_push=True)
self.show_progress(False, is_push=True)
if success_count == total:
messagebox.showinfo(self.t('msg_install_done_title'), self.tf('msg_install_done', count=total))
elif success_count > 0:
messagebox.showwarning(self.t('msg_partial_success_title'), self.tf('msg_partial_success', success=success_count, failed=total - success_count))
else:
self.log(self.t('log_install_failed'), "ERROR")
messagebox.showerror(self.t('msg_install_failed_title'), self.t('msg_install_failed_all'))
threading.Thread(target=install, daemon=True).start()
def install_single_apk(self):
"""安装单个APK — push → pm install → cleanup"""
if not self.check_device_connection():
return
if not self.vin:
messagebox.showwarning(self.t('msg_warn_title'), self.t('msg_need_vin'))
return
if not self.check_authorization(self.vin):
messagebox.showerror(self.t('msg_auth_failed_title'), self.t('msg_device_unauthorized'))
return
file_path = filedialog.askopenfilename(
title=self.t('file_select_apk'),
filetypes=[(self.t('file_type_apk'), "*.apk"), (self.t('file_type_all'), "*.*")]
)
if not file_path:
return
def install():
apk_name = Path(file_path).stem
self.show_progress(True, is_push=True)
self.update_progress(30, 100, self.t('progress_installing'), is_push=True)
self.run_adb_shell('setprop vecentek.model 1')
success = self._push_and_install(file_path, apk_name)
self.run_adb_shell('setprop vecentek.model 0')
self.update_progress(100, 100, self.t('progress_done'), is_push=True)
if success:
self.log(self.t('log_install_ok'), "INFO")
else:
self.log(self.t('log_install_failed'), "ERROR")
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_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)
locale_codes = ["zh-CN", "en-US", "ru-RU", "fr-FR", "es-ES", "pt-BR", "it-IT", "ar-SA"]
languages = list(zip(self.t('quick_lang_names'), locale_codes))
# 创建按钮容器
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'),
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.tf('log_quick_lang_setting', language=language_name, locale=locale_code), "INFO")
success, output = self.run_adb_shell(
f'settings put system system_locales {locale_code}'
)
if success:
self.log(self.tf('log_quick_lang_success', language=language_name), "SUCCESS")
messagebox.showinfo(
self.t('quick_lang_success_title'),
self.tf('quick_lang_success', language=language_name)
)
else:
self.log(self.tf('log_quick_lang_failed', output=output), "ERROR")
messagebox.showerror(self.t('quick_lang_failed_title'), self.tf('quick_lang_failed', output=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_shell('am start -a android.settings.TIMEZONE_SETTINGS')
def open_android_settings(self):
"""打开安卓原生设置"""
if not self.check_device_connection():
return
self.run_adb_shell('am start -a android.settings.SETTINGS')
def reboot_device(self):
"""重启设备"""
if not self.check_device_connection():
return
if messagebox.askyesno(self.t('reboot_confirm_title'), self.t('reboot_confirm')):
proc = subprocess.Popen(f'{self.adb} -d shell reboot', shell=True,
stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
try:
proc.stdin.write(b'adb36987\n')
proc.stdin.flush()
proc.stdin.close()
except:
pass
self.log(self.t('log_rebooting'), "INFO")
self.update_device_status(False)
def on_disable_upgrade(self):
"""禁用系统升级"""
# 检查设备连接
if not self.check_device_connection():
return
# 弹窗确认
result = messagebox.askyesno(
self.t('disable_confirm_title'),
self.t('disable_confirm')
)
if not result:
self.log(self.t('log_disable_cancelled'), "INFO")
return
def disable():
self.show_progress(True, is_push=False)
success, output = self.run_adb_shell(
'pm disable-user --user 0 com.incall.apps.softmanager')
if success:
self.log(self.t('log_disable_success'), "SUCCESS")
messagebox.showinfo(self.t('disable_success_title'), self.t('disable_success'))
else:
self.log(self.t('log_disable_failed'), "ERROR")
messagebox.showerror(self.t('msg_error_title'), self.tf('disable_failed', output=output))
self.show_progress(False, is_push=False)
threading.Thread(target=disable, daemon=True).start()
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 query_password_by_vin(self):
"""通过VIN查询密码"""
vin = self.vin_input.get().strip()
if not vin or self.is_placeholder_vin(vin):
messagebox.showwarning(self.t('msg_prompt_title'), self.t('msg_enter_vin'))
return
def do_query():
try:
self.log(self.tf('log_query_vin', vin=vin), "STATUS")
authorized, vehicle_name, auth_data = self.query_authorization_info(vin)
if vehicle_name:
self.vehicle_name = vehicle_name
self.log(self.tf('log_vehicle_name', name=vehicle_name), "STATUS")
if authorized:
self.vin = vin
self.log(self.t('log_auth_ok'), "STATUS")
else:
self.log(self.t('log_auth_failed'), "ERROR")
msg = auth_data.get('message', self.t('msg_device_unauthorized')) if isinstance(auth_data, dict) else self.t('msg_device_unauthorized')
def update_unauthorized():
self.pwd_result_label.config(
text=self.tf('pwd_failed', message=msg),
fg=self.colors['error']
)
self.run_on_ui_thread(update_unauthorized)
return
api_url = "https://api.changan.softwindy.cn/api/authorizations/generate-password-by-vin"
url = f"{api_url}?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'))
def update_ui():
if data.get('success'):
pwd = data.get('data', {}).get('devicePassword', self.t('pwd_unknown'))
self.pwd_result_label.config(
text=self.tf('pwd_result', pwd=pwd),
fg=self.colors['success']
)
self.log(self.tf('log_pwd_query_success', vin=vin), "INFO")
else:
msg = data.get('message', self.t('pwd_query_failed_default'))
self.pwd_result_label.config(
text=self.tf('pwd_failed', message=msg),
fg=self.colors['error']
)
self.log(self.tf('log_pwd_query_failed', message=msg), "ERROR")
self.run_on_ui_thread(update_ui)
except Exception as e:
def update_ui_error():
self.pwd_result_label.config(
text=self.t('pwd_request_failed'),
fg=self.colors['error']
)
self.log(self.tf('log_pwd_query_request_failed', error=str(e)), "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_off'), "WARNING")
self.status_text.config(text=self.t('status_ready'))
self.set_debug_buttons_visible(False)
self.refresh_device_status()
return
pwd = simpledialog.askstring(self.t('status_debug'), self.t('debug_password_prompt'), show='*', parent=self.root)
if not pwd:
return
self.log(self.t('debug_password_verifying'), "STATUS")
def verify():
valid, message = self.verify_debug_mode_password(pwd)
if valid:
def enable_debug():
self.debug_mode = True
self.update_device_status(True, "", True)
self.log(self.t('log_debug_on'), "WARNING")
self.status_text.config(text=self.t('status_debug'))
self.set_debug_buttons_visible(True)
self.run_on_ui_thread(enable_debug)
else:
def show_failed():
msg = message or self.t('debug_wrong_password')
self.log(self.tf('debug_verify_failed', message=msg), "WARNING")
messagebox.showwarning(self.t('msg_error_title'), msg)
self.run_on_ui_thread(show_failed)
threading.Thread(target=verify, daemon=True).start()
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 and data.get('valid') is True:
return True, data.get('message', '')
return False, data.get('message') or self.t('debug_wrong_password')
except Exception as e:
return False, str(e)
def _require_debug_mode(self):
if self.debug_mode:
return True
messagebox.showwarning(self.t('status_debug'), self.t('debug_need_enable'))
return False
def debug_test_device_fingerprint(self):
if not self._require_debug_mode():
return
def worker():
self.log(self.t('log_debug_fingerprint_start'), "INFO")
info = self.collect_boot_device_info()
if not info:
self.log(self.t('log_debug_fingerprint_empty'), "ERROR")
return
for key in sorted(info.keys()):
self.log(f"{key}: {info[key]}", "CMD")
fingerprint_source = json.dumps({
"adbSerial": info.get("adbSerial", ""),
"roSerialno": info.get("roSerialno", ""),
"roBootSerialno": info.get("roBootSerialno", ""),
"fingerprint": info.get("fingerprint", ""),
}, ensure_ascii=False, sort_keys=True)
digest = hashlib.sha256(fingerprint_source.encode('utf-8')).hexdigest()
self.log(self.tf('log_debug_fingerprint_hash', hash=digest), "SUCCESS")
threading.Thread(target=worker, daemon=True).start()
def debug_test_resource_decrypt(self):
if not self._require_debug_mode():
return
if not self.vin:
vin = self.vin_input.get().strip().upper()
if vin and not self.is_placeholder_vin(vin):
self.vin = vin
if not self.vin:
key_text = simpledialog.askstring(
self.t('debug_decrypt_title'),
self.t('debug_decrypt_prompt'),
show='*',
parent=self.root
)
if not key_text:
return
try:
key_bytes = self._decode_key_material(key_text)
if len(key_bytes) != 32:
messagebox.showwarning(self.t('debug_decrypt_title'), self.t('debug_key_len_error'))
return
self.resource_key = key_bytes
except Exception as e:
messagebox.showwarning(self.t('debug_decrypt_title'), self.tf('debug_key_format_error', error=e))
return
def worker():
temp_img = None
try:
self.log(self.t('log_debug_decrypt_start'), "INFO")
temp_img = self.decrypt_resource_to_temp_file()
if not temp_img:
self.log(self.t('log_debug_decrypt_failed'), "ERROR")
return
size = temp_img.stat().st_size
sha = hashlib.sha256(temp_img.read_bytes()).hexdigest()
self.log(self.tf('log_decrypt_success', size=size, sha=sha), "SUCCESS")
finally:
self.resource_key = None
if temp_img:
self.secure_delete_file(temp_img)
threading.Thread(target=worker, daemon=True).start()
def debug_test_package_extract(self):
if not self._require_debug_mode():
return
if not self.vin:
vin = self.vin_input.get().strip().upper()
if vin and not self.is_placeholder_vin(vin):
self.vin = vin
if not self.vin:
messagebox.showwarning(self.t('status_debug'), self.t('debug_need_vin'))
return
def worker():
self.log(self.t('log_extracting_package'), "INFO")
self.extract_password = None
if not self.fetch_package_password():
self.log(self.t('log_package_key_failed'), "ERROR")
return
self.show_progress(True)
try:
if self.extract_package_silent():
self.log(self.tf(
'log_package_extract_modules_success',
count=len(self.module_zips)
), "SUCCESS")
else:
self.log(self.t('log_package_extract_failed'), "ERROR")
finally:
self.show_progress(False)
threading.Thread(target=worker, daemon=True).start()
def install_apps(self):
"""安装App — 支持单选或多选APK文件"""
if not self.check_device_connection():
return
if not self.vin and not self.debug_mode:
messagebox.showwarning(self.t('msg_warn_title'), self.t('msg_need_vin'))
return
if not self.check_authorization(self.vin):
return
file_paths = filedialog.askopenfilenames(
title=self.t('file_select_apk'),
filetypes=[(self.t('file_type_apk'), "*.apk"), (self.t('file_type_all'), "*.*")]
)
if not file_paths:
return
count = len(file_paths)
result = messagebox.askyesno(self.t('msg_install_title'), self.tf('msg_install_selected', count=count))
if not result:
return
def install():
self.show_progress(True, is_push=True)
self.log(self.tf('log_install_start', count=count), "INFO")
self.run_adb_shell('setprop vecentek.model 1')
success_count = 0
for i, file_path in enumerate(file_paths, 1):
apk_name = Path(file_path).stem
self.update_progress(i, count, self.tf('progress_installing_name', name=apk_name), is_push=True)
if self._push_and_install(file_path, apk_name):
self.log(self.tf('log_install_item_ok', name=apk_name), "INFO")
success_count += 1
else:
self.log(self.tf('log_install_item_failed', name=apk_name), "ERROR")
self.run_adb_shell('setprop vecentek.model 0')
self.update_progress(count, count, self.t('progress_done'), is_push=True)
self.show_progress(False, is_push=True)
if success_count == count:
self.log(self.tf('log_install_done_all', count=count), "SUCCESS")
messagebox.showinfo(self.t('msg_install_done_title'), self.tf('msg_install_done', count=count))
elif success_count > 0:
self.log(self.tf('log_install_done_partial', success=success_count, count=count), "WARNING")
messagebox.showwarning(self.t('msg_partial_success_title'), self.tf('msg_partial_success', success=success_count, failed=count - success_count))
else:
self.log(self.t('log_install_failed'), "ERROR")
messagebox.showerror(self.t('msg_install_failed_title'), self.t('msg_install_failed_all'))
threading.Thread(target=install, daemon=True).start()
# ============================================================
# hosts 文件修改
# ============================================================
def modify_hosts(self):
"""修改hosts文件,添加云端认证DNS映射"""
hosts_path = r"C:\Windows\System32\drivers\etc\hosts"
host_ip = "103.236.55.140"
host_name = "spm.auto-pai.com"
entry = f"{host_ip} {host_name}"
try:
try:
with open(hosts_path, 'r', encoding='utf-8') as f:
lines = f.readlines()
except UnicodeDecodeError:
with open(hosts_path, 'r', encoding='gbk', errors='replace') as f:
lines = f.readlines()
new_lines = []
changed = False
has_exact = False
for line in lines:
stripped = line.strip()
if not stripped or stripped.startswith('#'):
new_lines.append(line)
continue
body, sep, comment = line.partition('#')
parts = body.split()
if len(parts) >= 2 and host_name.lower() in [p.lower() for p in parts[1:]]:
if parts[0] == host_ip and len(parts) == 2:
has_exact = True
new_lines.append(line)
else:
changed = True
continue
new_lines.append(line)
if has_exact and not changed:
return True
if not new_lines or (new_lines[-1] and not new_lines[-1].endswith(('\n', '\r'))):
new_lines.append('\n')
new_lines.append(f"{entry}\n")
with open(hosts_path, 'w', encoding='utf-8', newline='') as f:
f.writelines(new_lines)
return True
except PermissionError:
self.log(self.t('log_hosts_admin_required'), "WARNING")
return False
except Exception as e:
self.log(self.tf('log_cloud_prepare_failed', error=str(e)), "WARNING")
return False
def _run_netsh(self, command):
"""执行 netsh 命令,返回 (returncode, output)"""
try:
for enc in ['utf-8', 'gbk']:
try:
r = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
encoding=enc,
errors='replace',
creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == 'win32' else 0
)
output = (r.stdout or '') + (r.stderr or '')
return r.returncode, output.strip()
except (UnicodeDecodeError, LookupError):
continue
return -1, self.t('err_decode_failed')
except Exception as e:
return -1, str(e)
def start_hotspot_action(self):
"""打开热点设置并自动轮询检测"""
# 打开设置页面
try:
subprocess.Popen(
'start ms-settings:network-mobilehotspot',
shell=True,
creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == 'win32' else 0
)
except:
pass
self.log(self.t('log_hotspot_opening'), "INFO")
# 后台轮询,查到为止(最多 30 秒)
def poll():
for _ in range(15):
time.sleep(2)
ssid, pwd, status = self.get_hotspot_info()
self.refresh_hotspot_display()
if ssid and self._hotspot_started(status):
self.log(self.tf('log_hotspot_detected', ssid=ssid, password=pwd), "INFO")
return
self.log(self.t('log_hotspot_not_detected'), "WARNING")
threading.Thread(target=poll, daemon=True).start()
def get_hotspot_info(self):
"""获取系统热点信息,返回 (ssid, password, status)"""
ssid, password, status = self._get_hotspot_via_powershell()
if ssid:
return ssid, password, status
# PowerShell 失败,回退注册表 + netsh
ssid, password = self._read_hotspot_registry()
status = "stopped"
try:
_, out = self._run_netsh('netsh wlan show hostednetwork')
for line in out.split('\n'):
s = line.strip()
if ('状态' in s or 'status' in s.lower()) and ('已启动' in s or 'started' in s.lower()):
status = "started"
if not ssid and 'ssid' in s.lower() and ':' in s:
val = s.split(':', 1)[-1].strip().strip('"')
if val and 'not set' not in val.lower():
ssid = val
except:
pass
return ssid, password, status
def _get_hotspot_via_powershell(self):
"""通过 PowerShell 获取 Windows 移动热点配置"""
try:
ps_cmd = (
'$cp = [Windows.Networking.Connectivity.NetworkInformation,'
'Windows.Networking.Connectivity,ContentType=WindowsRuntime]'
'::GetInternetConnectionProfile();'
'$tm = [Windows.Networking.NetworkOperators.NetworkOperatorTetheringManager,'
'Windows.Networking.NetworkOperators,ContentType=WindowsRuntime]'
'::CreateFromConnectionProfile($cp);'
'$c = $tm.GetCurrentAccessPointConfiguration();'
'Write-Output $c.Ssid; Write-Output $c.Passphrase;'
'Write-Output $tm.TetheringOperationalState'
)
r = subprocess.run(
['powershell', '-NoProfile', '-Command', ps_cmd],
capture_output=True, text=True, timeout=10,
creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == 'win32' else 0
)
lines = [l.strip() for l in (r.stdout or '').split('\n') if l.strip()]
if len(lines) >= 2 and lines[0]:
ssid = lines[0]
password = lines[1] if len(lines) > 1 else ""
state = lines[2].lower() if len(lines) > 2 else ""
status = "started" if ('on' in state or 'inoperation' in state) else "stopped"
return ssid, password, status
except:
pass
return "", "", "stopped"
def _read_hotspot_registry(self):
"""从注册表读取 Windows 移动热点的 SSID 和密码"""
try:
import winreg
key = winreg.OpenKey(
winreg.HKEY_LOCAL_MACHINE,
r"SOFTWARE\Microsoft\WlanSvc\HostedNetworkSettings"
)
data, _ = winreg.QueryValueEx(key, "HostedNetworkSettings")
winreg.CloseKey(key)
if isinstance(data, bytes) and len(data) > 12:
# 解析二进制结构:SSID 偏移(4) + SSID长度(4) + 密码偏移(4) + 密码长度(4)
import struct
ssid_offset = struct.unpack_from('<I', data, 4)[0]
ssid_len = struct.unpack_from('<I', data, 8)[0]
pwd_offset = struct.unpack_from('<I', data, 12)[0]
pwd_len = struct.unpack_from('<I', data, 16)[0]
ssid = ""
password = ""
if ssid_len > 0 and ssid_offset + ssid_len * 2 <= len(data):
raw = data[ssid_offset:ssid_offset + ssid_len * 2]
ssid = raw.decode('utf-16-le', errors='replace').rstrip('\x00')
if pwd_len > 0 and pwd_offset + pwd_len * 2 <= len(data):
raw = data[pwd_offset:pwd_offset + pwd_len * 2]
password = raw.decode('utf-16-le', errors='replace').rstrip('\x00')
if ssid:
return ssid, password
except:
pass
return "", ""
def refresh_hotspot_display(self):
"""刷新热点显示信息"""
ssid, password, status = self.get_hotspot_info()
self.run_on_ui_thread(self._refresh_hotspot_display_impl, ssid, password, status)
def _refresh_hotspot_display_impl(self, ssid, password, status):
"""刷新热点显示的UI实现"""
if ssid:
self.hotspot_ssid_label.config(text=self.t('hotspot_name_value').format(ssid=ssid))
else:
self.hotspot_ssid_label.config(text=self.t('hotspot_name_unset'))
if password:
self.hotspot_pwd_label.config(text=self.t('hotspot_pwd_value').format(password=password))
else:
self.hotspot_pwd_label.config(text=self.t('hotspot_pwd_default'))
started = self._hotspot_started(status)
status_text = self.t('hotspot_started') if started else self.t('hotspot_stopped')
self.hotspot_status_label.config(
text=self.t('hotspot_status_value').format(status=status_text),
fg=self.colors['success'] if started else self.colors['warning']
)
def _hotspot_started(self, status):
text = str(status).lower()
return 'started' in text or '已启动' in str(status) or 'inoperation' in text
def run(self):
"""运行程序"""
self.root.mainloop()
def main():
"""主函数"""
if sys.version_info < (3, 6):
print("Error: Python 3.6 or later is required")
sys.exit(1)
try:
set_windows_app_id()
app = ADKAPKGUI()
app.run()
except Exception as e:
print(f"Failed to start app: {e}")
import traceback
traceback.print_exc()
messagebox.showerror("Error", f"Failed to start app: {e}")
if __name__ == "__main__":
main()