added individual file browsing and transfer capabilities
This commit is contained in:
@@ -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:
|
||||
|
||||
+79
-14
@@ -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
|
||||
|
||||
+126
-15
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user