From 312ef867945551fb5cf7345ad11163708d50ae14 Mon Sep 17 00:00:00 2001 From: Jason Ross Date: Thu, 21 Aug 2025 13:37:06 -0500 Subject: [PATCH] added individual file browsing and transfer capabilities --- src/adb_manager.py | 190 ++++++++++++++++++++++++++++++++++++++++ src/gui/file_browser.py | 93 +++++++++++++++++--- src/gui/main_window.py | 141 +++++++++++++++++++++++++---- 3 files changed, 395 insertions(+), 29 deletions(-) diff --git a/src/adb_manager.py b/src/adb_manager.py index ef09300..0937989 100644 --- a/src/adb_manager.py +++ b/src/adb_manager.py @@ -429,6 +429,196 @@ class ADBManager: self.current_process = None return False + def pull_file(self, remote_file_path: str, local_file_path: str) -> bool: + """Pull a single file from Android device to local machine.""" + # Normalize paths for better compatibility + local_file_path = os.path.normpath(local_file_path) + remote_file_path = remote_file_path.strip() + + # Ensure local directory exists + local_dir = os.path.dirname(local_file_path) + try: + if local_dir: # Only create if there's a directory part + os.makedirs(local_dir, exist_ok=True) + except Exception as e: + self._update_status(f"Failed to create local directory: {e}") + return False + + # For Windows root drives, ensure proper formatting + if os.name == "nt": + # Check if this is a root drive (like C:\, D:\, etc.) + normalized_path = os.path.abspath(local_dir) + drive_root = os.path.splitdrive(normalized_path)[0] + os.sep + if normalized_path == drive_root: + # Root drive path like C:\ - this might cause issues with ADB + self._update_status( + "Warning: Transferring to root drive. Consider using a subfolder." + ) + + cmd = [ADB_BINARY_PATH, "pull", remote_file_path, local_file_path] + + # Debug output for troubleshooting + self._update_status(f"Command: adb pull '{remote_file_path}' '{local_file_path}'") + + try: + # Start with initial progress + self._update_progress(0) + self._update_status("Starting file transfer...") + + proc = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + ) + self.current_process = proc + + line_count = 0 + last_progress = 0 + start_time = time.time() + last_update_time = start_time + + if proc.stdout: + for line in proc.stdout: + line_count += 1 + current_time = time.time() + pct = self.parse_progress(line) + + if pct is not None: + # Use explicit progress when available + self._update_progress(pct) + last_progress = pct + last_update_time = current_time + else: + # For single files, use simpler progress estimation + elapsed_time = current_time - start_time + if elapsed_time >= 1.0 and last_progress < 90: + # Simple time-based progress for files + new_progress = min(last_progress + 20, 90) + self._update_progress(int(new_progress)) + last_progress = new_progress + last_update_time = current_time + + self._update_status(line.strip()) + + proc.wait() + if proc.returncode == 0: + self._update_progress(100) + self._update_status("File transfer completed successfully.") + self.current_process = None + return True + else: + # Capture error output for better debugging + error_msg = f"File transfer failed with code {proc.returncode}" + if hasattr(proc, "stderr") and proc.stderr: + try: + stderr_output = proc.stderr.read() + if stderr_output: + error_msg += f". Error: {stderr_output}" + except: + pass + self._update_status(error_msg) + self.current_process = None + return False + + except Exception as e: + self._update_status(f"File transfer error: {e}") + self.current_process = None + return False + + def push_file(self, local_file_path: str, remote_file_path: str) -> bool: + """Push a single file from local machine to Android device.""" + # Normalize paths for better compatibility + local_file_path = os.path.normpath(local_file_path) + remote_file_path = remote_file_path.strip() + + # Validate local file exists + if not os.path.isfile(local_file_path): + self._update_status(f"Local file does not exist: {local_file_path}") + return False + + # For Windows root drives, ensure proper formatting + if os.name == "nt": + # Check if this is a root drive (like C:\, D:\, etc.) + normalized_path = os.path.abspath(local_file_path) + drive_root = os.path.splitdrive(normalized_path)[0] + os.sep + if os.path.dirname(normalized_path) == drive_root.rstrip(os.sep): + # File in root drive - this might cause issues with ADB + self._update_status( + "Warning: Pushing from root drive. Consider using a subfolder." + ) + + cmd = [ADB_BINARY_PATH, "push", local_file_path, remote_file_path] + + # Debug output for troubleshooting + self._update_status(f"Command: adb push '{local_file_path}' '{remote_file_path}'") + + try: + # Start with initial progress + self._update_progress(0) + self._update_status("Starting file transfer...") + + proc = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + ) + self.current_process = proc + except Exception as e: + self._update_status(f"Failed to start adb: {e}") + return False + + line_count = 0 + last_progress = 0 + start_time = time.time() + last_update_time = start_time + + if proc.stdout: + for line in proc.stdout: + line_count += 1 + current_time = time.time() + pct = self.parse_progress(line) + + if pct is not None: + # Use explicit progress when available + self._update_progress(pct) + last_progress = pct + last_update_time = current_time + else: + # For single files, use simpler progress estimation + elapsed_time = current_time - start_time + if elapsed_time >= 1.0 and last_progress < 90: + # Simple time-based progress for files + new_progress = min(last_progress + 20, 90) + self._update_progress(int(new_progress)) + last_progress = new_progress + last_update_time = current_time + + self._update_status(line.strip()) + + proc.wait() + if proc.returncode == 0: + self._update_progress(100) + self._update_status("File transfer completed successfully.") + self.current_process = None + return True + else: + # Capture error output for better debugging + error_msg = f"File push failed with code {proc.returncode}" + if hasattr(proc, "stderr") and proc.stderr: + try: + stderr_output = proc.stderr.read() + if stderr_output: + error_msg += f". Error: {stderr_output}" + except: + pass + self._update_status(error_msg) + self.current_process = None + return False + def cancel_transfer(self) -> bool: """Cancel the current transfer operation.""" if self.current_process is not None: diff --git a/src/gui/file_browser.py b/src/gui/file_browser.py index e4769f9..708bb74 100644 --- a/src/gui/file_browser.py +++ b/src/gui/file_browser.py @@ -29,14 +29,14 @@ class AndroidFileBrowser: # Create browsable folder dialog browser_window = tk.Toplevel(self.parent) - browser_window.title("Browse Android Folders") + browser_window.title("Browse Android Files and Folders") browser_window.geometry("500x400") browser_window.transient(self.parent) browser_window.grab_set() tk.Label( browser_window, - text="Browse Android device folders:", + text="Browse Android device files and folders:", font=("Arial", 10, "bold"), ).pack(pady=10) @@ -136,12 +136,13 @@ class AndroidFileBrowser: ) return - # Parse ls -la output to find directories + # Parse ls -la output to find directories and files folders = [] + files = [] lines = stdout.strip().split("\n") for line in lines: if line.startswith("d"): - # Extract folder name - improved parsing to prevent truncation + # Directory entry parts = line.split() if len(parts) >= 8: # Method 1: Use regex to find time pattern and extract name after it @@ -173,23 +174,67 @@ class AndroidFileBrowser: and not folder_name.startswith(".") ): folders.append(folder_name) + elif line.startswith("-"): + # Regular file entry + parts = line.split() + if len(parts) >= 8: + # Extract file name using same method as folders + import re - # Add folders to tree + time_pattern = r"\d{1,2}:\d{2}" + time_matches = list(re.finditer(time_pattern, line)) + + if time_matches: + # Take everything after the last time pattern + last_time_match = time_matches[-1] + file_name = line[ + last_time_match.end() : + ].strip() + else: + # Fallback: join from part 8 (skip date/time fields) + file_name = ( + " ".join(parts[8:]) + if len(parts) > 8 + else parts[7] + ) + + # Additional validation and cleanup + file_name = file_name.strip() + + if ( + file_name + and not file_name.startswith(".") + ): + files.append(file_name) + + # Add folders to tree first (sorted) if folders: for folder in sorted(folders): folder_path = f"{path.rstrip('/')}/{folder}" item = tree.insert( parent_item, "end", - text=folder, - values=[folder_path], + text=f"📁 {folder}", + values=[folder_path, "folder"], ) # Add a dummy child to make it expandable tree.insert(item, "end", text="Loading...") - else: - # No folders found, show indicator + + # Add files to tree (sorted) + if files: + for file in sorted(files): + file_path = f"{path.rstrip('/')}/{file}" + tree.insert( + parent_item, + "end", + text=f"📄 {file}", + values=[file_path, "file"], + ) + + # If no folders or files found, show indicator + if not folders and not files: tree.insert( - parent_item, "end", text="(No Folders)", values=[""] + parent_item, "end", text="(Empty Directory)", values=["", ""] ) self.parent.after(0, update_tree) @@ -395,17 +440,37 @@ class AndroidFileBrowser: button_frame = tk.Frame(browser_window) button_frame.pack(pady=10) - def select_current_folder(): - """Select the currently highlighted folder.""" + def select_current_item(): + """Intelligently select the currently highlighted item (file or folder).""" + selected_item = tree.selection()[0] if tree.selection() else None + + if selected_item: + # Get the item type and path from values + item_values = tree.item(selected_item, "values") + if len(item_values) >= 1: + item_path = item_values[0] + # Check if it's a file or folder (if we have type info) + if len(item_values) >= 2 and item_values[1] == "file": + # It's a file - select the file path directly + self.remote_path_var.set(item_path) + browser_window.destroy() + return + elif len(item_values) >= 2 and item_values[1] == "folder": + # It's a folder - select the folder path directly + self.remote_path_var.set(item_path) + browser_window.destroy() + return + + # Fallback: use current path (for backwards compatibility or when no specific item is selected) current_path = current_path_var.get() if current_path and current_path.strip(): self.remote_path_var.set(current_path) browser_window.destroy() else: - messagebox.showwarning("No Selection", "Please select a valid folder.") + messagebox.showwarning("No Selection", "Please select a file or folder.") tk.Button( - button_frame, text="Select This Folder", command=select_current_folder + button_frame, text="Select", command=select_current_item ).pack(side="left", padx=5) tk.Button(button_frame, text="Cancel", command=browser_window.destroy).pack( side="left", padx=5 diff --git a/src/gui/main_window.py b/src/gui/main_window.py index 670d0d8..418c7ac 100644 --- a/src/gui/main_window.py +++ b/src/gui/main_window.py @@ -243,14 +243,46 @@ class AndroidFileHandlerGUI(tk.Tk): self.enable_controls() def browse_remote_folder(self): - """Browse remote Android folders.""" + """Browse remote Android files and folders.""" self.browser.show_browser() def browse_local_folder(self): - """Browse for local folder.""" - folder = filedialog.askdirectory() - if folder: - self.local_path_var.set(folder) + """Browse for local folder or file based on direction.""" + direction = self.direction_var.get() + + if direction == "push": + # For push, show a unified dialog that handles both files and folders + # Start with file selection dialog that has "Open" and "Cancel" + # If user selects a file, use it. If they navigate and click "Cancel" on file dialog, + # we'll assume they want to select the current folder + + import tkinter.filedialog as fd + import os + + # Use askopenfilename but allow folder selection by providing both options + # First try file selection + selected_path = fd.askopenfilename( + title="Select file to push (or Cancel and then select folder)", + filetypes=[("All files", "*.*")] + ) + + # If no file was selected, offer folder selection + if not selected_path: + selected_path = fd.askdirectory( + title="Select folder to push to Android device" + ) + + # Set the path if something was selected + if selected_path: + self.local_path_var.set(selected_path) + + else: # pull direction + # For pull, only allow folder selection (destination) + folder = filedialog.askdirectory( + title="Select destination folder for pulled files" + ) + if folder: + self.local_path_var.set(folder) def disable_controls(self): """Disable UI controls during operations.""" @@ -406,13 +438,22 @@ class AndroidFileHandlerGUI(tk.Tk): # Validate inputs if not remote_path: - messagebox.showerror("Input Error", "Remote folder path cannot be empty.") - return - if not local_path or not os.path.isdir(local_path): - messagebox.showerror( - "Input Error", "Please select a valid local destination folder." - ) + messagebox.showerror("Input Error", "Remote path cannot be empty.") return + + # For push operations, validate local path + if direction == "push": + if not local_path or (not os.path.isfile(local_path) and not os.path.isdir(local_path)): + messagebox.showerror( + "Input Error", "Please select a valid local file or folder." + ) + return + else: # pull operations + if not local_path or not os.path.isdir(local_path): + messagebox.showerror( + "Input Error", "Please select a valid local destination folder for pulled files." + ) + return # Final device check before transfer device = self.adb_manager.check_device() @@ -435,22 +476,63 @@ class AndroidFileHandlerGUI(tk.Tk): # Start the transfer animation self._start_transfer_animation() + # Determine if we're dealing with files or folders if direction == "pull": + # For pull, check if remote path is a file + is_file = self._is_remote_file(remote_path) threading.Thread( - target=self._pull_thread, - args=(remote_path, local_path, transfer_id), + target=self._transfer_thread, + args=(direction, remote_path, local_path, transfer_id, is_file), daemon=True, ).start() elif direction == "push": + # For push, check if local path is a file + is_file = os.path.isfile(local_path) threading.Thread( - target=self._push_thread, - args=(local_path, remote_path, transfer_id), + target=self._transfer_thread, + args=(direction, local_path, remote_path, transfer_id, is_file), daemon=True, ).start() else: self.report_error("Invalid transfer direction selected.") self.enable_controls() + def _transfer_thread(self, direction: str, source_path: str, dest_path: str, transfer_id: int, is_file: bool): + """Unified thread function for all transfer operations.""" + try: + # Check if this transfer is still current + if self.current_transfer_id != transfer_id: + return + + # Get the appropriate transfer method + transfer_method, transfer_type = self._get_file_transfer_methods(direction, is_file) + + # For pull operations with files, we need to construct the full destination path + if direction == "pull" and is_file: + # Extract filename from source and append to destination directory + filename = source_path.split("/")[-1] + full_dest_path = os.path.join(dest_path, filename) + success = transfer_method(source_path, full_dest_path) + else: + success = transfer_method(source_path, dest_path) + + if success and self.current_transfer_id == transfer_id: + self._stop_transfer_animation() + transfer_desc = "File" if is_file else "Folder" + self._update_status(f"{transfer_desc} transfer completed successfully.") + self.show_disable_debugging_reminder() + except Exception as e: + if self.current_transfer_id == transfer_id: + self._stop_transfer_animation() + transfer_desc = "file" if is_file else "folder" + self.report_error(f"{direction.capitalize()} {transfer_desc} operation failed: {e}") + finally: + if self.current_transfer_id == transfer_id: + self._stop_transfer_animation() + self.enable_controls() + # Restore proper button state after transfer + self.after(0, self._restore_button_state) + def _pull_thread(self, remote_path: str, local_path: str, transfer_id: int): """Thread function for pull operations.""" try: @@ -504,6 +586,35 @@ class AndroidFileHandlerGUI(tk.Tk): else: self._switch_to_recheck_mode() + def _is_remote_file(self, remote_path: str) -> bool: + """Check if the remote path points to a file (not a directory).""" + try: + # Use ls -la to check if it's a file + result = self.adb_manager.run_adb_command( + ["shell", "ls", "-la", remote_path] + ) + if isinstance(result, tuple) and len(result) == 3: + stdout, stderr, returncode = result + if returncode == 0 and stdout: + # If the output starts with '-', it's a regular file + return stdout.strip().startswith('-') + return False + except Exception: + return False + + def _get_file_transfer_methods(self, direction: str, is_file: bool): + """Get the appropriate transfer methods based on direction and type.""" + if direction == "pull": + if is_file: + return self.adb_manager.pull_file, "file" + else: + return self.adb_manager.pull_folder, "folder" + else: # push + if is_file: + return self.adb_manager.push_file, "file" + else: + return self.adb_manager.push_folder, "folder" + def report_error(self, message: str): """Report an error to the user (thread-safe).""" # Schedule UI update on main thread