import os import logging from typing import List, Dict, Optional from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLabel, QProgressBar, QComboBox, QListWidget, QListWidgetItem, QMessageBox, QFileDialog, ) from PyQt6.QtCore import Qt, pyqtSignal, QUrl from PyQt6.QtGui import QDragEnterEvent, QDragMoveEvent, QDragLeaveEvent, QDropEvent from ipod_device import IPodDevice from device_monitor import DeviceMonitor from worker import WorkerThread logger = logging.getLogger(__name__) class iPodTab(QWidget): device_mounted = pyqtSignal(str) device_unmounted = pyqtSignal() tracks_dropped_for_transfer = pyqtSignal(list) def __init__(self, parent=None): super().__init__(parent) self.ipod_devices: List[Dict] = [] self.current_mount_point: Optional[str] = None self.worker_thread: Optional[WorkerThread] = None self._drag_hover_active = False self.setAcceptDrops(True) self._setup_ui() def _setup_ui(self): layout = QVBoxLayout(self) device_layout = QHBoxLayout() device_label = QLabel("iPod Device:") self.device_combo = QComboBox() self.device_combo.setPlaceholderText("No devices found") self.device_combo.currentIndexChanged.connect(self._on_device_combo_changed) refresh_button = QPushButton("Refresh") refresh_button.clicked.connect(self._on_refresh_devices_clicked) device_layout.addWidget(device_label) device_layout.addWidget(self.device_combo) device_layout.addWidget(refresh_button) layout.addLayout(device_layout) mount_layout = QHBoxLayout() self.mount_button = QPushButton("Mount") self.mount_button.setEnabled(False) self.mount_button.clicked.connect(self._on_mount_clicked) self.eject_button = QPushButton("Eject") self.eject_button.setEnabled(False) self.eject_button.clicked.connect(self._on_eject_clicked) mount_layout.addWidget(self.mount_button) mount_layout.addWidget(self.eject_button) mount_layout.addStretch() layout.addLayout(mount_layout) self.device_info_label = QLabel("No device selected") layout.addWidget(self.device_info_label) self.device_status_label = QLabel("Status: Not connected") layout.addWidget(self.device_status_label) self.track_count_label = QLabel("Tracks on device: 0") layout.addWidget(self.track_count_label) on_device_label = QLabel("On Device:") layout.addWidget(on_device_label) self.transferred_list = QListWidget() self.transferred_list.setSelectionMode(QListWidget.SelectionMode.ExtendedSelection) layout.addWidget(self.transferred_list) self._drop_label = QLabel("\u2B07 Drop tracks here to transfer to iPod") self._drop_label.setAlignment(Qt.AlignmentFlag.AlignCenter) self._drop_label.setFixedHeight(48) hl = self.palette().color(self.palette().ColorRole.Highlight) self._drop_label.setStyleSheet( f"background: {hl.name()}20; border: 2px dashed {hl.name()}; " f"border-radius: 8px; font-weight: bold; color: {hl.name()};" ) self._drop_label.setVisible(False) layout.addWidget(self._drop_label) remove_button = QPushButton("Remove Selected from iPod") remove_button.clicked.connect(self._on_remove_tracks_clicked) layout.addWidget(remove_button) orphan_button = QPushButton("Scan for Orphaned Files") orphan_button.clicked.connect(self._on_scan_orphans_clicked) layout.addWidget(orphan_button) dedup_button = QPushButton("Remove Duplicate Tracks") dedup_button.clicked.connect(self._on_remove_duplicates_clicked) layout.addWidget(dedup_button) self.export_button = QPushButton("\u2B07 Download to Computer...") self.export_button.setEnabled(False) self.export_button.clicked.connect(self._on_export_ipod_clicked) layout.addWidget(self.export_button) self.export_progress = QProgressBar() self.export_progress.setRange(0, 100) self.export_progress.setVisible(False) layout.addWidget(self.export_progress) self.export_status = QLabel("") self.export_status.setVisible(False) layout.addWidget(self.export_status) self.delete_progress = QProgressBar() self.delete_progress.setRange(0, 100) self.delete_progress.setVisible(False) layout.addWidget(self.delete_progress) self.delete_status = QLabel("") self.delete_status.setVisible(False) layout.addWidget(self.delete_status) def dragEnterEvent(self, event: QDragEnterEvent): if event.mimeData().hasUrls(): self._drag_hover_active = True self._drop_label.setVisible(True) event.acceptProposedAction() else: super().dragEnterEvent(event) def dragMoveEvent(self, event: QDragMoveEvent): if self._drag_hover_active and event.mimeData().hasUrls(): event.acceptProposedAction() else: super().dragMoveEvent(event) def dragLeaveEvent(self, event: QDragLeaveEvent): self._drag_hover_active = False self._drop_label.setVisible(False) super().dragLeaveEvent(event) def dropEvent(self, event: QDropEvent): self._drag_hover_active = False self._drop_label.setVisible(False) if event.mimeData().hasUrls(): paths = [url.toLocalFile() for url in event.mimeData().urls()] paths = [p for p in paths if os.path.exists(p)] if paths: if not self.current_mount_point: QMessageBox.warning( self, "iPod Not Mounted", "Mount the iPod first before transferring tracks." ) event.ignore() return self.tracks_dropped_for_transfer.emit(paths) event.acceptProposedAction() else: event.ignore() else: super().dropEvent(event) def _on_refresh_devices_clicked(self): if self._is_worker_running(): return self.device_combo.clear() self.device_info_label.setText("Detecting devices...") self.worker_thread = WorkerThread(task_type="detect_devices") self.worker_thread.progress_signal.connect(self._on_device_detection_progress) self.worker_thread.finished_signal.connect(self._on_device_detection_finished) self.worker_thread.start() def _on_device_detection_progress(self, progress, status): self.device_info_label.setText(status) def _on_device_detection_finished(self, success, message, result): self._cleanup_worker() if success: self.ipod_devices = result self.device_combo.blockSignals(True) self.device_combo.clear() for device in self.ipod_devices: mounted = device.get("mounted", False) mount_text = " [mounted]" if mounted else " [not mounted]" self.device_combo.addItem(device.get("name", "Unknown Device") + mount_text) idx = self.device_combo.count() - 1 self.device_combo.setItemData(idx, device, Qt.ItemDataRole.UserRole) if self.ipod_devices: self.device_combo.setCurrentIndex(0) self.device_combo.blockSignals(False) device = self.ipod_devices[0] detected_mount = device.get("mount_point") already_mounted = device.get("mounted", False) self._start_mount_and_load(device, detected_mount, already_mounted) else: self.device_combo.blockSignals(False) self._set_device_not_found() else: self._set_device_not_found() def _start_mount_and_load(self, device: dict, mount_point=None, already_mounted=False): self.device_status_label.setText("Mounting and loading tracks...") self.mount_button.setEnabled(False) self.worker_thread = WorkerThread( task_type="mount_and_load", device=device, mount_point=mount_point, already_mounted=already_mounted, ) self.worker_thread.progress_signal.connect(self._on_mount_load_progress) self.worker_thread.finished_signal.connect(self._on_mount_load_finished) self.worker_thread.start() def _on_mount_load_progress(self, progress, status): self.device_info_label.setText(status) def _on_mount_load_finished(self, success, message, result): if success: device = result["device"] mount_point = result["mount_point"] device_info = result["device_info"] tracks = result["tracks"] self.current_mount_point = mount_point device["info"] = device_info idx = self.device_combo.currentIndex() if idx >= 0: name = device.get("name", "Unknown Device") self.device_combo.setItemText(idx, name + " [mounted]") self.device_combo.setItemData(idx, device, Qt.ItemDataRole.UserRole) free_space = device_info.get("free_space", 0) total_space = device_info.get("total_space", 0) self.device_info_label.setText( f"Device: {device['name']}\n" f"Mount Point: {mount_point}\n" f"Free Space: {free_space / 1024**2:.1f} MB / {total_space / 1024**2:.1f} MB" ) self.device_status_label.setText("Status: Mounted and ready") self.mount_button.setEnabled(False) self.eject_button.setEnabled(True) self.device_mounted.emit(mount_point) self.transferred_list.clear() for track in tracks: display = f"{track['artist']} — {track['title']}" if track.get("album"): display += f" ({track['album']})" item = QListWidgetItem(display) item.setData(Qt.ItemDataRole.UserRole, track) self.transferred_list.addItem(item) self.track_count_label.setText(f"Tracks on device: {len(tracks)}") self.export_button.setEnabled(len(tracks) > 0) else: self.device_status_label.setText(f"Status: {message}") self.mount_button.setEnabled(True) if self.ipod_devices: self._set_device_unmounted(self.ipod_devices[0]) else: self._set_device_not_found() self._cleanup_worker() def _set_device_unmounted(self, device: dict): self.current_mount_point = None self.device_info_label.setText( f"Device: {device['name']}\n" f"Click 'Mount' to connect" ) self.device_status_label.setText("Status: Connected but not mounted") self.mount_button.setEnabled(True) self.eject_button.setEnabled(False) self.track_count_label.setText("Tracks on device: 0") self.transferred_list.clear() self.export_button.setEnabled(False) self.export_progress.setVisible(False) self.export_status.setVisible(False) self.device_unmounted.emit() def _set_device_not_found(self): self.current_mount_point = None self.device_info_label.setText("No iPod devices found") self.device_status_label.setText("Status: Not connected") self.device_combo.clear() self.mount_button.setEnabled(False) self.eject_button.setEnabled(False) self.track_count_label.setText("Tracks on device: 0") self.transferred_list.clear() self.export_button.setEnabled(False) self.export_progress.setVisible(False) self.export_status.setVisible(False) self.device_unmounted.emit() def _on_mount_clicked(self): idx = self.device_combo.currentIndex() if idx < 0: return device = self.device_combo.itemData(idx, Qt.ItemDataRole.UserRole) if not device: return detected_mount = device.get("mount_point") self._start_mount_and_load(device, detected_mount, already_mounted=False) def _on_eject_clicked(self): reply = QMessageBox.question( self, "Eject iPod", "Eject iPod? Make sure no transfers are in progress.", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No ) if reply != QMessageBox.StandardButton.Yes: return self.device_status_label.setText("Ejecting...") self.eject_button.setEnabled(False) try: ipod = IPodDevice(mount_point=self.current_mount_point) success = ipod.unmount_device() if success or True: self.current_mount_point = None if self.ipod_devices: self.ipod_devices[0]["mounted"] = False self.ipod_devices[0]["mount_point"] = None idx = self.device_combo.currentIndex() if idx >= 0: name = self.ipod_devices[0].get("name", "Unknown Device") self.device_combo.setItemText(idx, name + " [not mounted]") self.device_combo.setItemData(idx, self.ipod_devices[0], Qt.ItemDataRole.UserRole) self.device_info_label.setText("iPod ejected. You can now disconnect it.") self.device_status_label.setText("Status: Ejected - safe to disconnect") self.mount_button.setEnabled(False) self.eject_button.setEnabled(False) self.track_count_label.setText("Tracks on device: 0") self.transferred_list.clear() self.device_unmounted.emit() else: self.device_status_label.setText("Status: Eject failed") self.eject_button.setEnabled(True) QMessageBox.warning(self, "Eject Error", "Failed to eject iPod safely.") except Exception as e: self.device_status_label.setText("Status: Eject error") self.eject_button.setEnabled(True) QMessageBox.warning(self, "Eject Error", f"Error ejecting iPod: {e}") def _load_ipod_tracks(self): if not self.current_mount_point: return try: from ipod_nano7_db import Nano7Database db = Nano7Database(self.current_mount_point) tracks = db.get_all_tracks() self.transferred_list.clear() for track in tracks: display = f"{track['artist']} — {track['title']}" if track.get("album"): display += f" ({track['album']})" item = QListWidgetItem(display) item.setData(Qt.ItemDataRole.UserRole, track) self.transferred_list.addItem(item) self.track_count_label.setText(f"Tracks on device: {len(tracks)}") self.export_button.setEnabled(len(tracks) > 0) except Exception as e: logger.error(f"Failed to load iPod tracks: {e}") self.track_count_label.setText("Tracks on device: unknown") def _on_remove_tracks_clicked(self): selected = self.transferred_list.selectedItems() if not selected: QMessageBox.information(self, "Info", "Select tracks to remove") return count = len(selected) reply = QMessageBox.question( self, "Confirm Delete", f"Remove {count} track(s) from iPod?\nThis will permanently delete the files from the device.", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No ) if reply != QMessageBox.StandardButton.Yes: return if self._is_worker_running(): QMessageBox.information(self, "Info", "Please wait for the current task to finish") return pids = [] for item in selected: track_data = item.data(Qt.ItemDataRole.UserRole) if track_data and "pid" in track_data: pids.append(track_data["pid"]) if not pids: QMessageBox.warning(self, "Error", "No valid tracks selected") return self._set_delete_buttons_enabled(False) self.delete_progress.setVisible(True) self.delete_progress.setValue(0) self.delete_status.setVisible(True) self.delete_status.setText(f"Removing {len(pids)} track(s)...") self.worker_thread = WorkerThread( task_type="delete_tracks", pids=pids, mount_point=self.current_mount_point, ) self.worker_thread.progress_signal.connect(self._on_delete_progress) self.worker_thread.finished_signal.connect(self._on_delete_finished) self.worker_thread.start() def _set_delete_buttons_enabled(self, enabled: bool): for w in self.findChildren(QPushButton): if w.text() in ("Remove Selected from iPod", "Remove Duplicate Tracks", "Scan for Orphaned Files", "\u2B07 Download to Computer..."): w.setEnabled(enabled) def _on_delete_progress(self, progress, status): self.delete_progress.setValue(progress) self.delete_status.setText(status) def _on_delete_finished(self, success, message, result): self._set_delete_buttons_enabled(True) if success: self.delete_progress.setValue(100) self._load_ipod_tracks() QMessageBox.information(self, "Done" if success else "Error", message) self._cleanup_worker() def _on_export_ipod_clicked(self): if not self.current_mount_point: QMessageBox.warning(self, "Error", "No iPod device mounted.") return selected = self.transferred_list.selectedItems() if not selected: QMessageBox.information(self, "Info", "Select tracks to download") return dest_dir = QFileDialog.getExistingDirectory(self, "Select Download Destination") if not dest_dir: return tracks = [] for item in selected: data = item.data(Qt.ItemDataRole.UserRole) if data and data.get("file_path") and os.path.exists(data["file_path"]): tracks.append(data) if not tracks: QMessageBox.warning(self, "Error", "No valid track files found on device") return if self._is_worker_running(): QMessageBox.information(self, "Info", "Please wait for the current task to finish") return self.export_button.setEnabled(False) self.export_progress.setVisible(True) self.export_progress.setValue(0) self.export_status.setVisible(True) self.export_status.setText(f"Exporting {len(tracks)} track(s)...") self.worker_thread = WorkerThread( task_type="export_ipod", tracks=tracks, dest_dir=dest_dir, mount_point=self.current_mount_point, ) self.worker_thread.progress_signal.connect(self._on_export_ipod_progress) self.worker_thread.finished_signal.connect(self._on_export_ipod_finished) self.worker_thread.start() def _on_export_ipod_progress(self, progress, status): self.export_progress.setValue(progress) self.export_status.setText(status) def _on_export_ipod_finished(self, success, message, result): self.export_button.setEnabled(True) self.export_progress.setValue(100 if success else 0) if success: self.export_status.setText(message) QMessageBox.information(self, "Export Complete", message) else: self.export_status.setText(f"Error: {message}") QMessageBox.warning(self, "Export Error", message) self._cleanup_worker() def _on_scan_orphans_clicked(self): if not self.current_mount_point: QMessageBox.warning(self, "Error", "No iPod device mounted.") return try: from ipod_nano7_db import Nano7Database db = Nano7Database(self.current_mount_point) orphans = db.get_orphaned_files() if not orphans: QMessageBox.information(self, "Scan Complete", "No orphaned files found.") return total_size = sum(o["size"] for o in orphans) total_mb = total_size / (1024 * 1024) details = f"Found {len(orphans)} orphaned file(s) occupying {total_mb:.1f} MB:\n\n" for o in orphans[:20]: size_kb = o["size"] / 1024 details += f" {o['artist']} — {o['title']} ({size_kb:.0f} KB)\n" if len(orphans) > 20: details += f" ... and {len(orphans) - 20} more\n" reply = QMessageBox.question( self, "Orphaned Files Found", f"{details}\nDelete these files to free up space on iPod?", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No ) if reply == QMessageBox.StandardButton.Yes: deleted = db.delete_orphaned_files() QMessageBox.information(self, "Cleanup Complete", f"Deleted {deleted} orphaned file(s).") self._load_ipod_tracks() except Exception as e: logger.error(f"Failed to scan for orphaned files: {e}") QMessageBox.warning(self, "Error", f"Failed to scan: {e}") def _on_remove_duplicates_clicked(self): if not self.current_mount_point: QMessageBox.warning(self, "Error", "No iPod device mounted.") return if self._is_worker_running(): QMessageBox.information(self, "Info", "Please wait for the current task to finish") return self._set_delete_buttons_enabled(False) self.delete_progress.setVisible(True) self.delete_progress.setValue(0) self.delete_status.setVisible(True) self.delete_status.setText("Scanning for duplicates...") self.worker_thread = WorkerThread( task_type="remove_duplicates", mount_point=self.current_mount_point, ) self.worker_thread.progress_signal.connect(self._on_delete_progress) self.worker_thread.finished_signal.connect(self._on_delete_duplicates_finished) self.worker_thread.start() def _on_delete_duplicates_finished(self, success, message, result): self._set_delete_buttons_enabled(True) self._load_ipod_tracks() if success: self.delete_progress.setValue(100) QMessageBox.information(self, "Duplicates" if success else "Error", message) self._cleanup_worker() def start_monitoring(self, monitor: DeviceMonitor): monitor.state_changed.connect(self._on_auto_state_changed) def stop_monitoring(self, monitor: DeviceMonitor): try: monitor.state_changed.disconnect(self._on_auto_state_changed) except TypeError: pass def _on_auto_state_changed(self, devices): if self._is_worker_running(): return self._on_device_detection_finished(True, f"Found {len(devices)} devices", devices) def _on_device_combo_changed(self, index): device = self.device_combo.itemData(index, Qt.ItemDataRole.UserRole) if not device: return mp = device.get("mount_point") if mp and mp == self.current_mount_point: return if self._is_worker_running(): return self._start_mount_and_load( device, mount_point=mp, already_mounted=device.get("mounted", False), ) def _cleanup_worker(self): if self.worker_thread: self.worker_thread.quit() self.worker_thread.wait() self.worker_thread = None def _is_worker_running(self): if self.worker_thread: try: return self.worker_thread.isRunning() except RuntimeError: self.worker_thread = None return False