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())