a2d92b75e0
- Add self.library_source_files list to store manually added file paths - _on_library_add_files now adds to library_source_files instead of calling _scan_library() (which would clear the list) - _refresh_source_ui merges scanned sources + library_source_files with dedup by path. All items get proper UserRole data. - _on_library_remove_selected now removes from library_source_files too - Previously added files would be lost on every scan because _scan_library() called .clear() on the source list
1184 lines
45 KiB
Python
1184 lines
45 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Main Application Module for iPod Nano Transfer Tool
|
|
Provides a GUI for the YouTube Music to iPod Nano transfer tool
|
|
"""
|
|
|
|
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
|
|
)
|
|
from PyQt6.QtCore import Qt, QThread, pyqtSignal
|
|
|
|
from youtube_downloader import YouTubeDownloader, TrackInfo
|
|
from audio_converter import AudioConverter
|
|
from metadata_handler import MetadataHandler
|
|
from ipod_device import IPodDevice
|
|
|
|
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,
|
|
}
|
|
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 stop(self):
|
|
self.is_running = False
|
|
self.wait()
|
|
|
|
|
|
class MainWindow(QMainWindow):
|
|
"""Main application window"""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
self.setWindowTitle("YouTube to iPod Nano")
|
|
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.ipod_devices: List[Dict] = []
|
|
self.current_mount_point: Optional[str] = None
|
|
|
|
self._setup_ui()
|
|
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)
|
|
|
|
tab_widget = QTabWidget()
|
|
main_layout.addWidget(tab_widget)
|
|
|
|
download_tab = QWidget()
|
|
tab_widget.addTab(download_tab, "Download")
|
|
self._setup_download_tab(download_tab)
|
|
|
|
library_tab = QWidget()
|
|
tab_widget.addTab(library_tab, "Library")
|
|
self._setup_library_tab(library_tab)
|
|
|
|
ipod_tab = QWidget()
|
|
tab_widget.addTab(ipod_tab, "iPod")
|
|
self._setup_ipod_tab(ipod_tab)
|
|
|
|
settings_tab = QWidget()
|
|
tab_widget.addTab(settings_tab, "Settings")
|
|
self._setup_settings_tab(settings_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)
|
|
|
|
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)
|
|
refresh_btn = QPushButton("Refresh")
|
|
refresh_btn.clicked.connect(self._on_refresh_library_clicked)
|
|
output_layout.addWidget(output_label)
|
|
output_layout.addWidget(self.output_dir_input)
|
|
output_layout.addWidget(browse_btn)
|
|
output_layout.addWidget(refresh_btn)
|
|
layout.addLayout(output_layout)
|
|
|
|
ready_label = QLabel("Ready to Transfer:")
|
|
layout.addWidget(ready_label)
|
|
|
|
self.library_ready_list = QListWidget()
|
|
self.library_ready_list.setSelectionMode(QListWidget.SelectionMode.ExtendedSelection)
|
|
layout.addWidget(self.library_ready_list, stretch=2)
|
|
|
|
source_label = QLabel("Source Files — Need Conversion:")
|
|
layout.addWidget(source_label)
|
|
|
|
self.library_source_list = QListWidget()
|
|
self.library_source_list.setSelectionMode(QListWidget.SelectionMode.ExtendedSelection)
|
|
layout.addWidget(self.library_source_list, stretch=1)
|
|
|
|
btn_row = QHBoxLayout()
|
|
self.library_add_btn = QPushButton("Add Files...")
|
|
self.library_add_btn.clicked.connect(self._on_library_add_files)
|
|
self.library_remove_btn = QPushButton("Remove Selected")
|
|
self.library_remove_btn.clicked.connect(self._on_library_remove_selected)
|
|
btn_row.addWidget(self.library_add_btn)
|
|
btn_row.addWidget(self.library_remove_btn)
|
|
btn_row.addStretch()
|
|
layout.addLayout(btn_row)
|
|
|
|
self.library_convert_btn = QPushButton("Convert Selected")
|
|
self.library_convert_btn.clicked.connect(self._on_library_convert_clicked)
|
|
layout.addWidget(self.library_convert_btn)
|
|
|
|
self.library_transfer_btn = QPushButton("Transfer Selected to iPod ▸")
|
|
self.library_transfer_btn.clicked.connect(self._on_library_transfer_clicked)
|
|
layout.addWidget(self.library_transfer_btn)
|
|
|
|
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_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)
|
|
|
|
def _setup_settings_tab(self, tab):
|
|
layout = QVBoxLayout(tab)
|
|
|
|
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)
|
|
|
|
layout.addWidget(advanced_group)
|
|
|
|
layout.addStretch()
|
|
|
|
about_label = QLabel(
|
|
"YouTube to iPod Nano Transfer Tool\n"
|
|
"Version 1.0.0\n\n"
|
|
"An open-source Python tool that enables direct transfer of YouTube music/playlists to iPod Nano."
|
|
)
|
|
about_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
|
layout.addWidget(about_label)
|
|
|
|
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._refresh_library_ui()
|
|
return
|
|
|
|
ready = []
|
|
sources = []
|
|
|
|
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)
|
|
try:
|
|
handler = MetadataHandler()
|
|
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
|
|
except Exception:
|
|
duration = 0
|
|
artist = "Unknown"
|
|
title = fname
|
|
|
|
ready.append({
|
|
"path": fpath,
|
|
"title": title,
|
|
"artist": artist,
|
|
"duration_s": duration,
|
|
"size": size,
|
|
})
|
|
elif ext in SOURCE_EXTS:
|
|
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"
|
|
except Exception:
|
|
title = fname
|
|
artist = "Unknown"
|
|
sources.append({
|
|
"path": fpath,
|
|
"title": title,
|
|
"artist": artist,
|
|
"size": size,
|
|
})
|
|
|
|
self.library_ready = ready
|
|
self._refresh_library_ui()
|
|
self._refresh_source_ui(sources)
|
|
|
|
def _refresh_library_ui(self):
|
|
self.library_ready_list.clear()
|
|
for track in sorted(self.library_ready, key=lambda t: (t["artist"], t["title"])):
|
|
display = (
|
|
f"{track['artist']} — {track['title']}"
|
|
f" {self._format_duration(track['duration_s'])}"
|
|
f" {self._format_size(track['size'])}"
|
|
)
|
|
item = QListWidgetItem(display)
|
|
item.setData(Qt.ItemDataRole.UserRole, track)
|
|
self.library_ready_list.addItem(item)
|
|
|
|
count = len(self.library_ready)
|
|
ready_label = self.findChild(QLabel, "ready_label")
|
|
if count:
|
|
self.library_status.setText(f"{count} track(s) ready to transfer")
|
|
self.library_transfer_btn.setEnabled(True)
|
|
else:
|
|
self.library_status.setText("No tracks ready — convert source files or download")
|
|
self.library_transfer_btn.setEnabled(False)
|
|
|
|
def _refresh_source_ui(self, scanned_sources):
|
|
"""Build the Source Files list by combining scanned files + manual additions.
|
|
|
|
Deduplicates by path. All items get UserRole data for _get_selected_source_files().
|
|
"""
|
|
self.library_source_list.clear()
|
|
|
|
seen_paths = set()
|
|
all_sources = []
|
|
|
|
for src in scanned_sources:
|
|
path = src.get("path", "")
|
|
if path and path not in seen_paths and os.path.exists(path):
|
|
seen_paths.add(path)
|
|
all_sources.append(src)
|
|
|
|
for fpath in self.library_source_files:
|
|
if fpath in seen_paths:
|
|
continue
|
|
if 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"
|
|
except Exception:
|
|
title = fname
|
|
artist = "Unknown"
|
|
|
|
src = {
|
|
"path": fpath,
|
|
"title": title,
|
|
"artist": artist,
|
|
"size": size,
|
|
}
|
|
all_sources.append(src)
|
|
seen_paths.add(fpath)
|
|
|
|
for src in sorted(all_sources, key=lambda s: (s["artist"], s["title"])):
|
|
display = f"{src['artist']} — {src['title']} {self._format_size(src['size'])}"
|
|
item = QListWidgetItem(display)
|
|
item.setData(Qt.ItemDataRole.UserRole, src)
|
|
self.library_source_list.addItem(item)
|
|
|
|
count = len(all_sources)
|
|
if count:
|
|
self.library_convert_btn.setEnabled(True)
|
|
else:
|
|
self.library_convert_btn.setEnabled(False)
|
|
|
|
def _get_selected_ready_tracks(self) -> List[Tuple[TrackInfo, str]]:
|
|
selected = self.library_ready_list.selectedItems()
|
|
if not selected:
|
|
selected = [self.library_ready_list.item(i) for i in range(self.library_ready_list.count())]
|
|
|
|
tracks = []
|
|
for item in selected:
|
|
data = item.data(Qt.ItemDataRole.UserRole)
|
|
if data:
|
|
track_info = TrackInfo(
|
|
video_id=f"local_{hash(data['path'])}",
|
|
title=data['title'],
|
|
artist=data['artist'],
|
|
album="",
|
|
thumbnail_url="",
|
|
duration=data.get('duration_s', 0),
|
|
download_path=data['path'],
|
|
)
|
|
tracks.append((track_info, data['path']))
|
|
return tracks
|
|
|
|
def _get_selected_source_files(self) -> List[str]:
|
|
selected = self.library_source_list.selectedItems()
|
|
if not selected:
|
|
selected = [self.library_source_list.item(i) for i in range(self.library_source_list.count())]
|
|
|
|
paths = []
|
|
for item in selected:
|
|
data = item.data(Qt.ItemDataRole.UserRole)
|
|
if data and 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._refresh_source_ui([])
|
|
|
|
def _on_library_remove_selected(self):
|
|
"""Delete selected files from disk and clean up empty directories."""
|
|
files_to_delete = []
|
|
for lst in (self.library_ready_list, self.library_source_list):
|
|
for item in lst.selectedItems():
|
|
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_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()
|
|
|
|
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()
|
|
|
|
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)}")
|
|
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_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 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())
|