en2zmax/neo-pod-desktop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
neo-pod-desktop/src/main.py
Maksim Totmin 489d99c068 Add cover art pipeline (extract/encode/cache) and iTunes-style player 
- New src/artwork/ module with iPod Nano 7G cover art support
  - codecs: RGB565/RGB555/UYVY/JPEG encode/decode via numpy
  - presets: 50+ iPod artwork format definitions
  - chunks: ArtworkDB binary parser/writer (MHFD→MHSD→MHLI→MHII→MHNI)
  - writer: ithmb file generation + ArtworkDB assembly
  - extractor: cover extraction from audio files (APIC/covr/FLAC pictures)
  - cache: local JPEG cache for player UI
- iTunes-style playback bar with 60x60 cover art display
- Scan library now caches cover art by artist+album
- iPod Nano7 transfer writes artwork to ArtworkDB + ithmb
- FFmpeg conversion preserves embedded cover art (-map 0:t?)
- One-time reembed_covers.py script in /tmp
- MetadataHandler process_file() fallback to cover_path
- Bug fixes: missing read_existing_artwork import, signed PID in Q format
2026-05-31 17:45:01 +07:00

1711 lines
66 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
Main Application Module for neo-pod-desktop
Provides a GUI for downloading, converting, managing and transferring music to iPod Nano devices
"""
import os
import sys
import logging
from typing import List, Dict, Optional, Tuple
from pathlib import Path
from PyQt6.QtWidgets import (
QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
QPushButton, QLabel, QLineEdit, QProgressBar, QFileDialog,
QComboBox, QCheckBox, QTabWidget, QListWidget, QListWidgetItem,
QMessageBox, QSpinBox, QGroupBox, QHeaderView,
QTableWidget, QTableWidgetItem, QSlider, QMenu
)
from PyQt6.QtCore import Qt, QThread, pyqtSignal, QUrl
from PyQt6.QtGui import QPixmap
from PyQt6.QtMultimedia import QMediaPlayer, QAudioOutput
from youtube_downloader import YouTubeDownloader, TrackInfo
from audio_converter import AudioConverter
from metadata_handler import MetadataHandler
from ipod_device import IPodDevice
from artwork.cache import cover_cache
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
AUDIO_EXTS = {'.m4a', '.mp3', '.aac'}
SOURCE_EXTS = {'.flac', '.wav', '.ogg', '.opus', '.wma', '.ape', '.aiff', '.alac'}
class WorkerThread(QThread):
"""Worker thread for background tasks"""
progress_signal = pyqtSignal(int, str)
finished_signal = pyqtSignal(bool, str, object)
def __init__(self, task_type: str, **kwargs):
super().__init__()
self.task_type = task_type
self.kwargs = kwargs
self.is_running = True
def run(self):
try:
handlers = {
"download": self._run_download,
"convert": self._run_convert,
"transfer": self._run_transfer,
"detect_devices": self._run_detect_devices,
"export_ipod": self._run_export_ipod,
}
handler = handlers.get(self.task_type)
if handler:
handler()
else:
raise ValueError(f"Unknown task type: {self.task_type}")
except Exception as e:
logger.exception(f"Error in worker thread: {e}")
self.finished_signal.emit(False, str(e), None)
def _run_download(self):
url = self.kwargs.get("url")
output_dir = self.kwargs.get("output_dir", "downloads")
fmt = self.kwargs.get("format", "m4a")
quality = self.kwargs.get("quality", 256)
self.progress_signal.emit(0, f"Initializing download from {url}")
downloader = YouTubeDownloader(
output_dir=output_dir,
progress_callback=self._download_progress_callback
)
tracks = downloader.process_url(url, fmt, quality)
if not tracks:
self.finished_signal.emit(False, "No tracks were successfully downloaded", [])
return
self.finished_signal.emit(True, f"Downloaded {len(tracks)} tracks", tracks)
def _download_progress_callback(self, progress: int, status: str):
self.progress_signal.emit(progress, status)
def _run_convert(self):
tracks = self.kwargs.get("tracks", [])
local_files = self.kwargs.get("local_files", [])
output_dir = self.kwargs.get("output_dir", "converted")
fmt = self.kwargs.get("format", "m4a")
quality = self.kwargs.get("quality", 256)
valid_tracks = [
t for t in tracks
if hasattr(t, 'download_path') and t.download_path and os.path.exists(t.download_path)
]
metadata_handler = MetadataHandler()
for f in local_files:
if os.path.exists(f):
track = metadata_handler.extract_tags(f)
if track:
track.download_path = f
valid_tracks.append(track)
if not valid_tracks:
self.finished_signal.emit(False, "No valid tracks to convert", None)
return
self.progress_signal.emit(0, f"Initializing conversion of {len(valid_tracks)} tracks")
converter = AudioConverter(output_dir=output_dir)
metadata_handler = MetadataHandler()
converted_tracks = []
for i, track in enumerate(valid_tracks):
if not self.is_running:
break
progress = int(((i + 1) / len(valid_tracks)) * 100)
self.progress_signal.emit(progress, f"Converting {track.title}...")
try:
output_path = converter.convert_to_ipod_format(
track, track.download_path, fmt, quality
)
if output_path:
metadata_handler.process_file(output_path, track)
converted_tracks.append((track, output_path))
except Exception as e:
logger.error(f"Error converting track {track.title}: {e}")
continue
if not converted_tracks:
self.finished_signal.emit(False, "Failed to convert any tracks", None)
return
self.finished_signal.emit(True, f"Converted {len(converted_tracks)} tracks", converted_tracks)
def _run_transfer(self):
tracks = self.kwargs.get("tracks", [])
mount_point = self.kwargs.get("mount_point")
if not tracks:
self.finished_signal.emit(False, "No tracks to transfer", None)
return
if not mount_point:
self.finished_signal.emit(False, "No iPod device mounted", None)
return
self.progress_signal.emit(0, f"Initializing transfer of {len(tracks)} tracks")
ipod = IPodDevice(mount_point=mount_point)
device_info = ipod.get_device_info()
free_space = device_info.get("free_space", 0)
total_size = sum(os.path.getsize(path) for _, path in tracks if os.path.exists(path))
if total_size > free_space:
self.finished_signal.emit(
False,
f"Not enough space on iPod. Need {total_size / 1024**2:.1f} MB, "
f"but only {free_space / 1024**2:.1f} MB available",
None
)
return
transferred_tracks = []
for i, (track, path) in enumerate(tracks):
if not self.is_running:
break
progress = int(((i + 1) / len(tracks)) * 100)
self.progress_signal.emit(progress, f"Transferring {track.title}...")
if os.path.exists(path):
try:
success = ipod.transfer_file(path)
if success:
transferred_tracks.append((track, path))
except Exception as e:
logger.error(f"Error transferring track {track.title}: {e}")
continue
self.finished_signal.emit(
True, f"Transferred {len(transferred_tracks)} tracks to iPod", transferred_tracks
)
def _run_detect_devices(self):
self.progress_signal.emit(0, "Detecting iPod devices...")
ipod = IPodDevice()
devices = ipod.detect_devices()
if not devices:
self.finished_signal.emit(False, "No iPod devices found", [])
return
self.finished_signal.emit(True, f"Found {len(devices)} iPod devices", devices)
def _run_export_ipod(self):
tracks = self.kwargs.get("tracks", [])
dest_dir = self.kwargs.get("dest_dir", "")
mount_point = self.kwargs.get("mount_point")
if not tracks or not dest_dir or not mount_point:
self.finished_signal.emit(False, "Missing parameters for export", None)
return
self.progress_signal.emit(0, f"Exporting {len(tracks)} track(s)...")
ipod = IPodDevice(mount_point=mount_point)
def progress(p, s):
self.progress_signal.emit(p, s)
exported = ipod.export_tracks(tracks, dest_dir, progress_callback=progress)
self.finished_signal.emit(True, f"Exported {len(exported)} track(s)", exported)
def stop(self):
self.is_running = False
self.wait()
class MainWindow(QMainWindow):
"""Main application window"""
def __init__(self):
super().__init__()
self.setWindowTitle("neo-pod-desktop")
self.setMinimumSize(800, 600)
self.worker_thread = None
self.downloaded_tracks: List[TrackInfo] = []
self.library_ready: List[Dict] = []
self.library_source_files: List[str] = []
self.library_scanned_sources: List[Dict] = []
self.ipod_devices: List[Dict] = []
self.current_mount_point: Optional[str] = None
self.show_download_tab = False
self.player: Optional[QMediaPlayer] = None
self.audio_output: Optional[QAudioOutput] = None
self.current_playback_index: int = -1
self._is_playing = False
self._setup_ui()
self._setup_player()
self._scan_library()
self._on_refresh_devices_clicked()
def _cleanup_worker(self):
"""Safely clean up the worker thread after it finishes.
Calls wait() to ensure the C++ thread is fully cleaned up before
dropping the Python reference. This prevents the Qt crash:
'QThread: Destroyed while thread is still running'
"""
if self.worker_thread:
self.worker_thread.wait(3000)
self.worker_thread.deleteLater()
self.worker_thread = None
def _setup_ui(self):
main_widget = QWidget()
main_layout = QVBoxLayout(main_widget)
self.tab_widget = QTabWidget()
main_layout.addWidget(self.tab_widget)
download_tab = QWidget()
self.tab_widget.addTab(download_tab, "Download")
self._setup_download_tab(download_tab)
self.download_tab_index = 0
library_tab = QWidget()
self.tab_widget.addTab(library_tab, "Library")
self._setup_library_tab(library_tab)
ipod_tab = QWidget()
self.tab_widget.addTab(ipod_tab, "iPod")
self._setup_ipod_tab(ipod_tab)
settings_tab = QWidget()
self.tab_widget.addTab(settings_tab, "Settings")
self._setup_settings_tab(settings_tab)
self.tab_widget.setTabVisible(self.download_tab_index, self.show_download_tab)
self.statusBar().showMessage("Ready")
self.setCentralWidget(main_widget)
def _setup_download_tab(self, tab):
layout = QVBoxLayout(tab)
url_layout = QHBoxLayout()
url_label = QLabel("YouTube URL:")
self.url_input = QLineEdit()
self.url_input.setPlaceholderText("https://www.youtube.com/watch?v=...")
url_layout.addWidget(url_label)
url_layout.addWidget(self.url_input)
layout.addLayout(url_layout)
format_layout = QHBoxLayout()
format_label = QLabel("Format:")
self.format_combo = QComboBox()
self.format_combo.addItems(["m4a", "mp3"])
quality_label = QLabel("Quality (kbps):")
self.quality_spin = QSpinBox()
self.quality_spin.setRange(128, 320)
self.quality_spin.setValue(256)
self.quality_spin.setSingleStep(32)
format_layout.addWidget(format_label)
format_layout.addWidget(self.format_combo)
format_layout.addWidget(quality_label)
format_layout.addWidget(self.quality_spin)
format_layout.addStretch()
layout.addLayout(format_layout)
self.download_button = QPushButton("Download")
self.download_button.clicked.connect(self._on_download_clicked)
layout.addWidget(self.download_button)
self.download_progress = QProgressBar()
self.download_progress.setRange(0, 100)
layout.addWidget(self.download_progress)
self.download_status = QLabel("Ready to download")
layout.addWidget(self.download_status)
track_list_label = QLabel("Downloaded Tracks:")
layout.addWidget(track_list_label)
self.track_list = QListWidget()
layout.addWidget(self.track_list)
def _setup_library_tab(self, tab):
layout = QVBoxLayout(tab)
# === Playback controls bar ===
playback_group = QGroupBox()
playback_layout = QVBoxLayout(playback_group)
playback_layout.setContentsMargins(8, 6, 8, 6)
# Top row: transport buttons + progress + time + volume
controls_row = QHBoxLayout()
controls_row.setSpacing(8)
self.prev_btn = QPushButton("\u23EE")
self.prev_btn.setFixedWidth(36)
self.prev_btn.clicked.connect(self._on_prev_clicked)
self.play_btn = QPushButton("\u25B6")
self.play_btn.setFixedWidth(44)
self.play_btn.clicked.connect(self._on_play_pause_clicked)
self.next_btn = QPushButton("\u23ED")
self.next_btn.setFixedWidth(36)
self.next_btn.clicked.connect(self._on_next_clicked)
controls_row.addWidget(self.prev_btn)
controls_row.addWidget(self.play_btn)
controls_row.addWidget(self.next_btn)
self.time_label_start = QLabel("0:00")
self.time_label_start.setFixedWidth(40)
self.time_label_start.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter)
self.position_slider = QSlider(Qt.Orientation.Horizontal)
self.position_slider.setMinimum(0)
self.position_slider.setMaximum(1000)
self.position_slider.setValue(0)
self.position_slider.sliderPressed.connect(self._on_slider_pressed)
self.position_slider.sliderReleased.connect(self._on_slider_released)
self.time_label_end = QLabel("0:00")
self.time_label_end.setFixedWidth(40)
self.time_label_end.setAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter)
controls_row.addWidget(self.time_label_start)
controls_row.addWidget(self.position_slider, stretch=1)
controls_row.addWidget(self.time_label_end)
vol_icon = QLabel("\U0001F50A")
self.volume_slider = QSlider(Qt.Orientation.Horizontal)
self.volume_slider.setFixedWidth(80)
self.volume_slider.setRange(0, 100)
self.volume_slider.setValue(80)
self.volume_slider.valueChanged.connect(self._on_volume_changed)
controls_row.addWidget(vol_icon)
controls_row.addWidget(self.volume_slider)
playback_layout.addLayout(controls_row)
# Bottom row: now playing info with album art
now_playing_layout = QHBoxLayout()
self.cover_label = QLabel()
self.cover_label.setFixedSize(60, 60)
self.cover_label.setScaledContents(True)
self.cover_label.setStyleSheet("background: palette(window); border-radius: 4px;")
now_playing_layout.addWidget(self.cover_label)
self.now_playing_label = QLabel("Select a track to play")
self.now_playing_label.setAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter)
self.now_playing_label.setStyleSheet("color: palette(mid); font-size: 12px;")
now_playing_layout.addWidget(self.now_playing_label, stretch=1)
playback_layout.addLayout(now_playing_layout)
layout.addWidget(playback_group)
self._setup_library_toolbar(layout)
self.library_table = QTableWidget()
self.library_table.setColumnCount(7)
self.library_table.setHorizontalHeaderLabels(
["#", "Artist", "Title", "Album", "Duration", "Genre", "Size"]
)
self.library_table.setSelectionBehavior(QTableWidget.SelectionBehavior.SelectRows)
self.library_table.setSelectionMode(QTableWidget.SelectionMode.ExtendedSelection)
self.library_table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers)
self.library_table.setAlternatingRowColors(True)
self.library_table.setSortingEnabled(True)
self.library_table.cellDoubleClicked.connect(self._on_table_double_clicked)
self.library_table.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
self.library_table.customContextMenuRequested.connect(self._on_library_table_context_menu)
header = self.library_table.horizontalHeader()
header.setSectionResizeMode(1, QHeaderView.ResizeMode.Stretch)
header.setSectionResizeMode(2, QHeaderView.ResizeMode.Stretch)
header.setSectionResizeMode(3, QHeaderView.ResizeMode.Stretch)
header.setSectionResizeMode(0, QHeaderView.ResizeMode.ResizeToContents)
header.setSectionResizeMode(4, QHeaderView.ResizeMode.ResizeToContents)
header.setSectionResizeMode(5, QHeaderView.ResizeMode.ResizeToContents)
header.setSectionResizeMode(6, QHeaderView.ResizeMode.ResizeToContents)
layout.addWidget(self.library_table, stretch=1)
self.library_progress = QProgressBar()
self.library_progress.setRange(0, 100)
layout.addWidget(self.library_progress)
self.library_status = QLabel("Ready")
layout.addWidget(self.library_status)
def _setup_library_toolbar(self, layout):
toolbar = QHBoxLayout()
toolbar.setSpacing(4)
self.library_add_btn = QPushButton("\u2795 Add Files...")
self.library_add_btn.clicked.connect(self._on_library_add_files)
toolbar.addWidget(self.library_add_btn)
self.library_convert_btn = QPushButton("\uD83D\uDD04 Convert")
self.library_convert_btn.setEnabled(False)
self.library_convert_btn.clicked.connect(self._on_library_convert_clicked)
toolbar.addWidget(self.library_convert_btn)
self.library_transfer_btn = QPushButton("\uD83D\uDCE4 Transfer")
self.library_transfer_btn.setEnabled(False)
self.library_transfer_btn.clicked.connect(self._on_library_transfer_clicked)
toolbar.addWidget(self.library_transfer_btn)
self.library_remove_btn = QPushButton("\uD83D\uDDD1\uFE0F Remove")
self.library_remove_btn.setEnabled(False)
self.library_remove_btn.clicked.connect(self._on_library_remove_selected)
toolbar.addWidget(self.library_remove_btn)
refresh_btn = QPushButton("\uD83D\uDD04")
refresh_btn.setToolTip("Refresh Library")
refresh_btn.clicked.connect(self._on_refresh_library_clicked)
toolbar.addWidget(refresh_btn)
toolbar.addStretch()
self.library_search_input = QLineEdit()
self.library_search_input.setPlaceholderText("\uD83D\uDD0D Search Library...")
self.library_search_input.setFixedWidth(220)
self.library_search_input.textChanged.connect(self._on_search_text_changed)
toolbar.addWidget(self.library_search_input)
layout.addLayout(toolbar)
def _setup_ipod_tab(self, tab):
layout = QVBoxLayout(tab)
device_layout = QHBoxLayout()
device_label = QLabel("iPod Device:")
self.device_combo = QComboBox()
self.device_combo.setPlaceholderText("No devices found")
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)
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)
def _setup_settings_tab(self, tab):
layout = QVBoxLayout(tab)
library_group = QGroupBox("Library")
library_layout = QVBoxLayout(library_group)
output_layout = QHBoxLayout()
output_label = QLabel("Output Directory:")
self.output_dir_input = QLineEdit()
self.output_dir_input.setText(os.path.join(os.path.expanduser("~"), "Music", "iPod"))
browse_btn = QPushButton("Browse...")
browse_btn.clicked.connect(self._on_browse_output_clicked)
output_layout.addWidget(output_label)
output_layout.addWidget(self.output_dir_input)
output_layout.addWidget(browse_btn)
library_layout.addLayout(output_layout)
layout.addWidget(library_group)
video_group = QGroupBox("Video Support")
video_layout = QVBoxLayout(video_group)
self.enable_video_check = QCheckBox("Enable Video Download (iPod Nano 5th-7th gen)")
video_layout.addWidget(self.enable_video_check)
video_resolution_layout = QHBoxLayout()
video_resolution_label = QLabel("Video Resolution:")
self.video_resolution_combo = QComboBox()
self.video_resolution_combo.addItems(["640x480", "480x360", "320x240"])
self.video_resolution_combo.setEnabled(False)
self.enable_video_check.toggled.connect(self.video_resolution_combo.setEnabled)
video_resolution_layout.addWidget(video_resolution_label)
video_resolution_layout.addWidget(self.video_resolution_combo)
video_resolution_layout.addStretch()
video_layout.addLayout(video_resolution_layout)
layout.addWidget(video_group)
metadata_group = QGroupBox("Metadata Options")
metadata_layout = QVBoxLayout(metadata_group)
self.embed_artwork_check = QCheckBox("Embed Album Artwork")
self.embed_artwork_check.setChecked(True)
metadata_layout.addWidget(self.embed_artwork_check)
self.use_musicbrainz_check = QCheckBox("Use MusicBrainz for Enhanced Metadata (Experimental)")
metadata_layout.addWidget(self.use_musicbrainz_check)
layout.addWidget(metadata_group)
advanced_group = QGroupBox("Advanced Options")
advanced_layout = QVBoxLayout(advanced_group)
self.clean_temp_check = QCheckBox("Clean Temporary Files After Transfer")
self.clean_temp_check.setChecked(True)
advanced_layout.addWidget(self.clean_temp_check)
self.auto_detect_check = QCheckBox("Auto-Detect iPod on Startup")
self.auto_detect_check.setChecked(True)
advanced_layout.addWidget(self.auto_detect_check)
self.show_download_check = QCheckBox("Show YouTube Download Tab")
self.show_download_check.setChecked(False)
self.show_download_check.toggled.connect(self._on_show_download_toggled)
advanced_layout.addWidget(self.show_download_check)
layout.addWidget(advanced_group)
layout.addStretch()
about_label = QLabel(
"neo-pod-desktop\n"
"Version 1.0.0\n\n"
"Desktop application for downloading, converting, managing and transferring music to iPod Nano devices."
)
about_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
layout.addWidget(about_label)
def _on_show_download_toggled(self, checked: bool):
self.show_download_tab = checked
self.tab_widget.setTabVisible(self.download_tab_index, checked)
def _setup_player(self):
"""Initialize QMediaPlayer and QAudioOutput for playback."""
self.player = QMediaPlayer()
self.audio_output = QAudioOutput()
self.audio_output.setVolume(0.8)
self.player.setAudioOutput(self.audio_output)
self.player.positionChanged.connect(self._on_media_position_changed)
self.player.durationChanged.connect(self._on_media_duration_changed)
self.player.mediaStatusChanged.connect(self._on_media_status_changed)
def _play_track_from_index(self, index: int):
"""Start playing a track from the library table by row index."""
if index < 0 or index >= self.library_table.rowCount():
return
data_item = self.library_table.item(index, 0)
if data_item is None:
return
is_ready, track_data = data_item.data(Qt.ItemDataRole.UserRole)
if not track_data or not os.path.exists(track_data.get("path", "")):
return
self.current_playback_index = index
self.player.setSource(QUrl.fromLocalFile(track_data["path"]))
self.player.play()
self._is_playing = True
self.play_btn.setText("\u23F8")
self.now_playing_label.setText(f"{track_data['artist']} \u2014 {track_data['title']}")
self.now_playing_label.setStyleSheet(
"color: palette(text); font-weight: bold;"
)
cover_path = track_data.get("cover_path")
if cover_path and os.path.exists(cover_path):
pixmap = QPixmap(cover_path)
self.cover_label.setPixmap(pixmap)
self.cover_label.setToolTip(track_data.get("album", ""))
else:
# Fallback: try looking up by artist/album in cover cache
artist = track_data.get("artist", "")
album = track_data.get("album", "")
cached = cover_cache.get(artist, album) if artist else None
if cached and os.path.exists(cached):
pixmap = QPixmap(cached)
self.cover_label.setPixmap(pixmap)
self.cover_label.setToolTip(album)
else:
self.cover_label.clear()
self.cover_label.setToolTip("")
# Highlight the current row
self.library_table.clearSelection()
self.library_table.selectRow(index)
def _on_play_pause_clicked(self):
if self.player is None:
return
if self.player.playbackState() == QMediaPlayer.PlaybackState.PlayingState:
self.player.pause()
self._is_playing = False
self.play_btn.setText("\u25B6")
else:
if self.player.playbackState() == QMediaPlayer.PlaybackState.StoppedState:
# Nothing playing yet — play first selected row or row 0
selected_rows = {item.row() for item in self.library_table.selectedItems()}
start = min(selected_rows) if selected_rows else 0
self._play_track_from_index(start)
else:
self.player.play()
self._is_playing = True
self.play_btn.setText("\u23F8")
def _on_prev_clicked(self):
if self.player is None:
return
if self.player.position() > 3000:
self.player.setPosition(0)
return
idx = max(0, self.current_playback_index - 1)
self._play_track_from_index(idx)
def _on_next_clicked(self):
if self.player is None:
return
idx = min(self.library_table.rowCount() - 1, self.current_playback_index + 1)
self._play_track_from_index(idx)
def _on_slider_pressed(self):
if self.player:
self.player.setPosition(
int(self.position_slider.value() / 1000.0 * self.player.duration())
)
def _on_slider_released(self):
pass
def _on_volume_changed(self, value):
if self.audio_output:
self.audio_output.setVolume(value / 100.0)
def _on_media_position_changed(self, position: int):
if self.position_slider is None or self.player is None:
return
duration = self.player.duration()
if duration > 0 and not self.position_slider.isSliderDown():
self.position_slider.setValue(int(position / duration * 1000))
self.time_label_start.setText(self._format_time_ms(position))
def _on_media_duration_changed(self, duration: int):
self.time_label_end.setText(self._format_time_ms(duration))
if self.position_slider and not self.position_slider.isSliderDown():
self.position_slider.setValue(0)
def _on_media_status_changed(self, status):
if status == QMediaPlayer.MediaStatus.EndOfMedia:
self._on_next_clicked()
elif status == QMediaPlayer.MediaStatus.NoMedia:
self.play_btn.setText("\u25B6")
self._is_playing = False
def _on_table_double_clicked(self, row: int):
self._play_track_from_index(row)
@staticmethod
def _format_time_ms(ms: int) -> str:
total_seconds = ms // 1000
m, s = divmod(total_seconds, 60)
return f"{m}:{s:02d}"
def _format_duration(self, seconds: int) -> str:
if seconds <= 0:
return "??:??"
m, s = divmod(int(seconds), 60)
return f"{m}:{s:02d}"
def _format_size(self, size_bytes: int) -> str:
if size_bytes < 1024 * 1024:
return f"{size_bytes / 1024:.0f} KB"
return f"{size_bytes / (1024 * 1024):.1f} MB"
def _scan_library(self):
"""Scan Output Directory for ready-to-transfer and source files."""
output_dir = self.output_dir_input.text().strip()
if not output_dir or not os.path.isdir(output_dir):
self.library_ready.clear()
self.library_scanned_sources = []
self._refresh_library_ui()
return
ready = []
sources = []
handler = MetadataHandler()
for root, _dirs, files in os.walk(output_dir):
for fname in sorted(files):
fpath = os.path.join(root, fname)
ext = os.path.splitext(fname)[1].lower()
if not os.path.isfile(fpath):
continue
if ext in AUDIO_EXTS:
size = os.path.getsize(fpath)
cover_path = None
try:
info = handler.extract_tags(fpath)
duration = info.duration if info else 0
artist = info.artist if info else "Unknown"
title = info.title if info else fname
album = info.album if info else ""
genre = info.genre if info else ""
track_num = info.track_number if info else 0
cover_path = info.cover_path if info else None
except Exception:
duration = 0
artist = "Unknown"
title = fname
album = ""
genre = ""
track_num = 0
ready.append({
"path": fpath,
"title": title,
"artist": artist,
"album": album,
"duration_s": duration,
"size": size,
"genre": genre,
"track_num": track_num,
"cover_path": cover_path,
})
elif ext in SOURCE_EXTS:
size = os.path.getsize(fpath)
try:
info = handler.extract_tags(fpath)
title = info.title if info else fname
artist = info.artist if info else "Unknown"
track_num = info.track_number if info else 0
cover_path = info.cover_path if info else None
except Exception:
title = fname
artist = "Unknown"
track_num = 0
cover_path = None
sources.append({
"path": fpath,
"title": title,
"artist": artist,
"size": size,
"track_num": track_num,
"cover_path": cover_path,
})
# Deduplicate by file path (defensive — os.walk shouldn't produce dupes)
seen_paths = set()
deduped_ready = []
for track in ready:
if track["path"] not in seen_paths:
seen_paths.add(track["path"])
deduped_ready.append(track)
ready = deduped_ready
self.library_ready = ready
self.library_scanned_sources = sources
self._refresh_library_ui()
def _refresh_library_ui(self):
self.library_table.setSortingEnabled(False)
self.library_table.setRowCount(0)
seen_paths = set()
all_tracks = []
# Add ready tracks
for t in self.library_ready:
path = t.get("path", "")
if path and path not in seen_paths and os.path.exists(path):
seen_paths.add(path)
all_tracks.append({**t, "_status": "ready"})
# Add scanned source files
for s in self.library_scanned_sources:
path = s.get("path", "")
if path and path not in seen_paths and os.path.exists(path):
seen_paths.add(path)
all_tracks.append({**s, "duration_s": 0, "genre": "", "_status": "source"})
# Add manually added source files
for fpath in self.library_source_files:
if fpath in seen_paths or not os.path.exists(fpath):
continue
fname = os.path.basename(fpath)
size = os.path.getsize(fpath)
try:
handler = MetadataHandler()
info = handler.extract_tags(fpath)
title = info.title if info else fname
artist = info.artist if info else "Unknown"
album = info.album if info else ""
cover_path = info.cover_path if info else None
except Exception:
title = fname
artist = "Unknown"
album = ""
cover_path = None
seen_paths.add(fpath)
all_tracks.append({
"path": fpath, "title": title, "artist": artist,
"size": size, "duration_s": 0, "genre": "", "album": album,
"cover_path": cover_path,
"_status": "source",
})
all_tracks.sort(key=lambda t: (
(t.get("album") or "").lower(),
t.get("track_num", 0) or 0,
t.get("artist", "").lower(),
t.get("title", "").lower(),
))
for idx, track in enumerate(all_tracks):
row = self.library_table.rowCount()
self.library_table.insertRow(row)
is_ready = track["_status"] == "ready"
track_num_str = str(track.get("track_num") or "") if is_ready else ""
duration_str = self._format_duration(track.get("duration_s", 0)) if is_ready else "\u2014"
cells = [
QTableWidgetItem(track_num_str),
QTableWidgetItem(track["artist"]),
QTableWidgetItem(track["title"]),
QTableWidgetItem(track.get("album", "") or ""),
QTableWidgetItem(duration_str),
QTableWidgetItem(track.get("genre", "") or ""),
QTableWidgetItem(self._format_size(track["size"])),
]
# Dim source tracks (gray text) like iTunes
if not is_ready:
gray = self.palette().color(self.palette().ColorRole.PlaceholderText)
for item in cells:
item.setForeground(gray)
# Store full track data in column 0
data_key = track.copy()
data_key.pop("_status", None)
cells[0].setData(Qt.ItemDataRole.UserRole, (is_ready, data_key))
for col, item in enumerate(cells):
self.library_table.setItem(row, col, item)
self.library_table.setSortingEnabled(True)
ready_count = sum(1 for t in all_tracks if t["_status"] == "ready")
source_count = len(all_tracks) - ready_count
has_ready = ready_count > 0
has_source = source_count > 0
self.library_transfer_btn.setEnabled(has_ready)
self.library_convert_btn.setEnabled(has_source)
self.library_remove_btn.setEnabled(len(all_tracks) > 0)
parts = []
if ready_count:
parts.append(f"{ready_count} ready")
if source_count:
parts.append(f"{source_count} need conversion")
status = f"{len(all_tracks)} track(s) \u2014 {', '.join(parts)}" if parts else "No tracks in library"
self.library_status.setText(status)
def _get_selected_ready_tracks(self) -> List[Tuple[TrackInfo, str]]:
selected_rows = set()
for item in self.library_table.selectedItems():
selected_rows.add(item.row())
if not selected_rows:
selected_rows = set(range(self.library_table.rowCount()))
tracks = []
for row in sorted(selected_rows):
data_item = self.library_table.item(row, 0)
if data_item is None:
continue
is_ready, data = data_item.data(Qt.ItemDataRole.UserRole)
if not is_ready:
continue
track_info = TrackInfo(
video_id=f"local_{hash(data['path'])}",
title=data['title'],
artist=data['artist'],
album=data.get("album", ""),
thumbnail_url="",
duration=data.get('duration_s', 0),
track_number=data.get('track_num'),
genre=data.get('genre'),
download_path=data['path'],
)
tracks.append((track_info, data['path']))
return tracks
def _get_selected_source_files(self) -> List[str]:
selected_rows = set()
for item in self.library_table.selectedItems():
selected_rows.add(item.row())
if not selected_rows:
selected_rows = set(range(self.library_table.rowCount()))
paths = []
for row in sorted(selected_rows):
data_item = self.library_table.item(row, 0)
if data_item is None:
continue
is_ready, data = data_item.data(Qt.ItemDataRole.UserRole)
if is_ready:
continue
if os.path.exists(data.get("path", "")):
paths.append(data["path"])
return paths
def _on_browse_output_clicked(self):
directory = QFileDialog.getExistingDirectory(
self, "Select Output Directory", self.output_dir_input.text()
)
if directory:
self.output_dir_input.setText(directory)
self._scan_library()
def _on_refresh_library_clicked(self):
self.library_status.setText("Scanning library...")
self._scan_library()
def _on_library_add_files(self):
source_filter = "Audio Files (*.flac *.wav *.ogg *.opus *.wma *.ape *.aiff *.alac)"
files, _ = QFileDialog.getOpenFileNames(
self, "Select Audio Files", "", source_filter
)
new_files = []
for f in files:
ext = os.path.splitext(f)[1].lower()
if ext in SOURCE_EXTS and os.path.exists(f) and f not in self.library_source_files:
self.library_source_files.append(f)
new_files.append(f)
if new_files:
self._scan_library()
def _on_library_remove_selected(self):
"""Delete selected files from disk and clean up empty directories."""
files_to_delete = []
selected_rows = set()
for item in self.library_table.selectedItems():
selected_rows.add(item.row())
for row in selected_rows:
data_item = self.library_table.item(row, 0)
if data_item is None:
continue
is_ready, data = data_item.data(Qt.ItemDataRole.UserRole)
if data and os.path.exists(data.get("path", "")):
files_to_delete.append(data)
if not files_to_delete:
QMessageBox.information(self, "Info", "No files selected")
return
total_size = sum(f.get("size", 0) for f in files_to_delete)
details = ""
for f in files_to_delete[:15]:
details += f" {f.get('artist', 'Unknown')} {f['title']} ({self._format_size(f.get('size', 0))})\n"
if len(files_to_delete) > 15:
details += f" ... and {len(files_to_delete) - 15} more\n"
reply = QMessageBox.question(
self, "Delete Files",
f"Delete {len(files_to_delete)} file(s) from disk?\n\n{details}\n"
f"Total: {self._format_size(total_size)}\n\n"
f"This action cannot be undone.",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
)
if reply != QMessageBox.StandardButton.Yes:
return
deleted = 0
for f in files_to_delete:
path = f["path"]
try:
os.remove(path)
deleted += 1
if path in self.library_source_files:
self.library_source_files.remove(path)
parent = os.path.dirname(path)
while parent != os.path.dirname(parent):
if os.listdir(parent):
break
os.rmdir(parent)
parent = os.path.dirname(parent)
except OSError as e:
logger.error(f"Failed to delete {path}: {e}")
self._scan_library()
QMessageBox.information(self, "Done", f"Deleted {deleted} file(s)")
def _on_search_text_changed(self, text: str):
for row in range(self.library_table.rowCount()):
item = self.library_table.item(row, 3)
match = not text or text.lower() in item.text().lower() if item else True
self.library_table.setRowHidden(row, not match)
def _on_library_table_context_menu(self, pos):
item = self.library_table.itemAt(pos)
if item is None:
return
row = item.row()
data_item = self.library_table.item(row, 0)
if data_item is None:
return
is_ready, track_data = data_item.data(Qt.ItemDataRole.UserRole)
menu = QMenu(self)
play_action = menu.addAction("\u25B6 Play")
play_action.triggered.connect(lambda: self._play_track_from_index(row))
if is_ready:
transfer_action = menu.addAction("\uD83D\uDCE4 Transfer to iPod")
transfer_action.triggered.connect(self._on_library_transfer_clicked)
else:
convert_action = menu.addAction("\uD83D\uDD04 Convert to iPod Format")
convert_action.triggered.connect(self._on_library_convert_clicked)
menu.addSeparator()
show_action = menu.addAction("\uD83D\uDCC2 Show in File Manager")
show_action.triggered.connect(lambda: self._on_show_in_file_manager(track_data))
remove_action = menu.addAction("\uD83D\uDDD1\uFE0F Remove from Library")
remove_action.triggered.connect(self._on_library_remove_selected)
menu.exec(self.library_table.viewport().mapToGlobal(pos))
def _on_show_in_file_manager(self, track_data):
path = track_data.get("path", "")
if not path:
return
import subprocess
platform = sys.platform
if platform == "win32":
subprocess.Popen(["explorer", "/select,", os.path.normpath(path)])
elif platform == "darwin":
subprocess.Popen(["open", "-R", path])
else:
subprocess.Popen(["xdg-open", os.path.dirname(path)])
def _on_download_clicked(self):
url = self.url_input.text().strip()
if not url:
QMessageBox.warning(self, "Error", "Please enter a YouTube URL")
return
if not (url.startswith("http://") or url.startswith("https://")):
QMessageBox.warning(self, "Error", "Please enter a valid URL")
return
if self.worker_thread and self.worker_thread.isRunning():
QMessageBox.information(self, "Info", "Please wait for the current task to finish")
return
self.download_button.setEnabled(False)
self.download_status.setText("Downloading...")
self.download_progress.setValue(0)
self.track_list.clear()
self.worker_thread = WorkerThread(
task_type="download",
url=url,
format=self.format_combo.currentText(),
quality=self.quality_spin.value()
)
self.worker_thread.progress_signal.connect(self._on_download_progress)
self.worker_thread.finished_signal.connect(self._on_download_finished)
self.worker_thread.start()
def _on_download_progress(self, progress, status):
self.download_progress.setValue(progress)
self.download_status.setText(status)
def _on_download_finished(self, success, message, result):
self.download_button.setEnabled(True)
if success and result:
self.download_status.setText(message)
self.downloaded_tracks = result
self.track_list.clear()
for track in self.downloaded_tracks:
if hasattr(track, 'download_path') and track.download_path:
item = QListWidgetItem(f"{track.artist} - {track.title}")
self.track_list.addItem(item)
self._scan_library()
else:
self.download_status.setText(f"Error: {message}")
QMessageBox.warning(self, "Download Error", message)
self.downloaded_tracks = []
self.track_list.clear()
self._cleanup_worker()
def _on_library_convert_clicked(self):
source_files = self._get_selected_source_files()
has_downloaded = bool(self.downloaded_tracks)
if not source_files and not has_downloaded:
QMessageBox.warning(
self, "Error",
"No tracks to convert. Add local source files or download tracks first."
)
return
if self.worker_thread and self.worker_thread.isRunning():
QMessageBox.information(self, "Info", "Please wait for the current task to finish")
return
output_dir = self.output_dir_input.text().strip()
if not output_dir:
QMessageBox.warning(self, "Error", "Please set an output directory")
return
self.library_convert_btn.setEnabled(False)
self.library_status.setText("Converting...")
self.library_progress.setValue(0)
kwargs = {
"tracks": self.downloaded_tracks if has_downloaded else [],
"local_files": source_files,
"output_dir": output_dir,
"format": self.format_combo.currentText(),
"quality": self.quality_spin.value()
}
self.worker_thread = WorkerThread(task_type="convert", **kwargs)
self.worker_thread.progress_signal.connect(self._on_library_progress)
self.worker_thread.finished_signal.connect(self._on_library_convert_finished)
self.worker_thread.start()
def _on_library_progress(self, progress, status):
self.library_progress.setValue(progress)
self.library_status.setText(status)
def _on_library_convert_finished(self, success, message, result):
self.library_convert_btn.setEnabled(True)
if success:
self.library_status.setText(message)
self._scan_library()
else:
self.library_status.setText(f"Error: {message}")
QMessageBox.warning(self, "Conversion Error", message)
self._cleanup_worker()
def _on_library_transfer_clicked(self):
selected_tracks = self._get_selected_ready_tracks()
if not selected_tracks:
QMessageBox.warning(self, "Error", "No tracks selected. Select tracks to transfer.")
return
if not self.current_mount_point:
QMessageBox.warning(self, "Error", "No iPod device mounted. Go to iPod tab and mount first.")
return
if self.worker_thread and self.worker_thread.isRunning():
QMessageBox.information(self, "Info", "Please wait for the current task to finish")
return
self.library_transfer_btn.setEnabled(False)
self.library_status.setText("Transferring...")
self.library_progress.setValue(0)
self.worker_thread = WorkerThread(
task_type="transfer",
tracks=selected_tracks,
mount_point=self.current_mount_point
)
self.worker_thread.progress_signal.connect(self._on_library_transfer_progress)
self.worker_thread.finished_signal.connect(self._on_library_transfer_finished)
self.worker_thread.start()
def _on_library_transfer_progress(self, progress, status):
self.library_progress.setValue(progress)
self.library_status.setText(status)
def _on_library_transfer_finished(self, success, message, result):
self.library_transfer_btn.setEnabled(True)
if success:
self.library_status.setText(message)
self._scan_library()
self._load_ipod_tracks()
QMessageBox.information(self, "Transfer Complete", message)
else:
self.library_status.setText(f"Error: {message}")
QMessageBox.warning(self, "Transfer Error", message)
self._cleanup_worker()
def _on_refresh_devices_clicked(self):
if self.worker_thread and self.worker_thread.isRunning():
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):
if success:
self.ipod_devices = result
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)
if self.ipod_devices:
device = self.ipod_devices[0]
device_id = device["id"]
detected_mount = device.get("mount_point")
already_mounted = device.get("mounted", False)
if already_mounted and detected_mount:
self._set_device_mounted(device, detected_mount)
else:
ipod = IPodDevice()
mount_point = ipod.mount_device(device_id, detected_mount=detected_mount)
if mount_point:
device["mount_point"] = mount_point
device["mounted"] = True
self._set_device_mounted(device, mount_point)
else:
self._set_device_unmounted(device)
else:
self._set_device_not_found()
self.statusBar().showMessage(message)
else:
self._set_device_not_found()
self.statusBar().showMessage(message)
self._cleanup_worker()
def _set_device_mounted(self, device: dict, mount_point: str):
self.current_mount_point = mount_point
ipod = IPodDevice(mount_point=mount_point)
device_info = ipod.get_device_info()
device["info"] = device_info
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._load_ipod_tracks()
def _set_device_unmounted(self, device: dict):
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)
def _set_device_not_found(self):
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)
def _on_mount_clicked(self):
if not self.ipod_devices:
return
device = self.ipod_devices[0]
device_id = device["id"]
detected_mount = device.get("mount_point")
self.device_status_label.setText("Mounting...")
self.mount_button.setEnabled(False)
try:
ipod = IPodDevice()
mount_point = ipod.mount_device(device_id, detected_mount=detected_mount)
if mount_point:
device["mount_point"] = mount_point
device["mounted"] = True
self._set_device_mounted(device, mount_point)
self.statusBar().showMessage(f"iPod mounted at {mount_point}")
else:
self.device_status_label.setText("Status: Mount failed")
self.mount_button.setEnabled(True)
QMessageBox.warning(self, "Mount Error", "Failed to mount iPod. Check permissions and try again.")
except Exception as e:
self.device_status_label.setText("Status: Mount error")
self.mount_button.setEnabled(True)
QMessageBox.warning(self, "Mount Error", f"Error mounting iPod: {e}")
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
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.statusBar().showMessage("iPod ejected safely")
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
try:
from ipod_nano7_db import Nano7Database
db = Nano7Database(self.current_mount_point)
removed = 0
for item in selected:
track_data = item.data(Qt.ItemDataRole.UserRole)
if track_data and "pid" in track_data:
success = db.delete_track(track_data["pid"])
if success:
removed += 1
self._load_ipod_tracks()
QMessageBox.information(self, "Done", f"Successfully removed {removed} track(s) from iPod")
except Exception as e:
logger.error(f"Failed to remove tracks: {e}")
QMessageBox.warning(self, "Error", f"Failed to remove tracks: {e}")
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.worker_thread and self.worker_thread.isRunning():
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
try:
from ipod_nano7_db import Nano7Database
db = Nano7Database(self.current_mount_point)
removed = db.remove_duplicates()
if not removed:
QMessageBox.information(self, "Scan Complete", "No duplicate tracks found.")
return
details = f"Found and removed {len(removed)} duplicate entry(ies):\n\n"
for r in removed[:15]:
details += f" {r['artist']} \u2014 {r['title']} ({r['album']})\n"
if len(removed) > 15:
details += f" ... and {len(removed) - 15} more\n"
QMessageBox.information(self, "Duplicates Removed", details)
self._load_ipod_tracks()
except Exception as e:
logger.error(f"Failed to remove duplicates: {e}")
QMessageBox.warning(self, "Error", f"Failed to remove duplicates: {e}")
def closeEvent(self, event):
if self.worker_thread and self.worker_thread.isRunning():
self.worker_thread.stop()
self.worker_thread.wait(5000)
self.worker_thread.deleteLater()
event.accept()
# GTK3 platform theme paths for Linux desktop integration
_GTK3_PLUGIN_PATHS = [
"/usr/lib/qt6/plugins/platformthemes/libqgtk3.so",
"/usr/lib/x86_64-linux-gnu/qt6/plugins/platformthemes/libqgtk3.so",
"/usr/lib64/qt6/plugins/platformthemes/libqgtk3.so",
"/usr/lib/aarch64-linux-gnu/qt6/plugins/platformthemes/libqgtk3.so",
]
def _setup_linux_theming():
"""Configure Qt to follow the system GTK theme on Linux.
Must be called BEFORE QApplication is created.
Respects user-set QT_QPA_PLATFORMTHEME (doesn't override it).
Falls back through available platform themes: gtk3 → qt6ct → none.
"""
if sys.platform != "linux":
return
if "QT_QPA_PLATFORMTHEME" in os.environ:
return # User already set it
# Check for libqgtk3.so (official Qt GTK integration)
for path in _GTK3_PLUGIN_PATHS:
if os.path.exists(path):
os.environ["QT_QPA_PLATFORMTHEME"] = "gtk3"
logger.debug(f"Using GTK3 platform theme: {path}")
return
# Check for qt6ct (Qt configuration tool)
if os.path.exists("/usr/bin/qt6ct"):
os.environ["QT_QPA_PLATFORMTHEME"] = "qt6ct"
logger.debug("Using qt6ct platform theme")
return
logger.debug("No GTK platform theme plugin found, using Qt default style")
if __name__ == "__main__":
_setup_linux_theming()
app = QApplication(sys.argv)
# Apply Fusion as fallback style if no platform theme was applied
if not os.environ.get("QT_QPA_PLATFORMTHEME"):
app.setStyle("Fusion")
window = MainWindow()
window.show()
sys.exit(app.exec())
Reference in New Issue View Git Blame Copy Permalink