#!/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("", self._on_vin_input_focus_in) self.vin_input.bind("", 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('', 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('', on_enter) btn.bind('', 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'(?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(' 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()