reorganized file and folder structure to reflect responsibilities
This commit is contained in:
Jason Ross
@@ -0,0 +1,4 @@
|
||||
"""
|
||||
Utils Package
|
||||
Utility functions and helper modules.
|
||||
"""
|
||||
@@ -0,0 +1,300 @@
|
||||
"""
|
||||
File Deduplication Module
|
||||
Handles hash-based file comparison and duplicate detection for file transfers.
|
||||
"""
|
||||
|
||||
import os
|
||||
import hashlib
|
||||
from typing import Optional, Dict, List, Tuple, Callable
|
||||
|
||||
|
||||
class FileDeduplicator:
|
||||
"""Handles file hash computation and duplicate detection."""
|
||||
|
||||
def __init__(self, status_callback: Optional[Callable[[str], None]] = None,
|
||||
progress_callback: Optional[Callable[[int], None]] = None):
|
||||
"""Initialize the deduplicator with optional callbacks.
|
||||
|
||||
Args:
|
||||
status_callback: Function to call with status updates
|
||||
progress_callback: Function to call with progress updates (0-100)
|
||||
"""
|
||||
self.status_callback = status_callback
|
||||
self.progress_callback = progress_callback
|
||||
|
||||
def _update_status(self, message: str) -> None:
|
||||
"""Update status if callback is available."""
|
||||
if self.status_callback:
|
||||
self.status_callback(message)
|
||||
|
||||
def _update_progress(self, percentage: int) -> None:
|
||||
"""Update progress if callback is available."""
|
||||
if self.progress_callback:
|
||||
self.progress_callback(percentage)
|
||||
|
||||
def compute_local_file_hash(self, file_path: str, algorithm: str = 'sha256') -> Optional[str]:
|
||||
"""Compute hash of a local file.
|
||||
|
||||
Args:
|
||||
file_path: Path to the local file
|
||||
algorithm: Hash algorithm to use ('md5', 'sha1', 'sha256')
|
||||
|
||||
Returns:
|
||||
Hex digest of the file hash, or None if error
|
||||
"""
|
||||
try:
|
||||
if not os.path.isfile(file_path):
|
||||
return None
|
||||
|
||||
hash_obj = hashlib.new(algorithm)
|
||||
with open(file_path, 'rb') as file_handle:
|
||||
# Read in chunks to handle large files efficiently
|
||||
for chunk in iter(lambda: file_handle.read(8192), b""):
|
||||
hash_obj.update(chunk)
|
||||
return hash_obj.hexdigest()
|
||||
except Exception as exception:
|
||||
self._update_status(f"Error computing hash for {file_path}: {exception}")
|
||||
return None
|
||||
|
||||
def compute_remote_file_hash(self, remote_path: str, algorithm: str = 'sha256',
|
||||
adb_command_runner: Optional[Callable] = None) -> Optional[str]:
|
||||
"""Compute hash of a file on the Android device using ADB shell.
|
||||
|
||||
Args:
|
||||
remote_path: Path to the file on the Android device
|
||||
algorithm: Hash algorithm to use ('md5', 'sha1', 'sha256')
|
||||
adb_command_runner: Function that runs ADB commands and returns (stdout, stderr, returncode)
|
||||
|
||||
Returns:
|
||||
Hex digest of the file hash, or None if error
|
||||
"""
|
||||
if not adb_command_runner:
|
||||
self._update_status("No ADB command runner provided")
|
||||
return None
|
||||
|
||||
try:
|
||||
# Map algorithm names to Android shell commands
|
||||
hash_commands = {
|
||||
'md5': 'md5sum',
|
||||
'sha1': 'sha1sum',
|
||||
'sha256': 'sha256sum'
|
||||
}
|
||||
|
||||
if algorithm not in hash_commands:
|
||||
self._update_status(f"Unsupported hash algorithm: {algorithm}")
|
||||
return None
|
||||
|
||||
cmd = ['shell', hash_commands[algorithm], remote_path]
|
||||
stdout, stderr, returncode = adb_command_runner(cmd, capture_output=True)
|
||||
|
||||
if returncode != 0 or not stdout:
|
||||
self._update_status(f"Failed to compute remote hash: {stderr}")
|
||||
return None
|
||||
|
||||
# Parse output: "hash_value filename"
|
||||
hash_value = stdout.split()[0] if stdout else None
|
||||
return hash_value.lower() if hash_value else None
|
||||
|
||||
except Exception as exception:
|
||||
self._update_status(f"Error computing remote hash for {remote_path}: {exception}")
|
||||
return None
|
||||
|
||||
def check_files_identical(self, local_path: str, remote_path: str,
|
||||
adb_command_runner: Optional[Callable] = None,
|
||||
algorithm: str = 'sha256') -> bool:
|
||||
"""Check if local and remote files are identical by comparing hashes.
|
||||
|
||||
Args:
|
||||
local_path: Path to the local file
|
||||
remote_path: Path to the remote file
|
||||
adb_command_runner: Function that runs ADB commands
|
||||
algorithm: Hash algorithm to use
|
||||
|
||||
Returns:
|
||||
True if files are identical, False otherwise
|
||||
"""
|
||||
local_hash = self.compute_local_file_hash(local_path, algorithm)
|
||||
remote_hash = self.compute_remote_file_hash(remote_path, algorithm, adb_command_runner)
|
||||
|
||||
if local_hash is None or remote_hash is None:
|
||||
return False
|
||||
|
||||
return local_hash == remote_hash
|
||||
|
||||
def build_local_file_hash_map(self, file_paths: List[str],
|
||||
algorithm: str = 'sha256') -> Dict[str, str]:
|
||||
"""Build a mapping of local file paths to their hashes.
|
||||
|
||||
Args:
|
||||
file_paths: List of local file paths to hash
|
||||
algorithm: Hash algorithm to use
|
||||
|
||||
Returns:
|
||||
Dictionary mapping file paths to their hash values
|
||||
"""
|
||||
hash_map = {}
|
||||
total_files = len(file_paths)
|
||||
|
||||
for index, file_path in enumerate(file_paths):
|
||||
file_hash = self.compute_local_file_hash(file_path, algorithm)
|
||||
|
||||
if file_hash:
|
||||
hash_map[file_path] = file_hash
|
||||
|
||||
# Update progress
|
||||
if total_files > 0:
|
||||
progress_percentage = int((index + 1) * 100 / total_files)
|
||||
self._update_progress(progress_percentage)
|
||||
self._update_status(f"Computing local hashes... {index + 1}/{total_files}")
|
||||
|
||||
return hash_map
|
||||
|
||||
def build_remote_file_hash_map(self, file_paths: List[str],
|
||||
adb_command_runner: Optional[Callable] = None,
|
||||
algorithm: str = 'sha256') -> Dict[str, str]:
|
||||
"""Build a mapping of remote file paths to their hashes.
|
||||
|
||||
Args:
|
||||
file_paths: List of remote file paths to hash
|
||||
adb_command_runner: Function that runs ADB commands
|
||||
algorithm: Hash algorithm to use
|
||||
|
||||
Returns:
|
||||
Dictionary mapping file paths to their hash values
|
||||
"""
|
||||
if not adb_command_runner:
|
||||
self._update_status("No ADB command runner provided")
|
||||
return {}
|
||||
|
||||
hash_map = {}
|
||||
total_files = len(file_paths)
|
||||
|
||||
for index, file_path in enumerate(file_paths):
|
||||
file_hash = self.compute_remote_file_hash(file_path, algorithm, adb_command_runner)
|
||||
|
||||
if file_hash:
|
||||
hash_map[file_path] = file_hash
|
||||
|
||||
# Update progress
|
||||
if total_files > 0:
|
||||
progress_percentage = int((index + 1) * 100 / total_files)
|
||||
self._update_progress(progress_percentage)
|
||||
self._update_status(f"Computing remote hashes... {index + 1}/{total_files}")
|
||||
|
||||
return hash_map
|
||||
|
||||
def find_duplicate_files(self, source_files: List[str], target_files: List[str],
|
||||
is_remote_source: bool = False, is_remote_target: bool = False,
|
||||
adb_command_runner: Optional[Callable] = None,
|
||||
algorithm: str = 'sha256') -> Tuple[List[str], List[str]]:
|
||||
"""Find files that are duplicates between source and target lists.
|
||||
|
||||
Args:
|
||||
source_files: List of source file paths
|
||||
target_files: List of target file paths
|
||||
is_remote_source: True if source files are on Android device
|
||||
is_remote_target: True if target files are on Android device
|
||||
adb_command_runner: Function that runs ADB commands
|
||||
algorithm: Hash algorithm to use
|
||||
|
||||
Returns:
|
||||
Tuple of (files_to_transfer, duplicate_files)
|
||||
"""
|
||||
self._update_status("Building hash maps for duplicate detection...")
|
||||
|
||||
# Build hash maps for both source and target
|
||||
if is_remote_source:
|
||||
source_hashes = self.build_remote_file_hash_map(source_files, adb_command_runner, algorithm)
|
||||
else:
|
||||
source_hashes = self.build_local_file_hash_map(source_files, algorithm)
|
||||
|
||||
if is_remote_target:
|
||||
target_hashes = self.build_remote_file_hash_map(target_files, adb_command_runner, algorithm)
|
||||
else:
|
||||
target_hashes = self.build_local_file_hash_map(target_files, algorithm)
|
||||
|
||||
# Find duplicates by comparing hashes
|
||||
target_hash_values = set(target_hashes.values())
|
||||
files_to_transfer = []
|
||||
duplicate_files = []
|
||||
|
||||
for source_file in source_files:
|
||||
source_hash = source_hashes.get(source_file)
|
||||
if source_hash and source_hash in target_hash_values:
|
||||
duplicate_files.append(source_file)
|
||||
else:
|
||||
files_to_transfer.append(source_file)
|
||||
|
||||
self._update_status(f"Found {len(duplicate_files)} duplicates, {len(files_to_transfer)} files to transfer")
|
||||
|
||||
return files_to_transfer, duplicate_files
|
||||
|
||||
def get_file_size(self, file_path: str, is_remote: bool = False,
|
||||
adb_command_runner: Optional[Callable] = None) -> Optional[int]:
|
||||
"""Get the size of a file in bytes.
|
||||
|
||||
Args:
|
||||
file_path: Path to the file
|
||||
is_remote: True if file is on Android device
|
||||
adb_command_runner: Function that runs ADB commands
|
||||
|
||||
Returns:
|
||||
File size in bytes, or None if error
|
||||
"""
|
||||
try:
|
||||
if is_remote and adb_command_runner:
|
||||
cmd = ['shell', 'stat', '-c', '%s', file_path]
|
||||
stdout, stderr, returncode = adb_command_runner(cmd, capture_output=True)
|
||||
|
||||
if returncode == 0 and stdout.strip().isdigit():
|
||||
return int(stdout.strip())
|
||||
else:
|
||||
return None
|
||||
else:
|
||||
if os.path.isfile(file_path):
|
||||
return os.path.getsize(file_path)
|
||||
else:
|
||||
return None
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def calculate_transfer_savings(self, duplicate_files: List[str], is_remote: bool = False,
|
||||
adb_command_runner: Optional[Callable] = None) -> Tuple[int, int]:
|
||||
"""Calculate the number of bytes and files that would be saved by skipping duplicates.
|
||||
|
||||
Args:
|
||||
duplicate_files: List of duplicate file paths
|
||||
is_remote: True if files are on Android device
|
||||
adb_command_runner: Function that runs ADB commands
|
||||
|
||||
Returns:
|
||||
Tuple of (bytes_saved, files_saved)
|
||||
"""
|
||||
bytes_saved = 0
|
||||
files_saved = 0
|
||||
|
||||
for file_path in duplicate_files:
|
||||
file_size = self.get_file_size(file_path, is_remote, adb_command_runner)
|
||||
if file_size is not None:
|
||||
bytes_saved += file_size
|
||||
files_saved += 1
|
||||
|
||||
return bytes_saved, files_saved
|
||||
|
||||
def format_bytes(self, bytes_value: int) -> str:
|
||||
"""Format bytes into human-readable string.
|
||||
|
||||
Args:
|
||||
bytes_value: Number of bytes
|
||||
|
||||
Returns:
|
||||
Formatted string (e.g., "1.5 MB")
|
||||
"""
|
||||
if bytes_value < 1024:
|
||||
return f"{bytes_value} B"
|
||||
elif bytes_value < 1024 * 1024:
|
||||
return f"{bytes_value / 1024:.1f} KB"
|
||||
elif bytes_value < 1024 * 1024 * 1024:
|
||||
return f"{bytes_value / (1024 * 1024):.1f} MB"
|
||||
else:
|
||||
return f"{bytes_value / (1024 * 1024 * 1024):.1f} GB"
|
||||
Reference in New Issue
Block a user