alceawisteria
698 lines
28 KiB
Python
698 lines
28 KiB
Python
import sys
|
|
import subprocess
|
|
import ctypes
|
|
import os
|
|
import threading
|
|
import time
|
|
from datetime import datetime
|
|
from PyQt6.QtWidgets import (
|
|
QApplication, QWidget, QVBoxLayout, QPushButton, QComboBox,
|
|
QLabel, QLineEdit, QFileDialog, QProgressBar, QTextEdit,
|
|
QHBoxLayout, QMessageBox
|
|
)
|
|
from PyQt6.QtCore import Qt, pyqtSignal, QObject
|
|
from PyQt6.QtGui import QFont
|
|
|
|
class CopySignals(QObject):
|
|
"""Signals for copy thread"""
|
|
progress = pyqtSignal(int, int) # current, total
|
|
status = pyqtSignal(str)
|
|
finished = pyqtSignal(bool, str, list) # success, message, failed_files
|
|
file_progress = pyqtSignal(str, int, int) # filename, current, total
|
|
|
|
def get_adb_mounts():
|
|
"""Return a list of device mounts with SD card and internal storage at top."""
|
|
try:
|
|
# Check if any devices are connected
|
|
devices_result = subprocess.run(
|
|
["adb", "devices"],
|
|
capture_output=True, text=True, encoding='utf-8', errors='ignore', check=True
|
|
)
|
|
lines = devices_result.stdout.strip().splitlines()[1:] # skip header
|
|
connected = [line for line in lines if "device" in line]
|
|
if not connected:
|
|
return ["No device connected or unauthorized"]
|
|
|
|
# Get mounts from the first connected device
|
|
mount_result = subprocess.run(
|
|
["adb", "shell", "mount"],
|
|
capture_output=True, text=True, encoding='utf-8', errors='ignore', check=True
|
|
)
|
|
|
|
mounts = []
|
|
sd_card_mount = None
|
|
internal_storage = None
|
|
other_mounts = []
|
|
|
|
for line in mount_result.stdout.splitlines():
|
|
parts = line.split()
|
|
if len(parts) >= 3:
|
|
device, mount_point = parts[0], parts[2]
|
|
mount_entry = f"{device} -> {mount_point}"
|
|
|
|
# Check for SD card patterns (xxxx-xxxx format in device or common SD mount points)
|
|
if any(pattern in mount_point.lower() for pattern in ['/mnt/media_rw/', '/storage/', '/sdcard', '/external_sd']) \
|
|
or '-' in device and len(device.split('-')[0]) == 4:
|
|
sd_card_mount = mount_entry
|
|
# Check for internal storage
|
|
elif '/storage/emulated/0' in mount_point or '/data/media' in mount_point:
|
|
internal_storage = mount_entry
|
|
else:
|
|
other_mounts.append(mount_entry)
|
|
|
|
# Add SD card at top if found (try to get specific format)
|
|
if sd_card_mount:
|
|
# Try to extract the xxxx-xxxx pattern for better display
|
|
device_part = sd_card_mount.split('->')[0].strip()
|
|
mount_part = sd_card_mount.split('->')[1].strip()
|
|
if '-' in device_part and len(device_part) >= 9:
|
|
mounts.append(f"{device_part} -> {mount_part} (micro SD card)")
|
|
else:
|
|
mounts.append(f"{sd_card_mount} (micro SD card)")
|
|
else:
|
|
# Try alternative method to find SD card
|
|
try:
|
|
sd_result = subprocess.run(
|
|
["adb", "shell", "ls", "/storage/"],
|
|
capture_output=True, text=True, encoding='utf-8', errors='ignore', check=True
|
|
)
|
|
for item in sd_result.stdout.split():
|
|
if item and item not in ['emulated', 'self']:
|
|
mounts.append(f"SD Card -> /storage/{item}")
|
|
break
|
|
except:
|
|
pass
|
|
|
|
# Add internal storage second
|
|
if internal_storage:
|
|
mounts.append(f"{internal_storage} (Internal Storage)")
|
|
else:
|
|
# Try to find internal storage via other means
|
|
mounts.append("/data/media/0 -> /storage/emulated/0 (Internal Storage)")
|
|
|
|
# Add all other mounts
|
|
mounts.extend(other_mounts)
|
|
|
|
if not mounts:
|
|
mounts.append("No mounts found")
|
|
return mounts
|
|
except subprocess.CalledProcessError as e:
|
|
return [f"ADB error: {e.stderr.strip() or e}"]
|
|
except FileNotFoundError:
|
|
return ["ADB not found. Make sure it is installed and in PATH."]
|
|
|
|
def get_mount_size(mount_path):
|
|
"""Get total size of a mount point in bytes."""
|
|
try:
|
|
# Get the mount point path from the combo box entry
|
|
if '->' in mount_path:
|
|
mount_path = mount_path.split('->')[1].strip().split()[0] # Get just the path
|
|
|
|
# Get total size using 'df' command
|
|
result = subprocess.run(
|
|
["adb", "shell", "df", mount_path],
|
|
capture_output=True, text=True, encoding='utf-8', errors='ignore', check=True
|
|
)
|
|
|
|
# Parse the df output
|
|
lines = result.stdout.strip().splitlines()
|
|
if len(lines) >= 2:
|
|
# Get the total size (in 1K blocks, typically second column)
|
|
parts = lines[1].split()
|
|
if len(parts) >= 2:
|
|
# Convert from 1K blocks to bytes
|
|
total_blocks = int(parts[1]) # Total 1K blocks
|
|
return total_blocks * 1024
|
|
|
|
return 0
|
|
except Exception as e:
|
|
print(f"Error getting mount size: {e}")
|
|
return 0
|
|
|
|
def get_volume_id(drive_letter):
|
|
"""Return the volume serial number for a given drive letter."""
|
|
volume_name_buf = ctypes.create_unicode_buffer(1024)
|
|
file_system_name_buf = ctypes.create_unicode_buffer(1024)
|
|
serial_number = ctypes.c_uint()
|
|
max_component_length = ctypes.c_uint()
|
|
file_system_flags = ctypes.c_uint()
|
|
ret = ctypes.windll.kernel32.GetVolumeInformationW(
|
|
ctypes.c_wchar_p(drive_letter + "\\"),
|
|
volume_name_buf,
|
|
ctypes.sizeof(volume_name_buf),
|
|
ctypes.byref(serial_number),
|
|
ctypes.byref(max_component_length),
|
|
ctypes.byref(file_system_flags),
|
|
file_system_name_buf,
|
|
ctypes.sizeof(file_system_name_buf)
|
|
)
|
|
if ret:
|
|
return f"{serial_number.value:X}"
|
|
return "Unavailable"
|
|
|
|
def monitor_file_progress(dest_file, process, signals, source_filename, max_wait_time=300):
|
|
"""Monitor file transfer progress by watching file growth."""
|
|
start_time = time.time()
|
|
last_size = -1
|
|
no_progress_start = None
|
|
|
|
# Wait for file to start appearing
|
|
for _ in range(30): # Wait up to 3 seconds for file to appear
|
|
if os.path.exists(dest_file):
|
|
break
|
|
if process.poll() is not None: # Process finished
|
|
return False
|
|
time.sleep(0.1)
|
|
|
|
# Monitor file growth
|
|
while process.poll() is None: # While process is still running
|
|
current_time = time.time()
|
|
|
|
# Check if we've been waiting too long
|
|
if current_time - start_time > max_wait_time:
|
|
signals.status.emit(f"Stuck on file: {source_filename} (no progress for {max_wait_time}s)")
|
|
return False
|
|
|
|
if os.path.exists(dest_file):
|
|
current_size = os.path.getsize(dest_file)
|
|
|
|
if last_size != -1:
|
|
if current_size > last_size:
|
|
# File is growing - reset no-progress timer
|
|
no_progress_start = None
|
|
last_size = current_size
|
|
|
|
# Update progress occasionally
|
|
if int(time.time()) % 5 == 0: # Every 5 seconds
|
|
signals.status.emit(f"Copying {source_filename}: {current_size/(1024*1024):.1f} MB")
|
|
else:
|
|
# File size not increasing
|
|
if no_progress_start is None:
|
|
no_progress_start = current_time
|
|
elif current_time - no_progress_start > 30: # No progress for 30 seconds
|
|
signals.status.emit(f"File stuck: {source_filename} (no progress for 30s)")
|
|
return False
|
|
else:
|
|
last_size = current_size
|
|
else:
|
|
# File doesn't exist yet
|
|
if current_time - start_time > 10: # No file after 10 seconds
|
|
signals.status.emit(f"No file created: {source_filename}")
|
|
return False
|
|
|
|
time.sleep(1) # Check every second
|
|
|
|
return True # Process completed normally
|
|
|
|
def copy_mount_to_drive(mount_path, drive_path, signals):
|
|
"""Copy entire mount to drive root, collecting failed files."""
|
|
failed_files = [] # List to store failed file info
|
|
|
|
try:
|
|
# Extract just the mount path from the combo box entry
|
|
if '->' in mount_path:
|
|
mount_path = mount_path.split('->')[1].strip().split()[0] # Get just the path
|
|
|
|
# Clean drive path - ensure it's just the root (e.g., "F:\")
|
|
drive_path = drive_path.rstrip('/').rstrip('\\')
|
|
if not os.path.isdir(drive_path):
|
|
drive_path = drive_path + "\\"
|
|
|
|
if not drive_path.endswith('\\'):
|
|
drive_path = drive_path + '\\'
|
|
|
|
signals.status.emit(f"Copying to drive root: {drive_path}")
|
|
signals.status.emit(f"Source mount: {mount_path}")
|
|
|
|
# First, get total size for progress calculation
|
|
total_size = get_mount_size(mount_path)
|
|
if total_size == 0:
|
|
signals.status.emit("Warning: Could not determine total size.")
|
|
|
|
signals.progress.emit(0, total_size)
|
|
|
|
# Get list of all directories and create them first
|
|
signals.status.emit("Scanning directory structure...")
|
|
|
|
# Get all directories
|
|
dirs_result = subprocess.run(
|
|
["adb", "shell", "find", mount_path, "-type", "d"],
|
|
capture_output=True, text=True, encoding='utf-8', errors='ignore', check=True
|
|
)
|
|
|
|
dirs = []
|
|
if dirs_result.stdout.strip():
|
|
dirs = dirs_result.stdout.strip().splitlines()
|
|
signals.status.emit(f"Found {len(dirs)} directories. Creating them...")
|
|
|
|
# Create all directories on Windows first
|
|
for i, dir_path in enumerate(dirs):
|
|
if dir_path:
|
|
# Convert to Windows path
|
|
rel_path = os.path.relpath(dir_path, mount_path)
|
|
windows_dir = os.path.join(drive_path, rel_path.replace('/', '\\'))
|
|
|
|
if not os.path.exists(windows_dir):
|
|
try:
|
|
os.makedirs(windows_dir, exist_ok=True)
|
|
except Exception:
|
|
pass
|
|
|
|
# Update progress occasionally
|
|
if i % 500 == 0:
|
|
progress_percent = int((i / len(dirs)) * 15)
|
|
signals.progress.emit(progress_percent, 100)
|
|
|
|
signals.progress.emit(15, 100)
|
|
|
|
# Now get list of all files
|
|
signals.status.emit("Getting file list...")
|
|
files_result = subprocess.run(
|
|
["adb", "shell", "find", mount_path, "-type", "f"],
|
|
capture_output=True, text=True, encoding='utf-8', errors='ignore', check=True
|
|
)
|
|
|
|
files = []
|
|
if files_result.stdout.strip():
|
|
files = files_result.stdout.strip().splitlines()
|
|
signals.status.emit(f"Found {len(files)} files to copy")
|
|
|
|
# Copy files one by one with smart monitoring
|
|
copied_files = 0
|
|
skipped_files = 0
|
|
total_files = len(files)
|
|
|
|
start_time = datetime.now()
|
|
batch_size = max(1, total_files // 100) # Update progress every 1%
|
|
|
|
for i, source_file in enumerate(files):
|
|
if not source_file:
|
|
continue
|
|
|
|
# Calculate Windows destination path
|
|
rel_path = os.path.relpath(source_file, mount_path)
|
|
dest_file = os.path.join(drive_path, rel_path.replace('/', '\\'))
|
|
|
|
# Update progress
|
|
copied_files += 1
|
|
|
|
if i % batch_size == 0:
|
|
progress_percent = 15 + int((copied_files / total_files) * 80)
|
|
signals.progress.emit(progress_percent, 100)
|
|
|
|
# Update status with ETA
|
|
if copied_files > 10:
|
|
elapsed = (datetime.now() - start_time).total_seconds()
|
|
files_per_second = copied_files / elapsed
|
|
remaining_files = total_files - copied_files
|
|
if files_per_second > 0:
|
|
remaining_seconds = remaining_files / files_per_second
|
|
if remaining_seconds < 60:
|
|
eta = f"{int(remaining_seconds)}s"
|
|
elif remaining_seconds < 3600:
|
|
eta = f"{int(remaining_seconds/60)}m"
|
|
else:
|
|
eta = f"{int(remaining_seconds/3600)}h"
|
|
|
|
signals.status.emit(f"Progress: {copied_files}/{total_files} - ETA: {eta}")
|
|
|
|
# Show current file
|
|
if i % 10 == 0:
|
|
filename = os.path.basename(source_file)
|
|
if len(filename) > 30:
|
|
filename = filename[:27] + "..."
|
|
signals.file_progress.emit(filename, copied_files, total_files)
|
|
|
|
# Try to copy the file WITHOUT timeout
|
|
try:
|
|
# Start adb pull process
|
|
process = subprocess.Popen(
|
|
["adb", "pull", source_file, dest_file],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
universal_newlines=False
|
|
)
|
|
|
|
# Monitor the file transfer
|
|
filename_display = os.path.basename(source_file)
|
|
if len(filename_display) > 30:
|
|
filename_display = filename_display[:27] + "..."
|
|
|
|
# Monitor for actual progress (file growth)
|
|
transfer_ok = monitor_file_progress(dest_file, process, signals, filename_display)
|
|
|
|
# Wait for process to complete
|
|
process.wait()
|
|
|
|
if process.returncode != 0 or not transfer_ok:
|
|
# Get error message
|
|
error_msg = ""
|
|
try:
|
|
stderr = process.stderr.read().decode('utf-8', errors='ignore')
|
|
error_msg = stderr.strip()
|
|
except:
|
|
pass
|
|
|
|
if not error_msg and not transfer_ok:
|
|
error_msg = "Transfer stalled or no progress"
|
|
|
|
# Add to failed files
|
|
failed_files.append({
|
|
'source': source_file,
|
|
'destination': dest_file,
|
|
'error': error_msg[:100],
|
|
'adb_command': f'adb pull "{source_file}" "{dest_file}"'
|
|
})
|
|
|
|
skipped_files += 1
|
|
signals.status.emit(f"Failed: {os.path.basename(source_file)[:30]}...")
|
|
|
|
except Exception as e:
|
|
# Other error
|
|
error_msg = str(e)
|
|
failed_files.append({
|
|
'source': source_file,
|
|
'destination': dest_file,
|
|
'error': error_msg[:100],
|
|
'adb_command': f'adb pull "{source_file}" "{dest_file}"'
|
|
})
|
|
|
|
skipped_files += 1
|
|
if i % 100 == 0:
|
|
signals.status.emit(f"Error: {os.path.basename(source_file)[:30]}...")
|
|
|
|
# Final progress
|
|
signals.progress.emit(100, 100)
|
|
|
|
# Completion message
|
|
success_count = total_files - skipped_files
|
|
elapsed_total = (datetime.now() - start_time).total_seconds()
|
|
|
|
if elapsed_total > 0:
|
|
speed = success_count / elapsed_total
|
|
time_str = ""
|
|
if elapsed_total < 60:
|
|
time_str = f"{elapsed_total:.0f} seconds"
|
|
elif elapsed_total < 3600:
|
|
time_str = f"{elapsed_total/60:.1f} minutes"
|
|
else:
|
|
time_str = f"{elapsed_total/3600:.1f} hours"
|
|
|
|
signals.status.emit(f"Copy completed in {time_str} ({speed:.1f} files/sec)")
|
|
|
|
signals.status.emit(f"Success: {success_count}, Failed: {skipped_files}, Total: {total_files}")
|
|
|
|
if failed_files:
|
|
signals.status.emit(f"{len(failed_files)} files failed to copy. You can export ADB commands to retry them.")
|
|
|
|
signals.finished.emit(True, f"Copied {success_count}/{total_files} files to {drive_path}", failed_files)
|
|
|
|
except Exception as e:
|
|
signals.status.emit(f"Error during copy: {str(e)}")
|
|
signals.finished.emit(False, f"Copy failed: {str(e)}", failed_files)
|
|
|
|
class AdbStorageApp(QWidget):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.setWindowTitle("ADB Storage & Drive Info")
|
|
self.setGeometry(300, 300, 500, 500)
|
|
layout = QVBoxLayout()
|
|
|
|
# ADB Mounts Picker
|
|
self.mount_label = QLabel("Device Storage Mounts:")
|
|
self.mount_combo = QComboBox()
|
|
self.refresh_btn = QPushButton("Refresh Mounts")
|
|
self.refresh_btn.clicked.connect(self.load_mounts)
|
|
|
|
# Drive input
|
|
self.drive_label = QLabel("Drive:")
|
|
self.drive_input = QLineEdit("F:")
|
|
self.browse_btn = QPushButton("Browse")
|
|
self.browse_btn.clicked.connect(self.browse_drive)
|
|
|
|
# Volume ID display
|
|
self.volume_label = QLabel("Volume ID:")
|
|
self.volume_display = QLabel("")
|
|
|
|
# Copy button
|
|
self.copy_btn = QPushButton("Copy Selected Mount to Drive ROOT")
|
|
self.copy_btn.clicked.connect(self.start_copy)
|
|
self.copy_btn.setStyleSheet("background-color: #4CAF50; color: white; font-weight: bold; padding: 8px;")
|
|
|
|
# Export failed files button (initially hidden)
|
|
self.export_btn = QPushButton("Export Failed Files as ADB Commands")
|
|
self.export_btn.clicked.connect(self.export_failed_files)
|
|
self.export_btn.setStyleSheet("background-color: #ff9800; color: white; font-weight: bold; padding: 8px;")
|
|
self.export_btn.setVisible(False)
|
|
|
|
# Progress bar
|
|
self.progress_label = QLabel("Progress:")
|
|
self.progress_bar = QProgressBar()
|
|
self.progress_bar.setTextVisible(True)
|
|
|
|
# File progress label
|
|
self.file_progress_label = QLabel("")
|
|
|
|
# Failed files info
|
|
self.failed_files_label = QLabel("Failed files: 0")
|
|
self.failed_files_label.setStyleSheet("color: #f44336; font-weight: bold;")
|
|
|
|
# Status log
|
|
self.status_label = QLabel("Status:")
|
|
self.status_text = QTextEdit()
|
|
self.status_text.setReadOnly(True)
|
|
self.status_text.setMaximumHeight(100)
|
|
self.status_text.setFont(QFont("Courier", 9))
|
|
|
|
# Layout organization
|
|
layout.addWidget(self.mount_label)
|
|
layout.addWidget(self.mount_combo)
|
|
layout.addWidget(self.refresh_btn)
|
|
|
|
# Drive section
|
|
drive_layout = QHBoxLayout()
|
|
drive_layout.addWidget(self.drive_label)
|
|
drive_layout.addWidget(self.drive_input)
|
|
drive_layout.addWidget(self.browse_btn)
|
|
layout.addLayout(drive_layout)
|
|
|
|
layout.addWidget(self.volume_label)
|
|
layout.addWidget(self.volume_display)
|
|
layout.addWidget(self.copy_btn)
|
|
layout.addWidget(self.export_btn)
|
|
layout.addWidget(self.progress_label)
|
|
layout.addWidget(self.progress_bar)
|
|
layout.addWidget(self.file_progress_label)
|
|
layout.addWidget(self.failed_files_label)
|
|
layout.addWidget(self.status_label)
|
|
layout.addWidget(self.status_text)
|
|
|
|
self.setLayout(layout)
|
|
|
|
# Initialize
|
|
self.copy_thread = None
|
|
self.signals = CopySignals()
|
|
self.signals.progress.connect(self.update_progress)
|
|
self.signals.status.connect(self.update_status)
|
|
self.signals.finished.connect(self.copy_finished)
|
|
self.signals.file_progress.connect(self.update_file_progress)
|
|
|
|
self.failed_files = []
|
|
|
|
self.load_mounts()
|
|
self.update_volume_id()
|
|
|
|
def load_mounts(self):
|
|
self.mount_combo.clear()
|
|
mounts = get_adb_mounts()
|
|
self.mount_combo.addItems(mounts)
|
|
|
|
def browse_drive(self):
|
|
folder = QFileDialog.getExistingDirectory(self, "Select Drive")
|
|
if folder:
|
|
drive = folder.split(":")[0] + ":"
|
|
self.drive_input.setText(drive)
|
|
|
|
def update_volume_id(self):
|
|
drive = self.drive_input.text().strip()
|
|
if drive and len(drive) == 2 and drive[1] == ":":
|
|
vid = get_volume_id(drive)
|
|
self.volume_display.setText(vid)
|
|
else:
|
|
self.volume_display.setText("Invalid Drive")
|
|
|
|
def start_copy(self):
|
|
mount_selection = self.mount_combo.currentText()
|
|
drive_path = self.drive_input.text().strip()
|
|
|
|
if mount_selection.startswith("No device") or mount_selection.startswith("ADB error"):
|
|
self.update_status(f"Cannot copy: {mount_selection}")
|
|
return
|
|
|
|
if not drive_path or len(drive_path) != 2 or drive_path[1] != ":":
|
|
self.update_status("Please select a valid drive (e.g., F:)")
|
|
return
|
|
|
|
# Add backslash for Windows drive
|
|
drive_path = drive_path + "\\"
|
|
|
|
# Reset failed files
|
|
self.failed_files = []
|
|
self.failed_files_label.setText("Failed files: 0")
|
|
self.export_btn.setVisible(False)
|
|
|
|
# Disable controls during copy
|
|
self.copy_btn.setEnabled(False)
|
|
self.mount_combo.setEnabled(False)
|
|
self.refresh_btn.setEnabled(False)
|
|
self.drive_input.setEnabled(False)
|
|
self.browse_btn.setEnabled(False)
|
|
|
|
# Reset progress
|
|
self.progress_bar.setValue(0)
|
|
self.file_progress_label.setText("")
|
|
self.status_text.clear()
|
|
|
|
# Start copy in background thread
|
|
self.copy_thread = threading.Thread(
|
|
target=copy_mount_to_drive,
|
|
args=(mount_selection, drive_path, self.signals),
|
|
daemon=True
|
|
)
|
|
self.copy_thread.start()
|
|
|
|
def update_progress(self, current, total):
|
|
"""Update progress bar."""
|
|
if total > 0:
|
|
percent = int((current / total) * 100)
|
|
self.progress_bar.setValue(percent)
|
|
|
|
if total >= 1024*1024*1024:
|
|
current_str = f"{current/(1024*1024*1024):.2f}GB"
|
|
total_str = f"{total/(1024*1024*1024):.2f}GB"
|
|
elif total >= 1024*1024:
|
|
current_str = f"{current/(1024*1024):.2f}MB"
|
|
total_str = f"{total/(1024*1024):.2f}MB"
|
|
elif total >= 1024:
|
|
current_str = f"{current/1024:.2f}KB"
|
|
total_str = f"{total/1024:.2f}KB"
|
|
else:
|
|
current_str = f"{current}B"
|
|
total_str = f"{total}B"
|
|
|
|
self.progress_bar.setFormat(f"{percent}% ({current_str} / {total_str})")
|
|
else:
|
|
if current <= 100:
|
|
self.progress_bar.setValue(current)
|
|
self.progress_bar.setFormat(f"{current}%")
|
|
else:
|
|
self.progress_bar.setValue(current % 100)
|
|
self.progress_bar.setFormat("Copying...")
|
|
|
|
def update_file_progress(self, filename, current, total):
|
|
"""Update current file being copied."""
|
|
self.file_progress_label.setText(f"Copying: {filename} ({current}/{total})")
|
|
|
|
def update_status(self, message):
|
|
"""Add status message to log."""
|
|
timestamp = datetime.now().strftime("%H:%M:%S")
|
|
self.status_text.append(f"[{timestamp}] {message}")
|
|
self.status_text.verticalScrollBar().setValue(
|
|
self.status_text.verticalScrollBar().maximum()
|
|
)
|
|
|
|
def copy_finished(self, success, message, failed_files):
|
|
"""Handle copy completion."""
|
|
self.failed_files = failed_files
|
|
|
|
# Update failed files count
|
|
if failed_files:
|
|
self.failed_files_label.setText(f"Failed files: {len(failed_files)}")
|
|
self.failed_files_label.setStyleSheet("color: #f44336; font-weight: bold;")
|
|
self.export_btn.setVisible(True)
|
|
else:
|
|
self.failed_files_label.setText("Failed files: 0")
|
|
self.failed_files_label.setStyleSheet("color: black; font-weight: normal;")
|
|
|
|
self.update_status(message)
|
|
|
|
# Re-enable controls
|
|
self.copy_btn.setEnabled(True)
|
|
self.mount_combo.setEnabled(True)
|
|
self.refresh_btn.setEnabled(True)
|
|
self.drive_input.setEnabled(True)
|
|
self.browse_btn.setEnabled(True)
|
|
|
|
if success:
|
|
self.progress_bar.setValue(100)
|
|
self.progress_bar.setFormat("Copy completed!")
|
|
else:
|
|
self.progress_bar.setValue(0)
|
|
self.progress_bar.setFormat("Copy failed")
|
|
|
|
def export_failed_files(self):
|
|
"""Export failed files as ADB commands to a file."""
|
|
if not self.failed_files:
|
|
QMessageBox.information(self, "No Failed Files", "There are no failed files to export.")
|
|
return
|
|
|
|
# Ask user for save location
|
|
file_path, _ = QFileDialog.getSaveFileName(
|
|
self,
|
|
"Save ADB Commands",
|
|
f"adb_retry_commands_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt",
|
|
"Text Files (*.txt);;All Files (*.*)"
|
|
)
|
|
|
|
if file_path:
|
|
try:
|
|
with open(file_path, 'w', encoding='utf-8') as f:
|
|
f.write("# ADB commands to retry failed file copies\n")
|
|
f.write(f"# Generated on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
|
|
f.write(f"# Total failed files: {len(self.failed_files)}\n")
|
|
f.write("#\n")
|
|
f.write("# Commands:\n")
|
|
f.write("#\n")
|
|
|
|
for i, failed_file in enumerate(self.failed_files, 1):
|
|
f.write(f"\n# File {i}: {os.path.basename(failed_file['source'])}\n")
|
|
f.write(f"# Source: {failed_file['source']}\n")
|
|
f.write(f"# Destination: {failed_file['destination']}\n")
|
|
f.write(f"# Error: {failed_file['error']}\n")
|
|
f.write(f"{failed_file['adb_command']}\n")
|
|
|
|
# Also write a batch script version for Windows
|
|
f.write("\n" + "="*80 + "\n")
|
|
f.write("# Windows Batch Script (save as .bat file)\n")
|
|
f.write("@echo off\n")
|
|
f.write("echo Running ADB retry commands...\n")
|
|
f.write("\n")
|
|
|
|
for failed_file in self.failed_files:
|
|
# Escape quotes for batch file
|
|
cmd = failed_file['adb_command'].replace('"', '""')
|
|
f.write(f"{cmd}\n")
|
|
|
|
f.write("\necho All commands executed.\n")
|
|
f.write("pause\n")
|
|
|
|
QMessageBox.information(
|
|
self,
|
|
"Export Successful",
|
|
f"ADB commands exported to:\n{file_path}\n\n"
|
|
f"Total commands: {len(self.failed_files)}\n\n"
|
|
"You can also copy the commands to a .bat file for easy execution."
|
|
)
|
|
|
|
except Exception as e:
|
|
QMessageBox.critical(
|
|
self,
|
|
"Export Failed",
|
|
f"Failed to export ADB commands:\n{str(e)}"
|
|
)
|
|
|
|
if __name__ == "__main__":
|
|
app = QApplication(sys.argv)
|
|
window = AdbStorageApp()
|
|
window.show()
|
|
sys.exit(app.exec()) |