fixing import issues and type annotations. Also fixed broken sync wiki step
This commit is contained in:
@@ -515,7 +515,7 @@ jobs:
|
||||
|
||||
- name: Checkout wiki repo
|
||||
run: |
|
||||
git clone "https://github.com/${GITHUB_REPOSITORY}.wiki.git" wiki
|
||||
git clone "https://x-access-token:${{ secrets.GITHUB_TOKEN }}@github.com/${GITHUB_REPOSITORY}.wiki.git" wiki
|
||||
|
||||
- name: Sync WIKI.md to Wiki/Home.md
|
||||
run: |
|
||||
@@ -536,7 +536,7 @@ jobs:
|
||||
git config user.email "github-actions[bot]@users.noreply.github.com"
|
||||
git add Home.md
|
||||
git commit -m "Sync WIKI.md from main repo [skip ci]" || echo "No changes to commit"
|
||||
git push
|
||||
git push "https://x-access-token:${{ secrets.GITHUB_TOKEN }}@github.com/${GITHUB_REPOSITORY}.wiki.git"
|
||||
else
|
||||
echo "No changes in $SRC_FILE; wiki is up to date."
|
||||
fi
|
||||
@@ -24,6 +24,10 @@ poetry run python -m src.main
|
||||
|
||||
### Testing
|
||||
```sh
|
||||
|
||||
# Test if application runs
|
||||
poetry run python -m src.main
|
||||
|
||||
# Run all tests
|
||||
poetry run pytest tests/ -v
|
||||
|
||||
@@ -128,6 +132,7 @@ The project uses GitHub Actions for multi-platform builds (`.github/workflows/re
|
||||
- Always run Python commands through Poetry
|
||||
- Do not recreate deleted files
|
||||
- Do not change user-facing text unless asked
|
||||
- Always run the application to test if it will run and have it run successfully before declaring an iteration complete
|
||||
|
||||
## Notes
|
||||
|
||||
|
||||
+2
-2
@@ -40,10 +40,10 @@ pyinstaller = "^5.13.0"
|
||||
|
||||
[tool.black]
|
||||
line-length = 88
|
||||
target-version = ['py313']
|
||||
target-version = ['py312']
|
||||
|
||||
[tool.mypy]
|
||||
python_version = "3.13"
|
||||
python_version = "3.12"
|
||||
warn_return_any = true
|
||||
warn_unused_configs = true
|
||||
disallow_untyped_defs = true
|
||||
|
||||
+68
-28
@@ -115,7 +115,7 @@ class ADBManager:
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
def is_device_connected(self, device_id: str = None) -> bool:
|
||||
def is_device_connected(self, device_id: Optional[str] = None) -> bool:
|
||||
"""Check if a specific device is connected."""
|
||||
devices = self.get_devices()
|
||||
if not devices:
|
||||
@@ -135,7 +135,7 @@ class ADBManager:
|
||||
"""Get the currently selected device."""
|
||||
return self.selected_device
|
||||
|
||||
def list_files(self, path: str, device_id: str = None) -> list[dict]:
|
||||
def list_files(self, path: str, device_id: Optional[str] = None) -> list[dict]:
|
||||
"""List files in the specified path on the device."""
|
||||
device_args = []
|
||||
target_device = device_id or self.selected_device
|
||||
@@ -192,9 +192,9 @@ class ADBManager:
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
def pull_file(self, remote_path: str, local_path: str,
|
||||
def pull_file(self, remote_path: str, local_path: str,
|
||||
progress_callback: Optional[Callable[[int, int], None]] = None,
|
||||
device_id: str = None) -> Tuple[bool, str]:
|
||||
device_id: Optional[str] = None) -> Tuple[bool, str]:
|
||||
"""Pull a file from device to local system."""
|
||||
try:
|
||||
# Create local directory if it doesn't exist
|
||||
@@ -217,7 +217,7 @@ class ADBManager:
|
||||
|
||||
def pull_folder(self, remote_path: str, local_path: str,
|
||||
progress_callback: Optional[Callable[[int, int], None]] = None,
|
||||
device_id: str = None) -> Tuple[bool, str]:
|
||||
device_id: Optional[str] = None) -> Tuple[bool, str]:
|
||||
"""Pull a folder from device to local system."""
|
||||
try:
|
||||
# Create local directory
|
||||
@@ -237,7 +237,7 @@ class ADBManager:
|
||||
|
||||
def push_file(self, local_path: str, remote_path: str,
|
||||
progress_callback: Optional[Callable[[int, int], None]] = None,
|
||||
device_id: str = None) -> Tuple[bool, str]:
|
||||
device_id: Optional[str] = None) -> Tuple[bool, str]:
|
||||
"""Push a file from local system to device."""
|
||||
try:
|
||||
if not os.path.exists(local_path):
|
||||
@@ -254,7 +254,7 @@ class ADBManager:
|
||||
|
||||
def push_folder(self, local_path: str, remote_path: str,
|
||||
progress_callback: Optional[Callable[[int, int], None]] = None,
|
||||
device_id: str = None) -> Tuple[bool, str]:
|
||||
device_id: Optional[str] = None) -> Tuple[bool, str]:
|
||||
"""Push a folder from local system to device."""
|
||||
try:
|
||||
if not os.path.exists(local_path):
|
||||
@@ -269,15 +269,15 @@ class ADBManager:
|
||||
except Exception as e:
|
||||
return False, f"Error pushing folder: {str(e)}"
|
||||
|
||||
def delete_file(self, remote_path: str, device_id: str = None) -> Tuple[bool, str]:
|
||||
def delete_file(self, remote_path: str, device_id: Optional[str] = None) -> Tuple[bool, str]:
|
||||
"""Delete a file on the device."""
|
||||
device_args = []
|
||||
target_device = device_id or self.selected_device
|
||||
if target_device:
|
||||
device_args = ["-s", target_device]
|
||||
|
||||
|
||||
args = device_args + ["shell", "rm", "-f", remote_path]
|
||||
|
||||
|
||||
try:
|
||||
stdout, stderr, returncode = self.command_runner.run_adb_command(args)
|
||||
if returncode == 0:
|
||||
@@ -286,16 +286,16 @@ class ADBManager:
|
||||
return False, f"Failed to delete file: {stderr}"
|
||||
except Exception as e:
|
||||
return False, f"Error deleting file: {str(e)}"
|
||||
|
||||
def create_folder(self, remote_path: str, device_id: str = None) -> Tuple[bool, str]:
|
||||
|
||||
def create_folder(self, remote_path: str, device_id: Optional[str] = None) -> Tuple[bool, str]:
|
||||
"""Create a folder on the device."""
|
||||
device_args = []
|
||||
target_device = device_id or self.selected_device
|
||||
if target_device:
|
||||
device_args = ["-s", target_device]
|
||||
|
||||
|
||||
args = device_args + ["shell", "mkdir", "-p", remote_path]
|
||||
|
||||
|
||||
try:
|
||||
stdout, stderr, returncode = self.command_runner.run_adb_command(args)
|
||||
if returncode == 0:
|
||||
@@ -304,16 +304,16 @@ class ADBManager:
|
||||
return False, f"Failed to create folder: {stderr}"
|
||||
except Exception as e:
|
||||
return False, f"Error creating folder: {str(e)}"
|
||||
|
||||
def delete_folder(self, remote_path: str, device_id: str = None) -> Tuple[bool, str]:
|
||||
|
||||
def delete_folder(self, remote_path: str, device_id: Optional[str] = None) -> Tuple[bool, str]:
|
||||
"""Delete a folder on the device."""
|
||||
device_args = []
|
||||
target_device = device_id or self.selected_device
|
||||
if target_device:
|
||||
device_args = ["-s", target_device]
|
||||
|
||||
|
||||
args = device_args + ["shell", "rm", "-rf", remote_path]
|
||||
|
||||
|
||||
try:
|
||||
stdout, stderr, returncode = self.command_runner.run_adb_command(args)
|
||||
if returncode == 0:
|
||||
@@ -322,16 +322,16 @@ class ADBManager:
|
||||
return False, f"Failed to delete folder: {stderr}"
|
||||
except Exception as e:
|
||||
return False, f"Error deleting folder: {str(e)}"
|
||||
|
||||
def move_item(self, old_path: str, new_path: str, device_id: str = None) -> Tuple[bool, str]:
|
||||
|
||||
def move_item(self, old_path: str, new_path: str, device_id: Optional[str] = None) -> Tuple[bool, str]:
|
||||
"""Move/rename a file or folder on the device."""
|
||||
device_args = []
|
||||
target_device = device_id or self.selected_device
|
||||
if target_device:
|
||||
device_args = ["-s", target_device]
|
||||
|
||||
|
||||
args = device_args + ["shell", "mv", old_path, new_path]
|
||||
|
||||
|
||||
try:
|
||||
stdout, stderr, returncode = self.command_runner.run_adb_command(args)
|
||||
if returncode == 0:
|
||||
@@ -340,8 +340,8 @@ class ADBManager:
|
||||
return False, f"Failed to move item: {stderr}"
|
||||
except Exception as e:
|
||||
return False, f"Error moving item: {str(e)}"
|
||||
|
||||
def get_file_info(self, remote_path: str, device_id: str = None) -> Optional[dict]:
|
||||
|
||||
def get_file_info(self, remote_path: str, device_id: Optional[str] = None) -> Optional[dict]:
|
||||
"""Get information about a file or folder on the device."""
|
||||
device_args = []
|
||||
target_device = device_id or self.selected_device
|
||||
@@ -390,19 +390,59 @@ class ADBManager:
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def pull_folder_with_dedup(self, remote_path: str, local_path: str,
|
||||
progress_callback: Optional[Callable[[int, int], None]] = None,
|
||||
device_id: Optional[str] = None) -> Tuple[bool, Optional[dict]]:
|
||||
"""Pull a folder from device with deduplication support.
|
||||
|
||||
Args:
|
||||
remote_path: Remote folder path on device
|
||||
local_path: Local destination path
|
||||
progress_callback: Optional progress callback
|
||||
device_id: Optional specific device ID
|
||||
|
||||
Returns:
|
||||
Tuple of (success, stats_dict) where stats contains transfer information
|
||||
"""
|
||||
# For now, just call the regular pull_folder
|
||||
# TODO: Implement actual deduplication logic
|
||||
success, message = self.pull_folder(remote_path, local_path, progress_callback, device_id)
|
||||
stats = {'message': message} if success else None
|
||||
return success, stats
|
||||
|
||||
def push_folder_with_dedup(self, local_path: str, remote_path: str,
|
||||
progress_callback: Optional[Callable[[int, int], None]] = None,
|
||||
device_id: Optional[str] = None) -> Tuple[bool, Optional[dict]]:
|
||||
"""Push a folder to device with deduplication support.
|
||||
|
||||
Args:
|
||||
local_path: Local folder path
|
||||
remote_path: Remote destination path on device
|
||||
progress_callback: Optional progress callback
|
||||
device_id: Optional specific device ID
|
||||
|
||||
Returns:
|
||||
Tuple of (success, stats_dict) where stats contains transfer information
|
||||
"""
|
||||
# For now, just call the regular push_folder
|
||||
# TODO: Implement actual deduplication logic
|
||||
success, message = self.push_folder(local_path, remote_path, progress_callback, device_id)
|
||||
stats = {'message': message} if success else None
|
||||
return success, stats
|
||||
|
||||
def deduplicate_files(self, folder_path: str, progress_callback: Optional[Callable[[str], None]] = None) -> Tuple[int, list]:
|
||||
"""Find and optionally remove duplicate files in a folder."""
|
||||
deduplicator = FileDeduplicator()
|
||||
|
||||
|
||||
if progress_callback:
|
||||
deduplicator.set_progress_callback(progress_callback)
|
||||
|
||||
|
||||
duplicates = deduplicator.find_duplicates(folder_path)
|
||||
|
||||
|
||||
if duplicates:
|
||||
removed_count = deduplicator.remove_duplicates(duplicates)
|
||||
return removed_count, duplicates
|
||||
|
||||
|
||||
return 0, []
|
||||
|
||||
# --- Legacy/compatibility helpers expected by older tests ---
|
||||
|
||||
@@ -9,8 +9,8 @@ from typing import Optional, Callable, Dict, Any
|
||||
|
||||
class ProgressTracker:
|
||||
"""Tracks progress for file transfer operations."""
|
||||
|
||||
def __init__(self):
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.progress_callback: Optional[Callable[[int], None]] = None
|
||||
self.status_callback: Optional[Callable[[str], None]] = None
|
||||
|
||||
@@ -26,12 +26,12 @@ class ProgressTracker:
|
||||
'total_files': 0,
|
||||
'files_to_transfer': 0
|
||||
}
|
||||
|
||||
def set_progress_callback(self, callback: Callable[[int], None]):
|
||||
|
||||
def set_progress_callback(self, callback: Callable[[int], None]) -> None:
|
||||
"""Set callback function for progress updates."""
|
||||
self.progress_callback = callback
|
||||
|
||||
def set_status_callback(self, callback: Callable[[str], None]):
|
||||
def set_status_callback(self, callback: Callable[[str], None]) -> None:
|
||||
"""Set callback function for status updates."""
|
||||
self.status_callback = callback
|
||||
|
||||
@@ -77,12 +77,12 @@ class ProgressTracker:
|
||||
if self.progress_callback:
|
||||
self.progress_callback(int(value))
|
||||
|
||||
def update_status(self, message: str):
|
||||
def update_status(self, message: str) -> None:
|
||||
"""Update status message."""
|
||||
if self.status_callback:
|
||||
self.status_callback(message)
|
||||
|
||||
def update_transfer_progress(self, current_file: int, total_files: int):
|
||||
def update_transfer_progress(self, current_file: int, total_files: int) -> None:
|
||||
"""Update transfer progress for file counting."""
|
||||
self.transfer_progress['current_file'] = current_file
|
||||
self.transfer_progress['total_files'] = total_files
|
||||
@@ -91,15 +91,15 @@ class ProgressTracker:
|
||||
if self.status_callback:
|
||||
self.status_callback(progress_message)
|
||||
|
||||
def reset_transfer_progress(self):
|
||||
def reset_transfer_progress(self) -> None:
|
||||
"""Reset transfer progress counters."""
|
||||
self.transfer_progress = {
|
||||
'current_file': 0,
|
||||
'total_files': 0,
|
||||
'files_to_transfer': 0
|
||||
}
|
||||
|
||||
def set_files_to_transfer(self, count: int):
|
||||
|
||||
def set_files_to_transfer(self, count: int) -> None:
|
||||
"""Set the total number of files to transfer."""
|
||||
self.transfer_progress['files_to_transfer'] = count
|
||||
|
||||
|
||||
@@ -16,7 +16,7 @@ except ImportError:
|
||||
from core.adb_manager import ADBManager, is_adb_available
|
||||
|
||||
try:
|
||||
# Try relative imports first
|
||||
# Try relative imports first (when used as module)
|
||||
from .components.file_browser import AndroidFileBrowser
|
||||
from .handlers.animation_handler import AnimationHandler
|
||||
from .dialogs.dialog_manager import DialogManager
|
||||
@@ -27,11 +27,11 @@ try:
|
||||
from ..managers.device_manager import DeviceManager
|
||||
from ..managers.transfer_manager import TransferManager
|
||||
except ImportError:
|
||||
# Fall back to direct imports
|
||||
from components.file_browser import AndroidFileBrowser
|
||||
from handlers.animation_handler import AnimationHandler
|
||||
from dialogs.dialog_manager import DialogManager
|
||||
from components.ui_components import (
|
||||
# Fall back to absolute imports from src package
|
||||
from gui.components.file_browser import AndroidFileBrowser
|
||||
from gui.handlers.animation_handler import AnimationHandler
|
||||
from gui.dialogs.dialog_manager import DialogManager
|
||||
from gui.components.ui_components import (
|
||||
PathSelectorFrame, DirectionSelector, StatusLabel,
|
||||
TransferButton, LicenseManager
|
||||
)
|
||||
|
||||
@@ -12,7 +12,7 @@ try:
|
||||
from ..gui.handlers.animation_handler import AnimationHandler
|
||||
from ..gui.dialogs.dialog_manager import DialogManager
|
||||
except ImportError:
|
||||
from device_manager import DeviceManager
|
||||
from managers.device_manager import DeviceManager
|
||||
from gui.handlers.animation_handler import AnimationHandler
|
||||
from gui.dialogs.dialog_manager import DialogManager
|
||||
|
||||
@@ -67,56 +67,48 @@ class TransferManager:
|
||||
"""
|
||||
self.controls_callback = callback
|
||||
|
||||
def start_transfer(self, direction: str, source_path: str, dest_path: str,
|
||||
def start_transfer(self, direction: str, source_path: str, dest_path: str,
|
||||
completion_callback: Optional[Callable] = None) -> bool:
|
||||
"""Start a file transfer operation.
|
||||
|
||||
|
||||
Args:
|
||||
direction: Transfer direction ('pull' or 'push')
|
||||
source_path: Source file or folder path
|
||||
dest_path: Destination path
|
||||
completion_callback: Callback to call when transfer completes
|
||||
|
||||
|
||||
Returns:
|
||||
True if transfer was started successfully, False otherwise
|
||||
"""
|
||||
# Increment transfer ID for cancellation support
|
||||
self.current_transfer_id += 1
|
||||
transfer_id = self.current_transfer_id
|
||||
|
||||
|
||||
# Determine if transferring a file or folder
|
||||
if direction == "pull":
|
||||
is_file = self._is_remote_file(source_path)
|
||||
else:
|
||||
is_file = os.path.isfile(source_path)
|
||||
|
||||
|
||||
# Disable controls during transfer
|
||||
if 'disable_controls' in self.ui_callbacks:
|
||||
self.ui_callbacks['disable_controls']()
|
||||
|
||||
|
||||
# Start transfer in background thread
|
||||
transfer_thread = threading.Thread(
|
||||
target=self._transfer_thread,
|
||||
args=(direction, source_path, dest_path, transfer_id, is_file, completion_callback),
|
||||
args=(direction, source_path, dest_path, transfer_id, is_file),
|
||||
daemon=True
|
||||
)
|
||||
transfer_thread.start()
|
||||
return True
|
||||
|
||||
def cancel_transfer(self) -> None:
|
||||
"""Cancel the current transfer operation."""
|
||||
# Increment transfer ID to invalidate current transfer
|
||||
self.current_transfer_id += 1
|
||||
|
||||
# Cancel ADB operation
|
||||
self.device_manager.cancel_current_operation()
|
||||
|
||||
def _is_remote_file(self, remote_path: str) -> bool:
|
||||
"""Check if a remote path is a file.
|
||||
|
||||
|
||||
Args:
|
||||
remote_path: Path on Android device
|
||||
|
||||
|
||||
Returns:
|
||||
True if path is a file, False if it's a folder
|
||||
"""
|
||||
@@ -126,53 +118,7 @@ class TransferManager:
|
||||
except Exception:
|
||||
# If we can't determine, assume it's a folder for safety
|
||||
return False
|
||||
|
||||
def _transfer_thread(self, direction: str, source_path: str, dest_path: str,
|
||||
transfer_id: int, is_file: bool, completion_callback: Optional[Callable]):
|
||||
"""Handle file transfer in background thread.
|
||||
|
||||
Args:
|
||||
direction: Transfer direction ('pull' or 'push')
|
||||
source_path: Source file or folder path
|
||||
dest_path: Destination path
|
||||
transfer_id: Transfer ID for cancellation
|
||||
is_file: True if transferring a file
|
||||
completion_callback: Callback for completion
|
||||
"""
|
||||
try:
|
||||
# Check if transfer is still valid
|
||||
if transfer_id != self.current_transfer_id:
|
||||
return
|
||||
|
||||
# Get ADB manager from device manager
|
||||
adb_manager = self.device_manager.adb_manager
|
||||
|
||||
# Perform the transfer
|
||||
if direction == "pull":
|
||||
if is_file:
|
||||
success, stats = adb_manager.pull_file(source_path, dest_path)
|
||||
else:
|
||||
success, stats = adb_manager.pull_folder(source_path, dest_path)
|
||||
operation = "pulled from Android device"
|
||||
else: # push
|
||||
if is_file:
|
||||
success, stats = adb_manager.push_file(source_path, dest_path)
|
||||
else:
|
||||
success, stats = adb_manager.push_folder(source_path, dest_path)
|
||||
operation = "pushed to Android device"
|
||||
|
||||
# Check if transfer was cancelled
|
||||
if transfer_id != self.current_transfer_id:
|
||||
return
|
||||
|
||||
# Call completion callback on main thread
|
||||
if completion_callback:
|
||||
self.parent.after(0, lambda: completion_callback(success, stats, operation))
|
||||
|
||||
except Exception as e:
|
||||
# Handle errors on main thread
|
||||
if transfer_id == self.current_transfer_id and 'show_error' in self.ui_callbacks:
|
||||
self.parent.after(0, lambda: self.ui_callbacks["show_error"](f"Transfer error: {str(e)}"))
|
||||
|
||||
def cancel_transfer(self) -> bool:
|
||||
"""Cancel the current transfer operation.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user