597603f90e
- Expand Add Files dialog to accept all supported audio formats
- Copy ready files to output_dir/Artist/Album/{track} - Title.ext
- Fallback to output_dir root when metadata is missing
- Handle filename collisions with numeric suffix
1512 lines
58 KiB
Python
1512 lines
58 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Library Tab for neo-pod-desktop.
|
|
Provides the library UI, playback controls, and transfer/conversion workflows.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
import logging
|
|
import shutil
|
|
from typing import List, Dict, Optional, Tuple, Any
|
|
from pathlib import Path
|
|
|
|
from PyQt6.QtWidgets import (
|
|
QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLabel, QLineEdit,
|
|
QProgressBar, QFileDialog, QMessageBox, QGroupBox, QHeaderView,
|
|
QTableWidget, QTableWidgetItem, QSlider, QMenu, QInputDialog,
|
|
QStackedWidget, QListWidget, QListWidgetItem, QListView,
|
|
)
|
|
from PyQt6.QtCore import Qt, QTimer, pyqtSignal, QSize, QUrl, QMimeData
|
|
from PyQt6.QtGui import QPixmap, QColor, QIcon, QPainter, QPainterPath, QDrag
|
|
|
|
from track_info import TrackInfo
|
|
from metadata_handler import MetadataHandler
|
|
from ui.metadata_editor import MetadataEditorDialog
|
|
from library_cache import get_library_cache, invalidate_cache_for_path, save_library_cache
|
|
from artwork.cache import cover_cache
|
|
from config_loader import ConfigLoader
|
|
from worker import WorkerThread
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
AUDIO_EXTS = {'.m4a', '.mp3', '.aac'}
|
|
SOURCE_EXTS = {'.flac', '.wav', '.ogg', '.opus', '.wma', '.ape', '.aiff', '.alac'}
|
|
|
|
EQ_FRAMES = [
|
|
"\u2581\u2583\u2585", "\u2582\u2585\u2587", "\u2583\u2587\u2586",
|
|
"\u2585\u2586\u2584", "\u2586\u2584\u2583", "\u2587\u2583\u2582",
|
|
"\u2586\u2582\u2581", "\u2585\u2581\u2583", "\u2583\u2582\u2585",
|
|
"\u2581\u2583\u2587", "\u2582\u2585\u2586", "\u2583\u2587\u2584",
|
|
"\u2585\u2586\u2583", "\u2586\u2584\u2581",
|
|
]
|
|
|
|
class DraggableLibraryTable(QTableWidget):
|
|
"""QTableWidget subclass that supports dragging tracks as file URLs."""
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
self.setDragEnabled(True)
|
|
|
|
def startDrag(self, supportedActions):
|
|
paths = []
|
|
for item in self.selectedItems():
|
|
data_item = self.item(item.row(), 0)
|
|
if data_item is None:
|
|
continue
|
|
_is_ready, track_data = data_item.data(Qt.ItemDataRole.UserRole)
|
|
if track_data and track_data.get("path"):
|
|
paths.append(track_data["path"])
|
|
|
|
if not paths:
|
|
return
|
|
|
|
paths = list(set(paths))
|
|
mime = QMimeData()
|
|
mime.setUrls([QUrl.fromLocalFile(p) for p in paths])
|
|
|
|
drag = QDrag(self)
|
|
drag.setMimeData(mime)
|
|
|
|
count = len(paths)
|
|
label = f"{count} track{'s' if count != 1 else ''}"
|
|
pixmap = QPixmap(160, 28)
|
|
pixmap.fill(Qt.GlobalColor.transparent)
|
|
painter = QPainter(pixmap)
|
|
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
|
|
bg = self.palette().color(self.palette().ColorRole.Highlight)
|
|
painter.setBrush(bg)
|
|
painter.setPen(Qt.PenStyle.NoPen)
|
|
painter.drawRoundedRect(0, 0, 160, 28, 6, 6)
|
|
painter.setPen(self.palette().color(self.palette().ColorRole.HighlightedText))
|
|
painter.drawText(pixmap.rect(), Qt.AlignmentFlag.AlignCenter, label)
|
|
painter.end()
|
|
drag.setPixmap(pixmap)
|
|
drag.setHotSpot(pixmap.rect().center())
|
|
|
|
drag.exec(Qt.DropAction.CopyAction)
|
|
|
|
|
|
class NumericTableItem(QTableWidgetItem):
|
|
def __lt__(self, other):
|
|
_, self_data = self.data(Qt.ItemDataRole.UserRole) or (False, {})
|
|
_, other_data = other.data(Qt.ItemDataRole.UserRole) or (False, {})
|
|
self_num = (self_data or {}).get("track_num", 0) or 0
|
|
other_num = (other_data or {}).get("track_num", 0) or 0
|
|
return self_num < other_num
|
|
|
|
|
|
class LibraryTab(QWidget):
|
|
"""Library tab widget with library table, filtering, and transfer workflows."""
|
|
|
|
transfer_finished = pyqtSignal()
|
|
play_requested = pyqtSignal(dict, int, bool)
|
|
|
|
def __init__(self, config_loader: ConfigLoader, parent=None):
|
|
super().__init__(parent)
|
|
self.config_loader = config_loader
|
|
|
|
self.library_ready: List[Dict] = []
|
|
self.library_source_files: List[str] = []
|
|
self.library_scanned_sources: List[Dict] = []
|
|
|
|
self.current_playback_index: int = -1
|
|
self._current_track_path: str = ""
|
|
self._eq_active: bool = False
|
|
self._eq_timer: Optional[QTimer] = None
|
|
self._eq_frame: int = 0
|
|
self._eq_running: bool = False
|
|
self._halfway_passed: set[int] = set()
|
|
self._current_view_mode: str = "table"
|
|
self._current_filtered_album: str = ""
|
|
self._current_playlist_id: str = ""
|
|
|
|
self._library_toolbar_buttons: list = []
|
|
|
|
self.worker_thread: Optional[WorkerThread] = None
|
|
|
|
self._current_mount_point: Optional[str] = None
|
|
|
|
self._setup_ui()
|
|
self._eq_timer = QTimer()
|
|
self._eq_timer.setInterval(80)
|
|
self._eq_timer.timeout.connect(self._tick_eq_animation)
|
|
|
|
def set_mount_point(self, mount_point: Optional[str]):
|
|
self._current_mount_point = mount_point
|
|
|
|
def _setup_ui(self):
|
|
layout = QVBoxLayout(self)
|
|
|
|
self._setup_library_toolbar(layout)
|
|
|
|
self.library_table = DraggableLibraryTable()
|
|
self.library_table.setColumnCount(8)
|
|
self.library_table.setHorizontalHeaderLabels(
|
|
["#", "Artist", "Title", "Album", "Duration", "Genre", "Size", "Plays"]
|
|
)
|
|
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.verticalHeader().setVisible(False)
|
|
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)
|
|
self.library_table.horizontalHeader().sortIndicatorChanged.connect(self._on_sort_changed)
|
|
|
|
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)
|
|
header.setSectionResizeMode(7, QHeaderView.ResizeMode.ResizeToContents)
|
|
|
|
self.view_stack = QStackedWidget()
|
|
|
|
self.library_table.setParent(self.view_stack)
|
|
self.view_stack.addWidget(self.library_table)
|
|
|
|
self.album_grid = QListWidget()
|
|
self.album_grid.setViewMode(QListView.ViewMode.IconMode)
|
|
self.album_grid.setIconSize(QSize(140, 140))
|
|
self.album_grid.setGridSize(QSize(170, 210))
|
|
self.album_grid.setWordWrap(True)
|
|
self.album_grid.setResizeMode(QListWidget.ResizeMode.Adjust)
|
|
self.album_grid.setSpacing(8)
|
|
self.album_grid.setMovement(QListWidget.Movement.Static)
|
|
self.album_grid.setStyleSheet("QListWidget { padding: 24px 12px 12px 12px; }")
|
|
self.album_grid.setSelectionMode(QListWidget.SelectionMode.SingleSelection)
|
|
self.album_grid.setFlow(QListWidget.Flow.LeftToRight)
|
|
self.album_grid.setWrapping(True)
|
|
self.album_grid.itemClicked.connect(self._on_album_clicked)
|
|
self.view_stack.addWidget(self.album_grid)
|
|
|
|
self.view_stack.setCurrentIndex(0)
|
|
layout.addWidget(self.view_stack, stretch=1)
|
|
|
|
self.library_progress = QProgressBar()
|
|
self.library_progress.setRange(0, 100)
|
|
self.library_progress.setVisible(False)
|
|
layout.addWidget(self.library_progress)
|
|
|
|
self.library_status = QLabel("Ready")
|
|
self.library_status.setVisible(False)
|
|
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_sync_btn = QPushButton("\uD83D\uDD04 Sync Metadata")
|
|
self.library_sync_btn.setEnabled(False)
|
|
self.library_sync_btn.setToolTip("Sync metadata and artwork for all tracks on iPod")
|
|
self.library_sync_btn.clicked.connect(self._on_sync_metadata_clicked)
|
|
toolbar.addWidget(self.library_sync_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)
|
|
|
|
self.library_edit_btn = QPushButton("\u270F\uFE0F Edit")
|
|
self.library_edit_btn.setEnabled(False)
|
|
self.library_edit_btn.clicked.connect(self._on_edit_metadata)
|
|
toolbar.addWidget(self.library_edit_btn)
|
|
|
|
refresh_btn = QPushButton("\uD83D\uDD04")
|
|
refresh_btn.setToolTip("Refresh Library")
|
|
refresh_btn.clicked.connect(self._on_refresh_library_clicked)
|
|
toolbar.addWidget(refresh_btn)
|
|
|
|
self._library_toolbar_buttons = [
|
|
self.library_add_btn,
|
|
self.library_convert_btn,
|
|
self.library_transfer_btn,
|
|
self.library_sync_btn,
|
|
self.library_remove_btn,
|
|
self.library_edit_btn,
|
|
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 playable_row_indices(self) -> list[int]:
|
|
indices = []
|
|
for row in range(self.library_table.rowCount()):
|
|
if not self.library_table.isRowHidden(row):
|
|
indices.append(row)
|
|
return indices
|
|
|
|
def play_track_at_index(self, index: int, start_position_ms: int = 0, auto_play: bool = True):
|
|
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
|
|
|
|
old_index = self.current_playback_index
|
|
self._clear_playing_row(old_index)
|
|
self._halfway_passed.discard(old_index)
|
|
|
|
self.current_playback_index = index
|
|
self._current_track_path = track_data["path"]
|
|
|
|
self.library_table.clearSelection()
|
|
self.library_table.selectRow(index)
|
|
self._set_playing_row_bg(index)
|
|
|
|
if auto_play:
|
|
self._start_eq_animation()
|
|
|
|
self.play_requested.emit(track_data, start_position_ms, auto_play)
|
|
|
|
def on_position_changed(self, position_ms: int, duration_ms: int):
|
|
if duration_ms > 0 and position_ms >= duration_ms * 0.5:
|
|
self._halfway_passed.add(self.current_playback_index)
|
|
|
|
def on_track_ended(self):
|
|
self._increment_play_count()
|
|
self._stop_eq_animation()
|
|
|
|
def on_track_paused(self):
|
|
self._stop_eq_animation()
|
|
|
|
def on_playing_changed(self, is_playing: bool):
|
|
self._eq_active = is_playing
|
|
if is_playing:
|
|
self._start_eq_animation()
|
|
else:
|
|
self._stop_eq_animation()
|
|
|
|
def _focus_search(self):
|
|
self.library_search_input.setFocus()
|
|
self.library_search_input.selectAll()
|
|
|
|
def _select_all_current(self):
|
|
self.library_table.selectAll()
|
|
|
|
def _delete_selected_current(self):
|
|
self._on_library_remove_selected()
|
|
|
|
def _increment_play_count(self):
|
|
idx = self.current_playback_index
|
|
if idx not in self._halfway_passed:
|
|
return
|
|
if idx < 0 or idx >= self.library_table.rowCount():
|
|
return
|
|
data_item = self.library_table.item(idx, 0)
|
|
if data_item is None:
|
|
return
|
|
_is_ready, track_data = data_item.data(Qt.ItemDataRole.UserRole)
|
|
if not track_data:
|
|
return
|
|
path = track_data.get("path", "")
|
|
if not path:
|
|
return
|
|
cache = get_library_cache()
|
|
entry = cache.get(path)
|
|
if entry is None:
|
|
entry = track_data.copy()
|
|
entry["play_count"] = entry.get("play_count", 0) + 1
|
|
entry["last_played"] = int(time.time())
|
|
cache.put(path, entry)
|
|
save_library_cache()
|
|
self._halfway_passed.discard(idx)
|
|
plays_item = self.library_table.item(idx, 7)
|
|
if plays_item:
|
|
plays_item.setText(str(entry["play_count"]))
|
|
|
|
def _start_eq_animation(self):
|
|
if not self._eq_timer or self._eq_running:
|
|
return
|
|
self._eq_running = True
|
|
self._eq_frame = 0
|
|
self._tick_eq_animation()
|
|
self._eq_timer.start()
|
|
|
|
def _stop_eq_animation(self):
|
|
if not self._eq_timer:
|
|
return
|
|
self._eq_timer.stop()
|
|
self._eq_running = False
|
|
idx = self._find_row_by_path(self._current_track_path) if self._current_track_path else -1
|
|
if idx >= 0:
|
|
item = self.library_table.item(idx, 0)
|
|
if item:
|
|
is_ready, data = item.data(Qt.ItemDataRole.UserRole)
|
|
track_num = data.get("track_num", 0) if data else 0
|
|
item.setText(str(track_num) if (is_ready and track_num) else "")
|
|
item.setData(Qt.ItemDataRole.ForegroundRole, None)
|
|
|
|
def _tick_eq_animation(self):
|
|
if not self._current_track_path:
|
|
return
|
|
idx = self._find_row_by_path(self._current_track_path)
|
|
if idx < 0:
|
|
return
|
|
item = self.library_table.item(idx, 0)
|
|
if item is None:
|
|
return
|
|
glyph = EQ_FRAMES[self._eq_frame % len(EQ_FRAMES)]
|
|
self._eq_frame += 1
|
|
item.setText(glyph)
|
|
accent = self.palette().color(self.palette().ColorRole.Highlight)
|
|
item.setForeground(accent)
|
|
|
|
def _set_playing_row_bg(self, row: int):
|
|
if row < 0 or row >= self.library_table.rowCount():
|
|
return
|
|
bg = QColor(self.palette().color(self.palette().ColorRole.Highlight))
|
|
bg.setAlpha(50)
|
|
for col in range(self.library_table.columnCount()):
|
|
item = self.library_table.item(row, col)
|
|
if item:
|
|
item.setBackground(bg)
|
|
|
|
def _clear_playing_row_bg(self, row: int):
|
|
if row < 0 or row >= self.library_table.rowCount():
|
|
return
|
|
for col in range(self.library_table.columnCount()):
|
|
item = self.library_table.item(row, col)
|
|
if item:
|
|
item.setData(Qt.ItemDataRole.BackgroundRole, None)
|
|
|
|
def _clear_playing_row(self, row: int):
|
|
if row < 0 or row >= self.library_table.rowCount():
|
|
return
|
|
for col in range(self.library_table.columnCount()):
|
|
item = self.library_table.item(row, col)
|
|
if item:
|
|
item.setData(Qt.ItemDataRole.BackgroundRole, None)
|
|
self._eq_timer.stop() if self._eq_timer else None
|
|
self._eq_running = False
|
|
item0 = self.library_table.item(row, 0)
|
|
if item0:
|
|
is_ready, data = item0.data(Qt.ItemDataRole.UserRole)
|
|
track_num = data.get("track_num", 0) if data else 0
|
|
item0.setText(str(track_num) if (is_ready and track_num) else "")
|
|
item0.setData(Qt.ItemDataRole.ForegroundRole, None)
|
|
|
|
def _find_row_by_path(self, path: str) -> int:
|
|
if not path:
|
|
return -1
|
|
for row in range(self.library_table.rowCount()):
|
|
item = self.library_table.item(row, 0)
|
|
if item is None:
|
|
continue
|
|
_, data = item.data(Qt.ItemDataRole.UserRole)
|
|
if data and data.get("path", "") == path:
|
|
return row
|
|
return -1
|
|
|
|
def _on_sort_changed(self, _column: int, _order: int):
|
|
if self._current_track_path:
|
|
idx = self._find_row_by_path(self._current_track_path)
|
|
if idx >= 0:
|
|
self.current_playback_index = idx
|
|
self._set_playing_row_bg(idx)
|
|
|
|
def _on_table_double_clicked(self, row: int):
|
|
self.play_track_at_index(row)
|
|
|
|
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, force: bool = False):
|
|
output_dir = self.get_output_dir()
|
|
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 = []
|
|
cache_hits = 0
|
|
cache_misses = 0
|
|
|
|
cache = get_library_cache()
|
|
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
|
|
|
|
cached = None if force else cache.get(fpath)
|
|
|
|
if cached is not None:
|
|
cache_hits += 1
|
|
if ext in AUDIO_EXTS:
|
|
cached.pop("_mtime", None)
|
|
ready.append(cached)
|
|
elif ext in SOURCE_EXTS:
|
|
cached.pop("_mtime", None)
|
|
sources.append(cached)
|
|
continue
|
|
|
|
cache_misses += 1
|
|
|
|
if ext in AUDIO_EXTS:
|
|
track_data = self._extract_ready_track(fpath, fname, handler)
|
|
ready.append(track_data)
|
|
cache.put(fpath, track_data)
|
|
elif ext in SOURCE_EXTS:
|
|
track_data = self._extract_source_track(fpath, fname, handler)
|
|
sources.append(track_data)
|
|
cache.put(fpath, track_data)
|
|
|
|
cache.save()
|
|
|
|
if cache_misses:
|
|
self.library_status.setText(
|
|
f"Scanned {cache_misses} file(s) \u2014 {cache_hits} from cache")
|
|
|
|
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 _extract_ready_track(self, fpath: str, fname: str, handler) -> Dict[str, Any]:
|
|
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
|
|
album_artist = info.album_artist if info else ""
|
|
except Exception:
|
|
duration = 0
|
|
artist = "Unknown"
|
|
title = fname
|
|
album = ""
|
|
genre = ""
|
|
track_num = 0
|
|
album_artist = ""
|
|
return {
|
|
"path": fpath, "title": title, "artist": artist,
|
|
"album": album, "duration_s": duration, "size": size,
|
|
"genre": genre, "track_num": track_num, "cover_path": cover_path,
|
|
"album_artist": album_artist,
|
|
}
|
|
|
|
def _extract_source_track(self, fpath: str, fname: str, handler) -> Dict[str, Any]:
|
|
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
|
|
return {
|
|
"path": fpath, "title": title, "artist": artist,
|
|
"size": size, "track_num": track_num, "cover_path": cover_path,
|
|
}
|
|
|
|
def _refresh_library_ui(self):
|
|
self.library_table.setSortingEnabled(False)
|
|
self.library_table.setRowCount(0)
|
|
|
|
seen_paths = set()
|
|
all_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"})
|
|
|
|
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"})
|
|
|
|
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 = [
|
|
NumericTableItem(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"])),
|
|
QTableWidgetItem(str(track.get("play_count", "0") or "0") if is_ready else ""),
|
|
]
|
|
|
|
if not is_ready:
|
|
gray = self.palette().color(self.palette().ColorRole.PlaceholderText)
|
|
for item in cells:
|
|
item.setForeground(gray)
|
|
|
|
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
|
|
|
|
has_mount = self._current_mount_point is not None
|
|
self.library_transfer_btn.setEnabled(has_ready and has_mount)
|
|
self.library_sync_btn.setEnabled(has_ready and has_mount)
|
|
self.library_convert_btn.setEnabled(has_source)
|
|
self.library_remove_btn.setEnabled(len(all_tracks) > 0)
|
|
self.library_edit_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)
|
|
|
|
self._restore_playing_row()
|
|
self._build_album_grid()
|
|
|
|
def _restore_playing_row(self):
|
|
idx = self.current_playback_index
|
|
if idx < 0:
|
|
return
|
|
track_path = self._current_track_path
|
|
if not track_path and idx < self.library_table.rowCount():
|
|
item = self.library_table.item(idx, 0)
|
|
if item:
|
|
_, data = item.data(Qt.ItemDataRole.UserRole)
|
|
if data:
|
|
track_path = data.get("path", "")
|
|
|
|
if not track_path:
|
|
self.current_playback_index = -1
|
|
return
|
|
|
|
for row in range(self.library_table.rowCount()):
|
|
item = self.library_table.item(row, 0)
|
|
if item is None:
|
|
continue
|
|
_, data = item.data(Qt.ItemDataRole.UserRole)
|
|
if data and data.get("path") == track_path:
|
|
self.current_playback_index = row
|
|
self._set_playing_row_bg(row)
|
|
if self._eq_active:
|
|
self._start_eq_animation()
|
|
return
|
|
|
|
self.current_playback_index = -1
|
|
|
|
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(
|
|
source_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_refresh_library_clicked(self):
|
|
self.library_status.setText("Scanning library...")
|
|
self._scan_library(force=True)
|
|
|
|
def _on_library_add_files(self):
|
|
all_exts = sorted(SOURCE_EXTS | AUDIO_EXTS)
|
|
source_filter = f"Audio Files (*{' *'.join(all_exts)})"
|
|
files, _ = QFileDialog.getOpenFileNames(
|
|
self, "Select Audio Files", "", source_filter
|
|
)
|
|
new_source_files = []
|
|
copied_ready = []
|
|
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_source_files.append(f)
|
|
elif ext in AUDIO_EXTS and os.path.exists(f):
|
|
output_dir = self.get_output_dir()
|
|
if output_dir:
|
|
dest = self._copy_to_library(f, output_dir)
|
|
if dest:
|
|
copied_ready.append(dest)
|
|
|
|
if new_source_files or copied_ready:
|
|
self._scan_library()
|
|
|
|
def _copy_to_library(self, src: str, output_dir: str) -> str:
|
|
handler = MetadataHandler()
|
|
ext = os.path.splitext(src)[1].lower()
|
|
try:
|
|
info = handler.extract_tags(src)
|
|
artist = info.artist if info and info.artist else None
|
|
album = info.album if info and info.album else None
|
|
title = info.title if info and info.title else None
|
|
track_num = info.track_number if info else 0
|
|
except Exception:
|
|
artist = album = title = None
|
|
track_num = 0
|
|
|
|
if title:
|
|
safe_title = self._sanitize_filename(title)
|
|
safe_artist = self._sanitize_filename(artist) if artist else "_Unknown"
|
|
safe_album = self._sanitize_filename(album) if album else "_Unknown"
|
|
|
|
fname = f"{track_num:02d} - {safe_title}{ext}" if track_num > 0 else f"{safe_title}{ext}"
|
|
dest_dir = os.path.join(output_dir, safe_artist, safe_album)
|
|
else:
|
|
fname = os.path.basename(src)
|
|
dest_dir = output_dir
|
|
|
|
os.makedirs(dest_dir, exist_ok=True)
|
|
dest = os.path.join(dest_dir, fname)
|
|
|
|
if os.path.normpath(src) == os.path.normpath(dest):
|
|
return dest
|
|
if not os.path.exists(dest):
|
|
shutil.copy2(src, dest)
|
|
return dest
|
|
base, ext = os.path.splitext(fname)
|
|
for i in range(1, 100):
|
|
dest = os.path.join(dest_dir, f"{base}_{i}{ext}")
|
|
if not os.path.exists(dest):
|
|
shutil.copy2(src, dest)
|
|
return dest
|
|
return None
|
|
|
|
@staticmethod
|
|
def _sanitize_filename(filename: str) -> str:
|
|
invalid_chars = '<>:"/\\|?*'
|
|
for char in invalid_chars:
|
|
filename = filename.replace(char, '_')
|
|
if len(filename) > 100:
|
|
filename = filename[:97] + '...'
|
|
return filename.strip()
|
|
|
|
def _incremental_update_after_edit(self, file_paths: list):
|
|
cache = get_library_cache()
|
|
handler = MetadataHandler()
|
|
|
|
paths_set = set(file_paths)
|
|
|
|
for fpath in file_paths:
|
|
ext = os.path.splitext(fpath)[1].lower()
|
|
if ext in AUDIO_EXTS:
|
|
data = self._extract_ready_track(fpath, os.path.basename(fpath), handler)
|
|
cache.put(fpath, data)
|
|
for item in self.library_ready:
|
|
if item["path"] == fpath:
|
|
item.update(data)
|
|
break
|
|
else:
|
|
self.library_ready.append(data)
|
|
elif ext in SOURCE_EXTS:
|
|
data = self._extract_source_track(fpath, os.path.basename(fpath), handler)
|
|
cache.put(fpath, data)
|
|
for item in self.library_scanned_sources:
|
|
if item["path"] == fpath:
|
|
item.update(data)
|
|
break
|
|
else:
|
|
self.library_scanned_sources.append(data)
|
|
|
|
cache.save()
|
|
self._refresh_library_ui()
|
|
|
|
def _on_edit_metadata(self):
|
|
selected_rows = set()
|
|
for item in self.library_table.selectedItems():
|
|
selected_rows.add(item.row())
|
|
|
|
if not selected_rows:
|
|
QMessageBox.information(self, "Info", "No track selected")
|
|
return
|
|
|
|
file_paths = []
|
|
for row in sorted(selected_rows):
|
|
data_item = self.library_table.item(row, 0)
|
|
if data_item is None:
|
|
continue
|
|
is_ready, track_data = data_item.data(Qt.ItemDataRole.UserRole)
|
|
file_path = track_data.get("path", "")
|
|
if file_path and os.path.exists(file_path):
|
|
file_paths.append(file_path)
|
|
|
|
if not file_paths:
|
|
QMessageBox.warning(self, "Error", "No valid files found on disk")
|
|
return
|
|
|
|
dialog = MetadataEditorDialog(file_paths, self)
|
|
if dialog.exec() == MetadataEditorDialog.DialogCode.Accepted:
|
|
self._incremental_update_after_edit(file_paths)
|
|
|
|
def _get_first_selected_row(self):
|
|
for item in self.library_table.selectedItems():
|
|
return item.row()
|
|
return None
|
|
|
|
def _on_library_remove_selected(self):
|
|
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}")
|
|
|
|
for f in files_to_delete:
|
|
invalidate_cache_for_path(f["path"])
|
|
|
|
self._scan_library()
|
|
QMessageBox.information(self, "Done", f"Deleted {deleted} file(s)")
|
|
|
|
def _on_search_text_changed(self, text: str):
|
|
query = text.lower()
|
|
for row in range(self.library_table.rowCount()):
|
|
if not query:
|
|
self.library_table.setRowHidden(row, False)
|
|
continue
|
|
match = False
|
|
for col in (1, 2, 3):
|
|
item = self.library_table.item(row, col)
|
|
if item and query in item.text().lower():
|
|
match = True
|
|
break
|
|
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_at_index(row))
|
|
|
|
if is_ready and self._current_mount_point is not None:
|
|
sync_action = menu.addAction("\uD83D\uDD04 Sync Metadata to iPod")
|
|
sync_action.triggered.connect(self._on_sync_metadata_clicked)
|
|
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)
|
|
|
|
add_pl_menu = menu.addMenu("Add to Playlist")
|
|
playlist_manager = getattr(self, "_playlist_manager", None)
|
|
if playlist_manager is not None:
|
|
for pl in playlist_manager.get_all():
|
|
pl_action = add_pl_menu.addAction(f"\U0001F4CB {pl.name}")
|
|
pl_action.triggered.connect(
|
|
lambda checked, pid=pl.id: self._add_selected_to_playlist(pid)
|
|
)
|
|
add_pl_menu.addSeparator()
|
|
new_pl_action = add_pl_menu.addAction("\u2795 New Playlist...")
|
|
new_pl_action.triggered.connect(self._add_selected_to_new_playlist)
|
|
else:
|
|
add_pl_menu.setEnabled(False)
|
|
|
|
menu.addSeparator()
|
|
|
|
edit_action = menu.addAction("\u270F\uFE0F Edit Metadata")
|
|
edit_action.triggered.connect(lambda: self._on_edit_metadata())
|
|
|
|
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 _add_selected_to_playlist(self, playlist_id: str):
|
|
paths = self._get_selected_track_paths()
|
|
if not paths:
|
|
return
|
|
playlist_manager = getattr(self, "_playlist_manager", None)
|
|
if playlist_manager:
|
|
playlist_manager.add_tracks(playlist_id, paths)
|
|
|
|
def _add_selected_to_new_playlist(self):
|
|
from PyQt6.QtWidgets import QInputDialog
|
|
name, ok = QInputDialog.getText(self, "New Playlist", "Playlist name:")
|
|
if not (ok and name.strip()):
|
|
return
|
|
name = name.strip()
|
|
playlist_manager = getattr(self, "_playlist_manager", None)
|
|
if playlist_manager is None:
|
|
return
|
|
paths = self._get_selected_track_paths()
|
|
playlist_id = playlist_manager.create(name)
|
|
if paths:
|
|
playlist_manager.add_tracks(playlist_id, paths)
|
|
self._notify_sidebar_playlist_created(playlist_id, name)
|
|
|
|
def _notify_sidebar_playlist_created(self, playlist_id: str, name: str):
|
|
parent = self.parent()
|
|
while parent is not None:
|
|
sidebar = getattr(parent, "sidebar", None)
|
|
if sidebar is not None:
|
|
sidebar.add_playlist_item(playlist_id, name)
|
|
break
|
|
parent = parent.parent()
|
|
|
|
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_library_convert_clicked(self):
|
|
source_files = self._get_selected_source_files()
|
|
|
|
if not source_files:
|
|
QMessageBox.warning(
|
|
self, "Error",
|
|
"No tracks to convert. Add local source files first."
|
|
)
|
|
return
|
|
|
|
if self._is_worker_running():
|
|
QMessageBox.information(self, "Info", "Please wait for the current task to finish")
|
|
return
|
|
|
|
output_dir = self.get_output_dir()
|
|
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": [],
|
|
"local_files": source_files,
|
|
"output_dir": output_dir,
|
|
"format": self.get_convert_format(),
|
|
"quality": self.get_convert_quality()
|
|
}
|
|
|
|
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()
|
|
if result:
|
|
converted_source_paths = set()
|
|
for track, _output_path in result:
|
|
src = getattr(track, "download_path", None)
|
|
if src:
|
|
converted_source_paths.add(os.path.normpath(src))
|
|
if converted_source_paths:
|
|
self.library_scanned_sources = [
|
|
s for s in self.library_scanned_sources
|
|
if os.path.normpath(s.get("path", "")) not in converted_source_paths
|
|
]
|
|
self.library_source_files = [
|
|
p for p in self.library_source_files
|
|
if os.path.normpath(p) not in converted_source_paths
|
|
]
|
|
self._refresh_library_ui()
|
|
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._is_worker_running():
|
|
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.transfer_finished.emit()
|
|
QMessageBox.information(self, "Transfer Complete", message)
|
|
else:
|
|
self.library_status.setText(f"Error: {message}")
|
|
QMessageBox.warning(self, "Transfer Error", message)
|
|
|
|
self._cleanup_worker()
|
|
|
|
def transfer_tracks_by_paths(self, paths: List[str]):
|
|
"""Transfer tracks by file paths (used for drag-drop onto iPod tab)."""
|
|
ready_by_path = {}
|
|
for t in self.library_ready:
|
|
p = t.get("path", "")
|
|
if p:
|
|
ready_by_path[p] = t
|
|
|
|
handler = MetadataHandler()
|
|
tracks = []
|
|
for path in paths:
|
|
if not os.path.exists(path):
|
|
continue
|
|
if path in ready_by_path:
|
|
d = ready_by_path[path]
|
|
ti = TrackInfo(
|
|
source_id=f"local_{hash(path)}",
|
|
title=d["title"], artist=d["artist"],
|
|
album=d.get("album", ""), thumbnail_url="",
|
|
duration=d.get("duration_s", 0),
|
|
track_number=d.get("track_num"),
|
|
genre=d.get("genre", ""), download_path=path,
|
|
)
|
|
tracks.append((ti, path))
|
|
else:
|
|
try:
|
|
info = handler.extract_tags(path)
|
|
ti = TrackInfo(
|
|
source_id=f"local_{hash(path)}",
|
|
title=info.title if info else os.path.basename(path),
|
|
artist=info.artist if info else "Unknown",
|
|
album=info.album if info else "", thumbnail_url="",
|
|
duration=info.duration if info else 0,
|
|
track_number=info.track_number if info else 0,
|
|
genre=info.genre if info else "", download_path=path,
|
|
)
|
|
tracks.append((ti, path))
|
|
except Exception as e:
|
|
logger.warning(f"Skipping {path}: {e}")
|
|
continue
|
|
|
|
if not tracks:
|
|
QMessageBox.warning(self, "Error", "No valid 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._is_worker_running():
|
|
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=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_sync_metadata_clicked(self):
|
|
selected_tracks = self._get_selected_ready_tracks()
|
|
if not selected_tracks:
|
|
QMessageBox.warning(self, "Error", "No tracks selected.")
|
|
return
|
|
|
|
if not self._current_mount_point:
|
|
QMessageBox.warning(self, "Error", "No iPod device mounted.")
|
|
return
|
|
|
|
if self._is_worker_running():
|
|
QMessageBox.information(self, "Info", "Please wait for the current task to finish")
|
|
return
|
|
|
|
self.library_sync_btn.setEnabled(False)
|
|
self.library_status.setText("Matching tracks on iPod...")
|
|
self.library_progress.setValue(0)
|
|
|
|
from ipod_nano7_db import Nano7Database, content_hash as compute_content_hash
|
|
|
|
db = Nano7Database(self._current_mount_point)
|
|
ipod_tracks = db.get_all_tracks()
|
|
ipod_by_hash = {t.get("content_hash", ""): t for t in ipod_tracks if t.get("content_hash")}
|
|
|
|
sync_batch = []
|
|
errors = []
|
|
for track_info, local_path in selected_tracks:
|
|
ch = compute_content_hash(local_path)
|
|
ipod_track = ipod_by_hash.get(ch)
|
|
if ipod_track is None:
|
|
ipod_track = db.find_track_by_metadata(
|
|
track_info.title, track_info.artist, track_info.album or "",
|
|
)
|
|
if ipod_track is None:
|
|
errors.append(f"'{track_info.title}' — not found on iPod")
|
|
continue
|
|
|
|
sync_batch.append({
|
|
"ipod_pid": ipod_track["pid"],
|
|
"local_path": local_path,
|
|
"new_metadata": {
|
|
"title": track_info.title,
|
|
"artist": track_info.artist,
|
|
"album": track_info.album or "",
|
|
"track_number": track_info.track_number or 0,
|
|
"genre": track_info.genre or "",
|
|
},
|
|
})
|
|
|
|
if not sync_batch:
|
|
msg = "No matching tracks found on iPod"
|
|
if errors:
|
|
msg += ":\n" + "\n".join(errors[:5])
|
|
QMessageBox.warning(self, "Sync Error", msg)
|
|
return
|
|
|
|
if errors:
|
|
logger.warning("Sync metadata skipped for %d unmatched track(s)", len(errors))
|
|
|
|
self.worker_thread = WorkerThread(
|
|
task_type="sync_metadata",
|
|
sync_batch=sync_batch,
|
|
mount_point=self._current_mount_point,
|
|
)
|
|
self.worker_thread.progress_signal.connect(self._on_library_progress)
|
|
self.worker_thread.finished_signal.connect(self._on_sync_metadata_finished)
|
|
self.worker_thread.start()
|
|
|
|
def _on_sync_metadata_finished(self, success, message, result):
|
|
self.library_sync_btn.setEnabled(True)
|
|
if success:
|
|
self.library_status.setText(message)
|
|
QMessageBox.information(self, "Sync Complete", message)
|
|
else:
|
|
self.library_status.setText(f"Error: {message}")
|
|
QMessageBox.warning(self, "Sync Error", message)
|
|
self._cleanup_worker()
|
|
|
|
def _cleanup_worker(self):
|
|
if self.worker_thread:
|
|
self.worker_thread.quit()
|
|
self.worker_thread.wait()
|
|
self.worker_thread = None
|
|
|
|
def _is_worker_running(self):
|
|
if self.worker_thread:
|
|
try:
|
|
return self.worker_thread.isRunning()
|
|
except RuntimeError:
|
|
self.worker_thread = None
|
|
return False
|
|
|
|
def on_device_mounted(self, mount_point: str):
|
|
self._current_mount_point = mount_point
|
|
self.library_progress.setVisible(True)
|
|
self.library_status.setVisible(True)
|
|
|
|
def on_device_unmounted(self):
|
|
self._current_mount_point = None
|
|
self.library_transfer_btn.setEnabled(False)
|
|
self.library_sync_btn.setEnabled(False)
|
|
self.library_progress.setVisible(False)
|
|
self.library_status.setVisible(False)
|
|
|
|
def set_playlist_manager(self, playlist_manager):
|
|
self._playlist_manager = playlist_manager
|
|
|
|
def _get_selected_track_paths(self) -> list:
|
|
paths = []
|
|
for item in self.library_table.selectedItems():
|
|
data_item = self.library_table.item(item.row(), 0)
|
|
if data_item is None:
|
|
continue
|
|
_is_ready, track_data = data_item.data(Qt.ItemDataRole.UserRole)
|
|
if track_data and track_data.get("path"):
|
|
paths.append(track_data["path"])
|
|
return list(set(paths))
|
|
|
|
def apply_filter(self, kind: str, value):
|
|
if kind == "playlist":
|
|
self._apply_playlist_filter(value)
|
|
|
|
def set_view_mode(self, mode: str):
|
|
"""mode: 'table', 'recent', 'artists', 'albums', 'songs'"""
|
|
self._current_view_mode = mode
|
|
self._current_filtered_album = ""
|
|
self._current_playlist_id = ""
|
|
self.view_stack.setCurrentIndex(0)
|
|
self._unhide_all_rows()
|
|
|
|
if mode == "albums":
|
|
self.view_stack.setCurrentIndex(1)
|
|
self.album_grid.clearSelection()
|
|
elif mode == "artists":
|
|
self.library_table.sortItems(1, Qt.SortOrder.AscendingOrder)
|
|
elif mode == "songs":
|
|
self.library_table.sortItems(2, Qt.SortOrder.AscendingOrder)
|
|
elif mode == "recent":
|
|
cutoff = int(time.time()) - (30 * 86400)
|
|
cache = get_library_cache()
|
|
for row in range(self.library_table.rowCount()):
|
|
data_item = self.library_table.item(row, 0)
|
|
if data_item is None:
|
|
continue
|
|
_is_ready, track_data = data_item.data(Qt.ItemDataRole.UserRole)
|
|
path = track_data.get("path", "") if track_data else ""
|
|
entry = cache.get(path) if path else None
|
|
visible = False
|
|
if entry and isinstance(entry, dict):
|
|
added = entry.get("_mtime", 0) or entry.get("added_at", 0) or 0
|
|
visible = added >= cutoff
|
|
self.library_table.setRowHidden(row, not visible)
|
|
|
|
def clear_filter(self):
|
|
self._unhide_all_rows()
|
|
|
|
def _unhide_all_rows(self):
|
|
for row in range(self.library_table.rowCount()):
|
|
self.library_table.setRowHidden(row, False)
|
|
|
|
def _round_pixmap(self, source: QPixmap, radius: int) -> QPixmap:
|
|
size = min(source.width(), source.height())
|
|
result = QPixmap(size, size)
|
|
result.fill(Qt.GlobalColor.transparent)
|
|
p = QPainter(result)
|
|
p.setRenderHint(QPainter.RenderHint.Antialiasing)
|
|
path = QPainterPath()
|
|
path.addRoundedRect(0, 0, size, size, radius, radius)
|
|
p.setClipPath(path)
|
|
p.drawPixmap(0, 0, source.scaled(size, size, Qt.AspectRatioMode.IgnoreAspectRatio,
|
|
Qt.TransformationMode.SmoothTransformation))
|
|
p.end()
|
|
return result
|
|
|
|
def _build_album_grid(self):
|
|
self.album_grid.clear()
|
|
albums = {}
|
|
for track in self.library_ready:
|
|
album = track.get("album", "") or "Unknown Album"
|
|
key = album.lower()
|
|
if key not in albums:
|
|
albums[key] = (album, track.get("cover_path", ""))
|
|
for track in self.library_ready:
|
|
album = track.get("album", "") or "Unknown Album"
|
|
key = album.lower()
|
|
if key in albums and not albums[key][1]:
|
|
albums[key] = (album, track.get("cover_path", ""))
|
|
|
|
for album, cover_path in albums.values():
|
|
pixmap = None
|
|
if cover_path and os.path.exists(cover_path):
|
|
pixmap = QPixmap(cover_path)
|
|
if pixmap is None or pixmap.isNull():
|
|
for track in self.library_ready:
|
|
t_album = track.get("album", "") or "Unknown Album"
|
|
if t_album.lower() == album.lower() and track.get("cover_path"):
|
|
cached = cover_cache.get(
|
|
track.get("album_artist") or track.get("artist", ""),
|
|
album,
|
|
)
|
|
if cached and os.path.exists(cached):
|
|
pixmap = QPixmap(cached)
|
|
break
|
|
if pixmap is None or pixmap.isNull():
|
|
pixmap = QPixmap(140, 140)
|
|
clr = self.palette().color(self.palette().ColorRole.Midlight)
|
|
pixmap.fill(clr)
|
|
|
|
scaled = pixmap.scaled(140, 140, Qt.AspectRatioMode.IgnoreAspectRatio,
|
|
Qt.TransformationMode.SmoothTransformation)
|
|
rounded = self._round_pixmap(scaled, 8)
|
|
icon = QIcon(rounded)
|
|
item = QListWidgetItem(icon, album)
|
|
item.setData(Qt.ItemDataRole.UserRole, album)
|
|
self.album_grid.addItem(item)
|
|
|
|
def _on_album_clicked(self, item: QListWidgetItem):
|
|
album = item.data(Qt.ItemDataRole.UserRole)
|
|
self.album_grid.clearSelection()
|
|
self.filter_by_album(album)
|
|
|
|
def filter_by_album(self, album: str):
|
|
if not album:
|
|
return
|
|
self._current_filtered_album = album
|
|
self._current_view_mode = "albums"
|
|
self.view_stack.setCurrentIndex(0)
|
|
album_lower = album.lower()
|
|
for row in range(self.library_table.rowCount()):
|
|
t_item = self.library_table.item(row, 0)
|
|
if t_item is None:
|
|
self.library_table.setRowHidden(row, False)
|
|
continue
|
|
_is_ready, track_data = t_item.data(Qt.ItemDataRole.UserRole)
|
|
if not track_data:
|
|
self.library_table.setRowHidden(row, True)
|
|
continue
|
|
t_album = (track_data.get("album") or "").lower()
|
|
self.library_table.setRowHidden(row, t_album != album_lower)
|
|
|
|
def _apply_playlist_filter(self, track_paths):
|
|
path_set = set(os.path.normpath(p) for p in track_paths)
|
|
for row in range(self.library_table.rowCount()):
|
|
data_item = self.library_table.item(row, 0)
|
|
if data_item is None:
|
|
self.library_table.setRowHidden(row, False)
|
|
continue
|
|
_is_ready, track_data = data_item.data(Qt.ItemDataRole.UserRole)
|
|
row_path = os.path.normpath(track_data.get("path", "")) if track_data else ""
|
|
self.library_table.setRowHidden(row, row_path not in path_set)
|
|
|
|
def set_buttons_visible(self, visible: bool):
|
|
for btn in self._library_toolbar_buttons:
|
|
btn.setVisible(visible)
|
|
|
|
def get_output_dir(self):
|
|
return self.config_loader.get("General", "output_dir", fallback=os.path.join(os.path.expanduser("~"), "Music", "iPod"))
|
|
|
|
def get_convert_format(self):
|
|
return self.config_loader.get("General", "format", fallback="m4a")
|
|
|
|
def get_convert_quality(self):
|
|
return self.config_loader.get_int("General", "quality", fallback=256)
|
|
|
|
def set_current_playlist_id(self, playlist_id: str):
|
|
self._current_playlist_id = playlist_id
|
|
self._current_view_mode = "playlist"
|
|
|
|
def get_state(self) -> dict:
|
|
return {
|
|
"view_mode": self._current_view_mode,
|
|
"filtered_album": self._current_filtered_album,
|
|
"filtered_playlist_id": self._current_playlist_id,
|
|
}
|