125 lines
4.2 KiB
Python
125 lines
4.2 KiB
Python
# core/file_transfer.py
|
|
"""File transfer with progress tracking."""
|
|
|
|
import os
|
|
import shutil
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from core.logger_helper import setup_logger
|
|
|
|
logger = setup_logger(Path(__file__).parent.parent / "logs")
|
|
|
|
|
|
def copy_with_progress(src: Path, dst: Path, display_name: str = None) -> None:
|
|
"""
|
|
Copy a file with real-time progress display.
|
|
|
|
Shows:
|
|
- Current MB transferred / total MB
|
|
- Percentage complete
|
|
- Transfer speed (MB/s)
|
|
- Estimated time remaining
|
|
|
|
Args:
|
|
src: Source file path
|
|
dst: Destination file path
|
|
display_name: Optional custom name to display (defaults to src.name)
|
|
"""
|
|
if not src.exists():
|
|
raise FileNotFoundError(f"Source file not found: {src}")
|
|
|
|
display_name = display_name or src.name
|
|
total_size = src.stat().st_size
|
|
total_mb = total_size / (1024 * 1024)
|
|
|
|
# Track progress
|
|
bytes_copied = 0
|
|
start_time = time.time()
|
|
last_update = start_time
|
|
|
|
def progress_callback(chunk_size: int) -> None:
|
|
"""Called after each chunk is copied."""
|
|
nonlocal bytes_copied, last_update
|
|
bytes_copied += chunk_size
|
|
current_time = time.time()
|
|
elapsed = current_time - start_time
|
|
|
|
# Update display every 0.5 seconds to avoid flicker
|
|
if current_time - last_update >= 0.5 or bytes_copied == total_size:
|
|
copied_mb = bytes_copied / (1024 * 1024)
|
|
percent = (bytes_copied / total_size * 100) if total_size > 0 else 0
|
|
|
|
# Calculate transfer speed and ETA
|
|
if elapsed > 0:
|
|
speed_mb_s = copied_mb / elapsed
|
|
remaining_mb = total_mb - copied_mb
|
|
remaining_seconds = remaining_mb / speed_mb_s if speed_mb_s > 0 else 0
|
|
remaining_str = _format_time(remaining_seconds)
|
|
speed_str = f"{speed_mb_s:.1f} MB/s"
|
|
else:
|
|
remaining_str = "calculating..."
|
|
speed_str = "calculating..."
|
|
|
|
# Build progress bar
|
|
bar_width = 25
|
|
filled = int(bar_width * percent / 100)
|
|
bar = "█" * filled + "░" * (bar_width - filled)
|
|
|
|
# Print progress (carriage return to overwrite previous line)
|
|
print(
|
|
f"\r 📊 {display_name}: [{bar}] {copied_mb:.1f}/{total_mb:.1f} MB "
|
|
f"({percent:.1f}%) @ {speed_str} ETA: {remaining_str}",
|
|
end="", flush=True
|
|
)
|
|
|
|
last_update = current_time
|
|
|
|
try:
|
|
# Use shutil.copyfileobj with callback for progress tracking
|
|
with open(src, 'rb') as src_file:
|
|
with open(dst, 'wb') as dst_file:
|
|
# Copy in 10MB chunks for better progress updates
|
|
chunk_size = 10 * 1024 * 1024
|
|
while True:
|
|
chunk = src_file.read(chunk_size)
|
|
if not chunk:
|
|
break
|
|
dst_file.write(chunk)
|
|
progress_callback(len(chunk))
|
|
|
|
# Finalize display
|
|
print() # Newline after progress bar
|
|
elapsed = time.time() - start_time
|
|
avg_speed = total_mb / elapsed if elapsed > 0 else 0
|
|
logger.info(f"Copied {display_name}: {total_mb:.2f} MB in {_format_time(elapsed)} @ {avg_speed:.1f} MB/s")
|
|
|
|
# Copy file metadata (timestamps, permissions)
|
|
shutil.copystat(src, dst)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to copy {display_name}: {e}")
|
|
# Clean up partial destination
|
|
if Path(dst).exists():
|
|
try:
|
|
Path(dst).unlink()
|
|
except:
|
|
pass
|
|
raise
|
|
|
|
|
|
def _format_time(seconds: float) -> str:
|
|
"""Format seconds into human-readable time string."""
|
|
if seconds < 0:
|
|
return "0s"
|
|
if seconds < 60:
|
|
return f"{int(seconds)}s"
|
|
elif seconds < 3600:
|
|
minutes = int(seconds / 60)
|
|
secs = int(seconds % 60)
|
|
return f"{minutes}m {secs}s"
|
|
else:
|
|
hours = int(seconds / 3600)
|
|
minutes = int((seconds % 3600) / 60)
|
|
return f"{hours}h {minutes}m"
|