en2zmax/neo-pod-desktop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: en2zmax:54cd87654f6a31e1a92cacfcfa85830317559ad7
Branches Tags
en2zmax:main
..
compare: en2zmax:dd93440146bb994d456f74ebaba30f4c0bff639e
Branches Tags
en2zmax:main

No commits in common. "54cd87654f6a31e1a92cacfcfa85830317559ad7" and "dd93440146bb994d456f74ebaba30f4c0bff639e" have entirely different histories.
54cd87654f ... dd93440146

19 changed files with 357 additions and 2853 deletions
Show Stats Expand all files Collapse all files

230
src/app.py
View File

@ -6,7 +6,6 @@ Provides a GUI for converting, managing and transferring music to iPod Nano devi
import os import os
import sys import sys
import random
import logging import logging
from typing import Optional from typing import Optional
@ -14,7 +13,7 @@ from PyQt6.QtWidgets import (
QApplication, QMainWindow, QWidget, QVBoxLayout, QApplication, QMainWindow, QWidget, QVBoxLayout,
QSplitter, QStackedWidget, QSplitter, QStackedWidget,
) )
from PyQt6.QtCore import Qt, QUrl, QTimer from PyQt6.QtCore import Qt, QUrl
from PyQt6.QtGui import QPixmap from PyQt6.QtGui import QPixmap
from PyQt6.QtMultimedia import QMediaPlayer, QAudioOutput from PyQt6.QtMultimedia import QMediaPlayer, QAudioOutput
@ -22,14 +21,11 @@ from config_loader import ConfigLoader
from device_monitor import DeviceMonitor from device_monitor import DeviceMonitor
from hotkeys import HotkeyManager from hotkeys import HotkeyManager
from playlist_manager import PlaylistManager from playlist_manager import PlaylistManager
from session_store import SessionStore
from library_cache import get_library_cache
from ui.library_tab import LibraryTab from ui.library_tab import LibraryTab
from ui.ipod_tab import iPodTab from ui.ipod_tab import iPodTab
from ui.settings_tab import SettingsTab from ui.settings_tab import SettingsTab
from ui.sidebar_widget import SIDEBAR_DEFAULT_WIDTH from ui.sidebar_widget import SIDEBAR_DEFAULT_WIDTH
from ui.player_header import PlayerHeader from ui.player_header import PlayerHeader
from ui.splash_overlay import SplashOverlay
from artwork.cache import cover_cache from artwork.cache import cover_cache
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
@ -48,12 +44,10 @@ class MainWindow(QMainWindow):
self.config_loader = ConfigLoader() self.config_loader = ConfigLoader()
self.hotkey_manager: Optional[HotkeyManager] = None self.hotkey_manager: Optional[HotkeyManager] = None
self.device_monitor: Optional[DeviceMonitor] = None self.device_monitor: Optional[DeviceMonitor] = None
self.ipod_tab_index = 1
self.playlist_manager = PlaylistManager() self.playlist_manager = PlaylistManager()
self.session_store = SessionStore()
self._sidebar_visible = True self._sidebar_visible = True
self._pending_playlist_id: str = ""
self._pending_playlist_tracks: list = []
self._saved_volume: int = 80 self._saved_volume: int = 80
self._saved_track_path: str = "" self._saved_track_path: str = ""
@ -61,11 +55,6 @@ class MainWindow(QMainWindow):
self._saved_playing: bool = False self._saved_playing: bool = False
self._pending_seek_ms: int = 0 self._pending_seek_ms: int = 0
self._shuffle_active: bool = False
self._repeat_mode: str = "no_repeat"
self._auto_sync_worker = None
self._setup_player() self._setup_player()
self._setup_ui() self._setup_ui()
self.library_tab.set_playlist_manager(self.playlist_manager) self.library_tab.set_playlist_manager(self.playlist_manager)
@ -74,18 +63,10 @@ class MainWindow(QMainWindow):
self._setup_hotkeys() self._setup_hotkeys()
self._wire_signals() self._wire_signals()
cache = get_library_cache() self.library_tab._scan_library()
splash = SplashOverlay(self)
if len(cache) == 0:
splash.show()
QApplication.processEvents()
self.library_tab._scan_library(on_progress=splash.set_progress)
if splash.isVisible():
splash.close()
self._resume_playback_if_saved() self._resume_playback_if_saved()
self._restore_library_state() self._restore_library_state()
self.ipod_tab._on_refresh_devices_clicked()
self._start_device_monitor() self._start_device_monitor()
def _setup_player(self): def _setup_player(self):
@ -109,8 +90,6 @@ class MainWindow(QMainWindow):
self.player_header.prev_requested.connect(self._on_prev) self.player_header.prev_requested.connect(self._on_prev)
self.player_header.play_pause_requested.connect(self._on_play_pause) self.player_header.play_pause_requested.connect(self._on_play_pause)
self.player_header.next_requested.connect(self._on_next) self.player_header.next_requested.connect(self._on_next)
self.player_header.shuffle_requested.connect(self._on_shuffle_toggle)
self.player_header.repeat_requested.connect(self._on_repeat_cycle)
self.player_header.position_changed_by_user.connect(self._on_user_seek_frac) self.player_header.position_changed_by_user.connect(self._on_user_seek_frac)
self.player_header.volume_changed.connect(self._on_volume_changed) self.player_header.volume_changed.connect(self._on_volume_changed)
main_layout.addWidget(self.player_header) main_layout.addWidget(self.player_header)
@ -156,12 +135,6 @@ class MainWindow(QMainWindow):
) )
self.settings_tab.auto_detect_changed.connect(self._on_auto_detect_toggled) self.settings_tab.auto_detect_changed.connect(self._on_auto_detect_toggled)
self.sidebar.tracks_dropped_on_playlist.connect(self._on_tracks_dropped_on_playlist)
self.sidebar.tracks_dropped_for_nav_transfer.connect(self._on_tracks_dropped_for_nav_transfer)
self.sidebar.playlist_dropped_for_nav_transfer.connect(self._on_playlist_dropped_for_nav_transfer)
self.ipod_tab.tracks_dropped_for_transfer.connect(self._on_tracks_dropped_for_transfer)
self.library_tab.transfer_finished.connect(self._on_transfer_finished_for_playlist)
def _on_sidebar_item_selected(self, kind: str, value: str): def _on_sidebar_item_selected(self, kind: str, value: str):
if kind == "tab": if kind == "tab":
tab_index = {"library": 0, "ipod": 1, "settings": 2}.get(value, 0) tab_index = {"library": 0, "ipod": 1, "settings": 2}.get(value, 0)
@ -202,102 +175,18 @@ class MainWindow(QMainWindow):
self.library_tab.clear_filter() self.library_tab.clear_filter()
self.sidebar.select_library_music() self.sidebar.select_library_music()
def _on_tracks_dropped_on_playlist(self, playlist_id: str, paths: list):
self.playlist_manager.add_tracks(playlist_id, paths)
playlist = self.playlist_manager.get_by_id(playlist_id)
if playlist:
self._on_playlist_selected(playlist_id)
self.statusBar().showMessage(f"Added {len(paths)} track(s) to playlist", 5000)
def _on_tracks_dropped_for_transfer(self, paths: list):
self.library_tab.transfer_tracks_by_paths(paths)
def _on_tracks_dropped_for_nav_transfer(self, paths: list):
self.stack.setCurrentIndex(0)
self.library_tab.transfer_tracks_by_paths(paths)
def _on_playlist_dropped_for_nav_transfer(self, playlist_id: str):
playlist = self.playlist_manager.get_by_id(playlist_id)
if not playlist or not playlist.track_paths:
return
self.stack.setCurrentIndex(0)
self._pending_playlist_id = playlist_id
self._pending_playlist_tracks = list(playlist.track_paths)
self.library_tab.transfer_tracks_by_paths(playlist.track_paths)
def _on_transfer_finished_for_playlist(self):
playlist_id = self._pending_playlist_id
track_paths = self._pending_playlist_tracks
self._pending_playlist_id = ""
self._pending_playlist_tracks = []
if not playlist_id or not track_paths:
return
mount_point = getattr(self.ipod_tab, "current_mount_point", None)
if not mount_point:
self.statusBar().showMessage("Cannot create playlist: iPod not mounted", 5000)
return
playlist = self.playlist_manager.get_by_id(playlist_id)
if not playlist:
return
self.statusBar().showMessage(f"Creating playlist '{playlist.name}' on iPod...")
from worker import WorkerThread
self._pl_worker = WorkerThread(
task_type="create_playlist",
mount_point=mount_point,
playlist_name=playlist.name,
track_paths=track_paths,
)
self._pl_worker.finished_signal.connect(self._on_playlist_creation_finished)
self._pl_worker.start()
def _on_playlist_creation_finished(self, success, message, result):
self.statusBar().showMessage(message, 8000)
try:
self._pl_worker.quit()
self._pl_worker.wait()
except Exception:
pass
self._pl_worker = None
def _load_playlists_into_sidebar(self): def _load_playlists_into_sidebar(self):
for pl in self.playlist_manager.get_all(): for pl in self.playlist_manager.get_all():
self.sidebar.add_playlist_item(pl.id, pl.name) self.sidebar.add_playlist_item(pl.id, pl.name)
def _on_device_mounted(self, mount_point: str): def _on_device_mounted(self, mount_point: str):
self.library_tab.on_device_mounted(mount_point) self.library_tab.on_device_mounted(mount_point)
self.library_tab.refresh_play_counts_from_cache()
self.statusBar().showMessage(f"iPod mounted at {mount_point}") self.statusBar().showMessage(f"iPod mounted at {mount_point}")
QTimer.singleShot(2000, lambda: self._start_auto_sync(mount_point))
def _on_device_unmounted(self): def _on_device_unmounted(self):
self.library_tab.on_device_unmounted() self.library_tab.on_device_unmounted()
self.statusBar().showMessage("iPod disconnected") self.statusBar().showMessage("iPod disconnected")
def _start_auto_sync(self, mount_point: str):
if self._auto_sync_worker is not None:
return
from worker import WorkerThread
self._auto_sync_worker = WorkerThread(
"auto_sync_playcounts", parent=self, mount_point=mount_point,
)
self._auto_sync_worker.progress_signal.connect(self._on_auto_sync_progress)
self._auto_sync_worker.finished_signal.connect(self._on_auto_sync_finished)
self._auto_sync_worker.start()
def _on_auto_sync_progress(self, pct: int, msg: str):
pass
def _on_auto_sync_finished(self, success: bool, msg: str, _result):
if success and msg.startswith("Auto-sync complete"):
self.library_tab.refresh_play_counts_from_cache()
self._auto_sync_worker = None
def _toggle_sidebar(self): def _toggle_sidebar(self):
visible = not self.sidebar.isVisible() visible = not self.sidebar.isVisible()
self.sidebar.setVisible(visible) self.sidebar.setVisible(visible)
@ -330,43 +219,29 @@ class MainWindow(QMainWindow):
hide_buttons = config.get_boolean("Advanced", "hide_library_buttons", fallback=False) hide_buttons = config.get_boolean("Advanced", "hide_library_buttons", fallback=False)
self.library_tab.set_buttons_visible(not hide_buttons) self.library_tab.set_buttons_visible(not hide_buttons)
# Migrate Playback section → SessionStore if needed self._saved_volume = config.get_int("Playback", "last_volume", fallback=80)
self.session_store.migrate_from_config(config) self._saved_track_path = config.get("Playback", "last_track_path", fallback="")
self._saved_position_ms = config.get_int("Playback", "last_position_ms", fallback=0)
self._saved_volume = self.session_store.get_int("last_volume", 80) self._saved_playing = config.get_boolean("Playback", "last_playing", fallback=False)
self._saved_track_path = self.session_store.get_str("last_track_path", "")
self._saved_position_ms = self.session_store.get_int("last_position_ms", 0)
self._saved_playing = self.session_store.get_bool("last_playing", False)
self._shuffle_active = self.session_store.get_bool("shuffle_active", False)
self._repeat_mode = self.session_store.get_str("repeat_mode", "no_repeat")
self.player_header.set_shuffle_active(self._shuffle_active)
self.player_header.set_repeat_mode(self._repeat_mode)
self.audio_output.setVolume(self._saved_volume / 100.0) self.audio_output.setVolume(self._saved_volume / 100.0)
self.player_header.set_volume(self._saved_volume) self.player_header.set_volume(self._saved_volume)
def _save_settings(self): def _save_settings(self):
config = self.config_loader config = self.config_loader
session = self.session_store
self.settings_tab.save_settings(config) self.settings_tab.save_settings(config)
config.set("Playback", "last_volume", str(self.player_header.volume_slider.value()))
session.set("last_volume", self.player_header.volume_slider.value())
session.set("shuffle_active", self._shuffle_active)
session.set("repeat_mode", self._repeat_mode)
if self.current_track_path and os.path.exists(self.current_track_path): if self.current_track_path and os.path.exists(self.current_track_path):
session.set("last_track_path", self.current_track_path) config.set("Playback", "last_track_path", self.current_track_path)
position = self.player.position() position = self.player.position()
duration = self.player.duration() duration = self.player.duration()
if duration > 0 and position > duration - 2000: if duration > 0 and position > duration - 2000:
position = 0 position = 0
session.set("last_position_ms", position) config.set("Playback", "last_position_ms", str(position))
session.set("last_playing", self._is_playing_now) config.set("Playback", "last_playing", str(self._is_playing_now).lower())
else: else:
session.set("last_track_path", "") config.set("Playback", "last_track_path", "")
session.set("last_position_ms", 0) config.set("Playback", "last_position_ms", "0")
session.save()
state = self.library_tab.get_state() state = self.library_tab.get_state()
config.set("Library", "view_mode", state.get("view_mode", "table")) config.set("Library", "view_mode", state.get("view_mode", "table"))
config.set("Library", "filtered_album", state.get("filtered_album", "")) config.set("Library", "filtered_album", state.get("filtered_album", ""))
@ -395,7 +270,7 @@ class MainWindow(QMainWindow):
"seek_backward": self._seek_backward, "seek_backward": self._seek_backward,
"toggle_sidebar": self._toggle_sidebar, "toggle_sidebar": self._toggle_sidebar,
"tab_library": lambda: self.stack.setCurrentIndex(0), "tab_library": lambda: self.stack.setCurrentIndex(0),
"tab_ipod": lambda: self.stack.setCurrentIndex(1), "tab_ipod": lambda: self.stack.setCurrentIndex(self.ipod_tab_index),
"tab_settings": lambda: self.stack.setCurrentIndex(2), "tab_settings": lambda: self.stack.setCurrentIndex(2),
"search_focus": self.library_tab._focus_search, "search_focus": self.library_tab._focus_search,
"select_all": self.library_tab._select_all_current, "select_all": self.library_tab._select_all_current,
@ -462,40 +337,9 @@ class MainWindow(QMainWindow):
idx = max(0, self.library_tab.current_playback_index - 1) idx = max(0, self.library_tab.current_playback_index - 1)
self.library_tab.play_track_at_index(idx) self.library_tab.play_track_at_index(idx)
def _get_next_index(self):
playable = self.library_tab.playable_row_indices()
if not playable:
return None
current = self.library_tab.current_playback_index
if self._shuffle_active:
candidates = [r for r in playable if r != current]
if not candidates:
candidates = playable
return random.choice(candidates)
try:
pos = playable.index(current)
except ValueError:
pos = -1
if pos >= 0 and pos < len(playable) - 1:
return playable[pos + 1]
if self._repeat_mode == "repeat_all" and playable:
return playable[0]
return None
def _stop_playback(self):
self.player.stop()
self.library_tab.on_playing_changed(False)
self.player_header.set_playing(False)
def _on_next(self): def _on_next(self):
idx = self._get_next_index() rows = self.library_tab.library_table.rowCount()
if idx is not None: idx = min(rows - 1, self.library_tab.current_playback_index + 1)
self.library_tab.play_track_at_index(idx) self.library_tab.play_track_at_index(idx)
def _seek_forward(self): def _seek_forward(self):
@ -519,13 +363,8 @@ class MainWindow(QMainWindow):
if self.audio_output: if self.audio_output:
self.audio_output.setVolume(value / 100.0) self.audio_output.setVolume(value / 100.0)
def _on_user_seek_frac(self, value: int): def _on_user_seek_frac(self, frac: int):
duration = self.player.duration() self.player.setPosition(int(frac * self.player.duration()))
if duration <= 0:
return
new_pos = int(value / 1000.0 * duration)
new_pos = max(0, min(duration - 1, new_pos))
self.player.setPosition(new_pos)
def _on_player_position(self, position: int): def _on_player_position(self, position: int):
duration = self.player.duration() duration = self.player.duration()
@ -536,29 +375,10 @@ class MainWindow(QMainWindow):
pos = self.player.position() pos = self.player.position()
self.player_header.set_position(pos, duration) self.player_header.set_position(pos, duration)
def _on_shuffle_toggle(self):
self._shuffle_active = not self._shuffle_active
self.player_header.set_shuffle_active(self._shuffle_active)
def _on_repeat_cycle(self):
cycle = {"no_repeat": "repeat_all", "repeat_all": "repeat_one", "repeat_one": "no_repeat"}
self._repeat_mode = cycle[self._repeat_mode]
self.player_header.set_repeat_mode(self._repeat_mode)
def _on_media_status(self, status): def _on_media_status(self, status):
if status == QMediaPlayer.MediaStatus.EndOfMedia: if status == QMediaPlayer.MediaStatus.EndOfMedia:
self.library_tab.on_track_ended() self.library_tab.on_track_ended()
self._on_next()
if self._repeat_mode == "repeat_one":
self.player.setPosition(0)
self.player.play()
return
next_idx = self._get_next_index()
if next_idx is not None:
self.library_tab.play_track_at_index(next_idx)
else:
self._stop_playback()
elif status == QMediaPlayer.MediaStatus.NoMedia: elif status == QMediaPlayer.MediaStatus.NoMedia:
self.library_tab.on_playing_changed(False) self.library_tab.on_playing_changed(False)
self.player_header.reset_info() self.player_header.reset_info()
@ -607,17 +427,7 @@ class MainWindow(QMainWindow):
def closeEvent(self, event): def closeEvent(self, event):
if self.device_monitor: if self.device_monitor:
self.device_monitor.stop() self.device_monitor.stop()
self.ipod_tab.stop_monitoring(self.device_monitor)
for tab in (self.ipod_tab, self.library_tab):
if hasattr(tab, 'worker_thread') and tab.worker_thread:
try:
if tab.worker_thread.isRunning():
tab.worker_thread.stop()
except RuntimeError:
tab.worker_thread = None
self._save_settings() self._save_settings()
from PyQt6.QtWidgets import QApplication
QApplication.processEvents()
event.accept() event.accept()

15
src/artwork/cache.py
View File

@ -10,11 +10,22 @@ import hashlib
import logging import logging
import os import os
from xdg_base import xdg_cache_path
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def xdg_cache_path(subdir: str) -> str:
"""Return path inside XDG cache directory for neo-pod-desktop.
Creates the directory structure if missing.
"""
base = os.environ.get(
"XDG_CACHE_HOME",
os.path.join(os.path.expanduser("~"), ".cache"),
)
path = os.path.join(base, "neo-pod-desktop", subdir)
os.makedirs(path, exist_ok=True)
return path
class CoverCache: class CoverCache:
"""Manages a local cache of cover art images for the player UI. """Manages a local cache of cover art images for the player UI.

13
src/config_loader.py
View File

@ -163,12 +163,17 @@ class ConfigLoader:
if not self.config.has_option('Hotkeys', key): if not self.config.has_option('Hotkeys', key):
self.config.set('Hotkeys', key, value) self.config.set('Hotkeys', key, value)
# Playback section is DEPRECATED — session state moved to # Playback section
# session.json in ~/.local/share/. The defaults below are
# kept only for backward-compatible reads during migration.
# All new code should use SessionStore instead.
if not self.config.has_section('Playback'): if not self.config.has_section('Playback'):
self.config.add_section('Playback') self.config.add_section('Playback')
if not self.config.has_option('Playback', 'last_track_path'):
self.config.set('Playback', 'last_track_path', '')
if not self.config.has_option('Playback', 'last_position_ms'):
self.config.set('Playback', 'last_position_ms', '0')
if not self.config.has_option('Playback', 'last_volume'):
self.config.set('Playback', 'last_volume', '80')
if not self.config.has_option('Playback', 'last_playing'):
self.config.set('Playback', 'last_playing', 'false')
def get(self, section: str, option: str, fallback: Any = None) -> Any: def get(self, section: str, option: str, fallback: Any = None) -> Any:
""" """

25
src/device_monitor.py
View File

@ -11,9 +11,6 @@ logger = logging.getLogger(__name__)
class _DetectWorker(QThread): class _DetectWorker(QThread):
finished = pyqtSignal(object) finished = pyqtSignal(object)
def __init__(self, parent=None):
super().__init__(parent)
def run(self): def run(self):
try: try:
devices = IPodDevice().detect_devices() devices = IPodDevice().detect_devices()
@ -41,29 +38,19 @@ class DeviceMonitor(QObject):
def stop(self): def stop(self):
self.timer.stop() self.timer.stop()
if self._poll_worker: if self._poll_worker and self._poll_worker.isRunning():
self._poll_worker.quit() self._poll_worker.quit()
self._poll_worker.wait() self._poll_worker.wait(3000)
self._poll_worker = None
self._device_ids.clear() self._device_ids.clear()
def _poll(self): def _poll(self):
if self._poll_worker: if self._poll_worker and self._poll_worker.isRunning():
try:
if self._poll_worker.isRunning():
return return
except RuntimeError: self._poll_worker = _DetectWorker()
self._poll_worker = None self._poll_worker.finished.connect(self._on_devices_detected)
worker = _DetectWorker() self._poll_worker.start()
worker.finished.connect(self._on_devices_detected)
worker.start()
self._poll_worker = worker
def _on_devices_detected(self, devices): def _on_devices_detected(self, devices):
if self._poll_worker:
self._poll_worker.quit()
self._poll_worker.wait()
self._poll_worker = None
current_ids = {d["id"] for d in devices} current_ids = {d["id"] for d in devices}
if current_ids != self._device_ids: if current_ids != self._device_ids:
self._device_ids = current_ids self._device_ids = current_ids

48
src/hotkeys.py
View File

@ -1,7 +1,7 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
Hotkey Manager for neo-pod-desktop Hotkey Manager for neo-pod-desktop
Provides configurable keyboard shortcuts. Provides configurable keyboard shortcuts with scope awareness.
""" """
import logging import logging
@ -73,27 +73,28 @@ def _is_editable_focused() -> bool:
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Default shortcut definitions # Default shortcut definitions
# (action_name, default_key_string, display_label) # (action_name, default_key_string, display_label, scope)
# scope: "global" — always active; "tab" — handler decides based on current tab
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
DEFAULTS = [ DEFAULTS = [
("play_pause", "Space", "Play / Pause"), ("play_pause", "Space", "Play / Pause", "global"),
("prev_track", "Ctrl+Left", "Previous Track"), ("prev_track", "Ctrl+Left", "Previous Track", "global"),
("next_track", "Ctrl+Right", "Next Track"), ("next_track", "Ctrl+Right", "Next Track", "global"),
("volume_up", "Ctrl+Up", "Volume Up"), ("volume_up", "Ctrl+Up", "Volume Up", "global"),
("volume_down", "Ctrl+Down", "Volume Down"), ("volume_down", "Ctrl+Down", "Volume Down", "global"),
("seek_forward", "Right", "Seek Forward 5s"), ("seek_forward", "Right", "Seek Forward 5s", "global"),
("seek_backward", "Left", "Seek Backward 5s"), ("seek_backward", "Left", "Seek Backward 5s", "global"),
("tab_library", "Ctrl+1", "Switch to Library"), ("tab_library", "Ctrl+1", "Switch to Library", "global"),
("tab_ipod", "Ctrl+2", "Switch to iPod"), ("tab_ipod", "Ctrl+2", "Switch to iPod", "global"),
("tab_settings", "Ctrl+3", "Switch to Settings"), ("tab_settings", "Ctrl+3", "Switch to Settings", "global"),
("search_focus", "Ctrl+F", "Focus Search"), ("search_focus", "Ctrl+F", "Focus Search", "global"),
("select_all", "Ctrl+A", "Select All"), ("select_all", "Ctrl+A", "Select All", "global"),
("library_refresh", "F5", "Refresh Library"), ("library_refresh", "F5", "Refresh Library", "global"),
("ipod_refresh", "Ctrl+R", "Refresh Devices"), ("ipod_refresh", "Ctrl+R", "Refresh Devices", "global"),
("delete_selected", "Delete", "Delete Selected"), ("delete_selected", "Delete", "Delete Selected", "global"),
("edit_metadata", "F2", "Edit Metadata"), ("edit_metadata", "F2", "Edit Metadata", "global"),
("toggle_sidebar", "Ctrl+B", "Toggle Sidebar"), ("toggle_sidebar", "Ctrl+B", "Toggle Sidebar", "global"),
("library_add_files", "Ctrl+O", "Library: Add Files"), ("library_add_files", "Ctrl+O", "Library: Add Files", "global"),
] ]
@ -172,15 +173,17 @@ class HotkeyManager:
self._parent = parent self._parent = parent
self._config_loader = config_loader self._config_loader = config_loader
# action_name → (QShortcut, callback, default_seq) # action_name → (QShortcut, callback, default_seq, scope)
self._shortcuts: Dict[str, QShortcut] = {} self._shortcuts: Dict[str, QShortcut] = {}
self._callbacks: Dict[str, Callable[[], None]] = {} self._callbacks: Dict[str, Callable[[], None]] = {}
self._defaults: Dict[str, QKeySequence] = {} self._defaults: Dict[str, QKeySequence] = {}
self._labels: Dict[str, str] = {} self._labels: Dict[str, str] = {}
self._scopes: Dict[str, str] = {}
for name, key_str, label in DEFAULTS: for name, key_str, label, scope in DEFAULTS:
self._defaults[name] = _build_seq(key_str) self._defaults[name] = _build_seq(key_str)
self._labels[name] = label self._labels[name] = label
self._scopes[name] = scope
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# Registration # Registration
@ -198,6 +201,7 @@ class HotkeyManager:
seq = _build_seq(saved_key) if saved_key else self._defaults.get(action_name, QSeq()) seq = _build_seq(saved_key) if saved_key else self._defaults.get(action_name, QSeq())
self._callbacks[action_name] = callback self._callbacks[action_name] = callback
scope = self._scopes.get(action_name, "global")
context = Qt.ShortcutContext.WindowShortcut context = Qt.ShortcutContext.WindowShortcut
sc = QShortcut(seq, self._parent, context=context) sc = QShortcut(seq, self._parent, context=context)

307
src/ipod_nano7_db.py
View File

@ -32,7 +32,7 @@ import sqlite3
import struct import struct
import subprocess import subprocess
import time import time
from typing import Any, Callable, Optional from typing import Any, Optional
from mutagen import File as MutagenFile from mutagen import File as MutagenFile
from mutagen.easyid3 import EasyID3 from mutagen.easyid3 import EasyID3
@ -118,10 +118,6 @@ def _import_iop(module_name: str):
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# iOpenPod parsers/writers log ERROR for transient I/O (USB resettling)
logging.getLogger("iTunesDB_Parser").setLevel(logging.CRITICAL)
logging.getLogger("iTunesDB_Writer").setLevel(logging.CRITICAL)
# ── constants ──────────────────────────────────────────────────────── # ── constants ────────────────────────────────────────────────────────
FILETYPE_CODES = { FILETYPE_CODES = {
"mp3": 0x4D503320, "mp3": 0x4D503320,
@ -210,16 +206,6 @@ def _read_audio_tags(filepath: str) -> dict[str, Any]:
ds = str(val[0]) ds = str(val[0])
if len(ds) >= 4: if len(ds) >= 4:
info["year"] = int(ds[:4]) info["year"] = int(ds[:4])
# Compilation flag (TCMP frame) — only set when explicitly present
try:
id3 = ID3(filepath)
if "TCMP" in id3:
tcmp = id3["TCMP"]
if hasattr(tcmp, "text") and tcmp.text:
info["compilation"] = str(tcmp.text[0]) == "1"
except Exception:
pass
except Exception: except Exception:
pass pass
elif ext in (".m4a", ".aac", ".mp4", ".m4p"): elif ext in (".m4a", ".aac", ".mp4", ".m4p"):
@ -242,14 +228,6 @@ def _read_audio_tags(filepath: str) -> dict[str, Any]:
tr = t["trkn"] tr = t["trkn"]
if isinstance(tr, list) and tr and isinstance(tr[0], tuple): if isinstance(tr, list) and tr and isinstance(tr[0], tuple):
info["track_number"] = tr[0][0] info["track_number"] = tr[0][0]
# Compilation flag (cpil atom)
if "cpil" in t:
cpil = t["cpil"]
if isinstance(cpil, bool):
info["compilation"] = cpil
elif isinstance(cpil, list) and len(cpil) > 0:
info["compilation"] = bool(cpil[0])
else: else:
if audio and audio.tags: if audio and audio.tags:
for tag_k, dest in [("title", "title"), ("artist", "artist"), for tag_k, dest in [("title", "title"), ("artist", "artist"),
@ -261,15 +239,6 @@ def _read_audio_tags(filepath: str) -> dict[str, Any]:
info[dest] = str(v[0]) if isinstance(v, list) and v else str(v) info[dest] = str(v[0]) if isinstance(v, list) and v else str(v)
except (ValueError, TypeError): except (ValueError, TypeError):
pass pass
# Compilation flag (vorbis comment)
try:
if "compilation" in audio.tags:
comp = audio.tags["compilation"]
comp_val = str(comp[0]) if isinstance(comp, list) else str(comp)
info["compilation"] = comp_val == "1"
except (ValueError, TypeError):
pass
except Exception: except Exception:
pass pass
return info return info
@ -360,7 +329,6 @@ def _scanned_tracks_to_iop(mountpoint: str,
sample_count=sample_count, sample_count=sample_count,
gapless_track_flag=1, gapless_track_flag=1,
gapless_album_flag=1 if has_album else 0, gapless_album_flag=1 if has_album else 0,
compilation_flag=s.get("compilation", False),
play_count=s.get("play_count", 0), play_count=s.get("play_count", 0),
rating=s.get("rating", 0), rating=s.get("rating", 0),
last_played=s.get("last_played", 0), last_played=s.get("last_played", 0),
@ -515,20 +483,17 @@ class Nano7Database:
location = f":iPod_Control:Music:{rel.replace(os.sep, ':')}" location = f":iPod_Control:Music:{rel.replace(os.sep, ':')}"
bs = base_stats.get(location, {}) bs = base_stats.get(location, {})
dd = delta_stats.get(location, {}) dd = delta_stats.get(location, {})
pc = bs.get("play_count", 0) pc = bs.get("play_count", 0) + dd.get("play_count", 0)
sc = bs.get("skip_count", 0) sc = bs.get("skip_count", 0) + dd.get("skip_count", 0)
dd_rt = dd.get("rating", 0) rating = dd.get("rating", 0) or bs.get("rating", 0)
rating = dd_rt if dd_rt >= 0 else bs.get("rating", 0)
tracks.append({ tracks.append({
"pid": s["pid"], "pid": s["pid"],
"title": s.get("title") or os.path.splitext(s["file_name"])[0], "title": s.get("title") or os.path.splitext(s["file_name"])[0],
"artist": s.get("artist") or "Unknown Artist", "artist": s.get("artist") or "Unknown Artist",
"album": s.get("album") or "", "album": s.get("album") or "",
"album_artist": s.get("album_artist") or "", "album_artist": s.get("album_artist") or "",
"genre": s.get("genre", ""),
"duration_ms": s.get("duration_ms", 0), "duration_ms": s.get("duration_ms", 0),
"track_number": s.get("track_number", 0), "track_number": s.get("track_number", 0),
"file_size": s.get("file_size", 0),
"disc_number": s.get("disc_number", 0), "disc_number": s.get("disc_number", 0),
"physical_order": 0, "physical_order": 0,
"date_modified": 0, "date_modified": 0,
@ -661,204 +626,46 @@ class Nano7Database:
deleted += 1 deleted += 1
return deleted return deleted
# ── playlist creation ────────────────────────────────────────────
def create_playlist(self, name: str, track_paths: list[str]) -> bool:
"""Create a user playlist on the iPod.
Two-pass matching: content hash first, then title+artist+album
fallback. Only tracks already present on the iPod will be
included in the playlist.
Args:
name: Playlist name.
track_paths: Local file paths of the tracks to include.
Returns:
True if the playlist was created, False otherwise.
"""
scanned = self._scan_ipod_files()
if not scanned:
logger.warning("No tracks on iPod — cannot create playlist")
return False
playlist_created = False
self._merge_play_stats(scanned)
def _build_pl_cb(scanned, tracks):
"""Build playlist using real db_track_id from TrackInfo."""
nonlocal playlist_created
try:
IopPlaylistInfo = _import_iop(
"iTunesDB_Writer.mhyp_writer"
).PlaylistInfo
except Exception as e:
logger.exception("Failed to import PlaylistInfo: %s", e)
return None
# Pass 1: content_hash → db_track_id
hash_to_db_id: dict[str, int] = {}
for i, s in enumerate(scanned):
ch = s.get("content_hash", "")
if ch:
hash_to_db_id[ch] = tracks[i].db_track_id
found_ids: list[int] = []
unmatched: list[str] = []
for path in track_paths:
if not os.path.exists(path):
continue
ch = content_hash(path)
db_id = hash_to_db_id.get(ch)
if db_id is not None:
found_ids.append(db_id)
else:
unmatched.append(path)
hash_matched_count = len(found_ids)
# Pass 2: metadata fallback
if unmatched:
scanned_meta = []
for i, s in enumerate(scanned):
scanned_meta.append((
s.get("title", "").lower(),
s.get("artist", "").lower(),
(s.get("album") or "").lower(),
tracks[i].db_track_id,
))
found_set = set(found_ids)
for path in unmatched:
tags = _read_audio_tags(path)
title = (tags.get("title") or "").lower()
artist = (tags.get("artist") or "").lower()
album = (tags.get("album") or "").lower()
if not title or not artist:
continue
for s_title, s_artist, s_album, db_id in scanned_meta:
if (db_id not in found_set
and s_title == title
and s_artist == artist
and s_album == album):
found_ids.append(db_id)
found_set.add(db_id)
break
if not found_ids:
logger.warning(
"No tracks from playlist '%s' found on iPod",
name,
)
return None
matched_by_hash = hash_matched_count
matched_by_meta = len(found_ids) - hash_matched_count
logger.info(
"Playlist '%s': %d track(s) matched (%d by hash, %d by metadata)",
name, len(found_ids), matched_by_hash, matched_by_meta,
)
playlist_created = True
return [IopPlaylistInfo(name=name, track_ids=found_ids)]
db_ok = self._write_databases_from_tracks(
scanned, build_extra_playlists=_build_pl_cb,
)
self._cleanup_play_counts_file()
return db_ok and playlist_created
# ── sync: regenerate everything via iOpenPod ────────────────────── # ── sync: regenerate everything via iOpenPod ──────────────────────
def sync_itunescdb(self) -> bool: def sync_itunescdb(self) -> None:
"""Regenerate ALL databases using iOpenPod's writers. """Regenerate ALL databases using iOpenPod's writers."""
Returns True if the databases were written successfully.
"""
try: try:
self._sync_itunescdb_impl() self._sync_itunescdb_impl()
return True
except Exception as e: except Exception as e:
logger.exception("iTunesDB sync failed: %s", e) logger.exception("iTunesDB sync failed: %s", e)
return False
def _merge_play_stats(self, scanned: list[dict]) -> None: def _merge_play_stats(self, scanned: list[dict]) -> None:
"""Merge iPod + local play stats into *scanned* track dicts in-place. """Merge iPod + local play stats into *scanned* track dicts in-place.
Uses additive model (like iTunes): Uses additive model (like iTunes):
``new_value = max(ipod_base, local_cache, play_stats_store) + ipod_delta`` ``new_value = max(ipod_base, local_count) + ipod_delta``
Includes a **validation gate** that refuses to produce all-zero
play counts when we have previously-recorded non-zero data in the
local PlayStatsStore — this prevents accidental data loss from
silent iTunesDB parse failures.
""" """
base_stats, delta_stats = self.read_play_stats() base_stats, delta_stats = self.read_play_stats()
local_play_stats = self._load_local_play_stats() local_play_stats = self._load_local_play_stats()
# Supplement local_play_stats with durable PlayStatsStore (SQLite)
from library_cache import get_library_cache
db_store = get_library_cache()
for s in scanned: for s in scanned:
rel = os.path.relpath(s["file_path"], self.music_base) rel = os.path.relpath(s["file_path"], self.music_base)
location = f":iPod_Control:Music:{rel.replace(os.sep, ':')}" location = f":iPod_Control:Music:{rel.replace(os.sep, ':')}"
bs = base_stats.get(location, {}) bs = base_stats.get(location, {})
dd = delta_stats.get(location, {}) dd = delta_stats.get(location, {})
ch = s.get("content_hash", "") ch = s.get("content_hash", "")
# Three-tier fallback: iTunesDB → LibraryCache (SQLite) → old JSON cache
local_pc = local_play_stats.get(ch, {}) local_pc = local_play_stats.get(ch, {})
db_entry = db_store.get_by_content_hash(ch) if ch else None
s["play_count"] = max( s["play_count"] = max(
bs.get("play_count", 0), bs.get("play_count", 0), local_pc.get("play_count", 0),
(db_entry or {}).get("play_count", 0), ) + dd.get("play_count", 0)
local_pc.get("play_count", 0),
)
s["skip_count"] = max( s["skip_count"] = max(
bs.get("skip_count", 0), bs.get("skip_count", 0), local_pc.get("skip_count", 0),
(db_entry or {}).get("skip_count", 0), ) + dd.get("skip_count", 0)
local_pc.get("skip_count", 0),
)
dd_rating = dd.get("rating", 0) s["rating"] = dd.get("rating", 0) or bs.get("rating", 0) or local_pc.get("rating", 0)
store_rating = (db_entry or {}).get("rating", 0)
s["rating"] = dd_rating if dd_rating >= 0 else (
bs.get("rating", 0) or store_rating or local_pc.get("rating", 0)
)
s["last_played"] = max( s["last_played"] = max(
bs.get("last_played", 0), bs.get("last_played", 0), local_pc.get("last_played", 0),
(db_entry or {}).get("last_played", 0),
local_pc.get("last_played", 0),
) )
# ── validation gate ────────────────────────────────────────────
# If every scanned track would get play_count=0 but the local DB
# previously had non-zero values, restore them.
all_zero = all(s.get("play_count", 0) == 0 for s in scanned)
if all_zero:
restored = 0
for s in scanned:
ch = s.get("content_hash", "")
if not ch:
continue
db_entry = db_store.get_by_content_hash(ch) if ch else None
stored_pc = (db_entry or {}).get("play_count", 0)
if stored_pc > 0:
s["play_count"] = stored_pc
restored += 1
if restored:
logger.warning(
"Validation gate triggered — restored play_count for %d track(s) "
"from local database after merge produced all zeros",
restored,
)
# ───────────────────────────────────────────────────────────────
def _cleanup_play_counts_file(self) -> None: def _cleanup_play_counts_file(self) -> None:
"""Delete the iPod 'Play Counts' delta file after a full sync.""" """Delete the iPod 'Play Counts' delta file after a full sync."""
pc_file = os.path.join( pc_file = os.path.join(
@ -871,6 +678,12 @@ class Nano7Database:
except OSError as e: except OSError as e:
logger.debug("Could not delete Play Counts file: %s", e) logger.debug("Could not delete Play Counts file: %s", e)
def consume_play_deltas(self) -> None:
"""Public entry point: delete the Play Counts file after consuming
its data during mount. Prevents double-counting on subsequent
``read_play_stats()`` calls."""
self._cleanup_play_counts_file()
def _sync_itunescdb_impl(self) -> None: def _sync_itunescdb_impl(self) -> None:
scanned = self._scan_ipod_files() scanned = self._scan_ipod_files()
if not scanned: if not scanned:
@ -879,9 +692,7 @@ class Nano7Database:
return return
self._merge_play_stats(scanned) self._merge_play_stats(scanned)
if not self._write_databases_from_tracks(scanned): self._write_databases_from_tracks(scanned)
logger.error("iTunesDB sync failed — play counts not cleaned up")
return
self._cleanup_play_counts_file() self._cleanup_play_counts_file()
def find_track_by_content_hash(self, content_hash: str) -> dict | None: def find_track_by_content_hash(self, content_hash: str) -> dict | None:
@ -943,7 +754,7 @@ class Nano7Database:
tracks = data.get("mhlt", []) tracks = data.get("mhlt", [])
for track in tracks: for track in tracks:
loc = track.get("Location") or track.get("location", "") loc = track.get("location", "")
if not loc: if not loc:
continue continue
base_stats[loc] = { base_stats[loc] = {
@ -962,7 +773,7 @@ class Nano7Database:
for i, entry in enumerate(entries): for i, entry in enumerate(entries):
if i >= len(tracks): if i >= len(tracks):
break break
loc = tracks[i].get("Location") or tracks[i].get("location", "") loc = tracks[i].get("location", "")
if not loc: if not loc:
continue continue
delta_stats[loc] = { delta_stats[loc] = {
@ -979,41 +790,6 @@ class Nano7Database:
) )
return base_stats, delta_stats return base_stats, delta_stats
def read_play_counts_delta(self) -> dict[str, dict[str, int]]:
"""Read Play Counts file using filesystem scan (no iTunesDB needed).
Positional mapping uses ``_scan_ipod_files()`` order (correct after
any sync written by this app). Returns ``{location: {play_count,
skip_count, rating}}`` or empty dict on failure.
"""
pc_path = os.path.join(
self.mountpoint, "iPod_Control", "iTunes", "Play Counts",
)
if not os.path.exists(pc_path):
return {}
pc_mod = _import_iop("iTunesDB_Parser.playcounts")
try:
entries = pc_mod.parse_playcounts(pc_path)
except Exception:
return {}
if not entries:
return {}
scanned = self._scan_ipod_files()
delta: dict[str, dict[str, int]] = {}
for i, entry in enumerate(entries):
if i >= len(scanned):
break
if entry.play_count == 0 and entry.skip_count == 0 and entry.rating < 0:
continue
rel = os.path.relpath(scanned[i]["file_path"], self.music_base)
loc = f":iPod_Control:Music:{rel.replace(os.sep, ':')}"
delta[loc] = {
"play_count": entry.play_count,
"skip_count": entry.skip_count,
"rating": entry.rating,
}
return delta
@staticmethod @staticmethod
def _load_local_play_stats() -> dict[str, dict[str, int]]: def _load_local_play_stats() -> dict[str, dict[str, int]]:
"""Read accumulated play stats from the local library cache. """Read accumulated play stats from the local library cache.
@ -1024,8 +800,8 @@ class Nano7Database:
from library_cache import get_library_cache from library_cache import get_library_cache
cache = get_library_cache() cache = get_library_cache()
result: dict[str, dict[str, int]] = {} result: dict[str, dict[str, int]] = {}
for _path, entry in cache.iter_entries(): for entry in cache._data.values():
ch = entry.get("content_hash", "") ch = entry.get("_content_hash", "")
pc = entry.get("play_count", 0) pc = entry.get("play_count", 0)
rating = entry.get("rating", 0) rating = entry.get("rating", 0)
lp = entry.get("last_played", 0) lp = entry.get("last_played", 0)
@ -1123,9 +899,7 @@ class Nano7Database:
self, self,
scanned: list[dict], scanned: list[dict],
artwork_overrides: dict[str, object] | None = None, artwork_overrides: dict[str, object] | None = None,
extra_playlists: list | None = None, ) -> None:
build_extra_playlists: Callable | None = None,
) -> bool:
"""Build artwork, convert to TrackInfo, write iTunesDB + SQLite. """Build artwork, convert to TrackInfo, write iTunesDB + SQLite.
Args: Args:
@ -1135,15 +909,6 @@ class Nano7Database:
artwork should come from a local file instead of the iPod file. artwork should come from a local file instead of the iPod file.
Use the ``_REMOVE_ARTWORK`` sentinel to explicitly remove Use the ``_REMOVE_ARTWORK`` sentinel to explicitly remove
artwork for a track. Omit a key to keep existing iPod artwork. artwork for a track. Omit a key to keep existing iPod artwork.
extra_playlists: Pre-built list of PlaylistInfo (ignored when
build_extra_playlists is provided).
build_extra_playlists: Optional callable(scanned, tracks) → list
of PlaylistInfo. Called after TrackInfo conversion and
db_track_id generation, so the callback can use real
db_track_id values from TrackInfo objects.
Returns:
True if databases were written successfully.
""" """
artwork_overrides = artwork_overrides or {} artwork_overrides = artwork_overrides or {}
# 1. Prepare artwork for all tracks # 1. Prepare artwork for all tracks
@ -1203,20 +968,6 @@ class Nano7Database:
tracks = _scanned_tracks_to_iop(self.mountpoint, scanned, artwork_map) tracks = _scanned_tracks_to_iop(self.mountpoint, scanned, artwork_map)
logger.info("Built %d iOpenPod TrackInfo objects", len(tracks)) logger.info("Built %d iOpenPod TrackInfo objects", len(tracks))
# 2b. Pre-generate db_track_id for all tracks so callbacks can use them
generate_db_track_id = _import_iop("iTunesDB_Writer.mhit_writer").generate_db_track_id
for track in tracks:
if track.db_track_id == 0:
track.db_track_id = generate_db_track_id()
# 2c. Build extra playlists via callback if provided
if build_extra_playlists is not None:
result = build_extra_playlists(scanned, tracks)
if result is not None:
extra_playlists = result
else:
extra_playlists = None
# 3. Determine capabilities for this device # 3. Determine capabilities for this device
ipod_device = _import_iop("ipod_device") ipod_device = _import_iop("ipod_device")
capabilities = None capabilities = None
@ -1245,11 +996,10 @@ class Nano7Database:
backup=True, backup=True,
capabilities=capabilities, capabilities=capabilities,
master_playlist_name="iPod", master_playlist_name="iPod",
playlists=extra_playlists or None,
) )
if not ok: if not ok:
logger.error("iTunesDB write failed") logger.error("iTunesDB write failed")
return False return
logger.info("iTunesDB written (%d tracks)", len(tracks)) logger.info("iTunesDB written (%d tracks)", len(tracks))
# 5. Extract db_pid from the binary DB for SQLite cross-reference # 5. Extract db_pid from the binary DB for SQLite cross-reference
@ -1277,18 +1027,13 @@ class Nano7Database:
capabilities=capabilities, capabilities=capabilities,
firewire_id=firewire_id, firewire_id=firewire_id,
backup=True, backup=True,
playlists=extra_playlists or None,
) )
if sqlite_ok: if sqlite_ok:
logger.info("SQLite databases written") logger.info("SQLite databases written")
else: else:
logger.error("SQLite database write failed") logger.error("SQLite database write failed")
return False
except Exception as e: except Exception as e:
logger.exception("SQLite write error: %s", e) logger.exception("SQLite write error: %s", e)
return False
return True
def _write_empty_databases(self) -> None: def _write_empty_databases(self) -> None:
"""Write minimal empty databases when no tracks exist.""" """Write minimal empty databases when no tracks exist."""

543
src/library_cache.py
View File

@ -1,83 +1,25 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
Library metadata cache for neo-pod-desktop. Library metadata cache for neo-pod-desktop.
Caches extracted tag data per file path with mtime-based invalidation.
Stored as SQLite-format database in ``~/.local/share/neo-pod-desktop/library_cache.db``. Stored as JSON in XDG cache directory.
Each row is keyed by its **file path** and also carries a stable **UUID** so
play-count / rating / last-played data survives file renames, moves, and
re-encodings. The database is thread-safe (WAL + busy_timeout) and
automatically rotated to a backup file on every write.
Migration from the old JSON file (``~/.cache/neo-pod-desktop/library_cache.json``)
happens automatically on the first access.
""" """
import hashlib import hashlib
import json import json
import logging import logging
import os import os
import sqlite3 from typing import Any, Dict, Optional
import threading
import time
import uuid
from typing import Any, Iterator
from xdg_base import xdg_data_path from artwork.cache import xdg_cache_path
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
_DB_FILE = "library_cache.db" _CACHE_FILE = "library_cache.json"
_BACKUP_FILE = "library_cache.db.backup"
_OLD_JSON_FILE = "library_cache.json"
_SCHEMA_VERSION = 2
_SCHEMA = """
CREATE TABLE IF NOT EXISTS library_cache (
path TEXT PRIMARY KEY,
uuid TEXT UNIQUE NOT NULL,
title TEXT DEFAULT '',
artist TEXT DEFAULT '',
album TEXT DEFAULT '',
album_artist TEXT DEFAULT '',
genre TEXT DEFAULT '',
duration_s REAL DEFAULT 0,
size INTEGER DEFAULT 0,
track_num INTEGER DEFAULT 0,
source_type TEXT DEFAULT '',
bitrate INTEGER DEFAULT 0,
sample_rate INTEGER DEFAULT 44100,
cover_path TEXT DEFAULT '',
thumbnail_url TEXT DEFAULT '',
play_count INTEGER DEFAULT 0,
rating INTEGER DEFAULT 0,
last_played INTEGER DEFAULT 0,
skip_count INTEGER DEFAULT 0,
mtime REAL NOT NULL DEFAULT 0,
content_hash TEXT NOT NULL DEFAULT '',
updated_at INTEGER DEFAULT (cast(strftime('%s','now') as int))
);
CREATE INDEX IF NOT EXISTS idx_lc_uuid ON library_cache(uuid);
CREATE INDEX IF NOT EXISTS idx_lc_hash ON library_cache(content_hash);
CREATE INDEX IF NOT EXISTS idx_lc_artist ON library_cache(artist);
CREATE INDEX IF NOT EXISTS idx_lc_album ON library_cache(album);
CREATE TABLE IF NOT EXISTS playcounts_log (
firewire_id TEXT PRIMARY KEY,
last_synced_delta TEXT NOT NULL DEFAULT '{}'
);
"""
# ---------------------------------------------------------------------------
# Fast content fingerprint
# ---------------------------------------------------------------------------
def _content_hash(filepath: str, max_bytes: int = 65536) -> str: def _content_hash(filepath: str, max_bytes: int = 65536) -> str:
"""SHA1 of first *max_bytes* of audio data.""" """SHA1 of first max_bytes of audio data (fast content fingerprint)."""
h = hashlib.sha1() h = hashlib.sha1()
try: try:
with open(filepath, "rb") as f: with open(filepath, "rb") as f:
@ -87,19 +29,96 @@ def _content_hash(filepath: str, max_bytes: int = 65536) -> str:
return h.hexdigest() return h.hexdigest()
# --------------------------------------------------------------------------- class LibraryCache:
# Singleton helpers (kept for backward compatibility) """JSON-based cache of audio file metadata.
# ---------------------------------------------------------------------------
_library_cache: "LibraryCache | None" = None Keys are absolute file paths. Values are dicts of metadata
_migrated_flag = False plus internal ``_mtime`` and ``_content_hash`` fields for
change detection and track matching.
Cached entries are returned only when the file still exists
and its ``os.path.getmtime`` matches the stored value.
"""
def __init__(self, cache_path: Optional[str] = None):
if cache_path is None:
cache_path = os.path.join(xdg_cache_path(""), _CACHE_FILE)
self._path = cache_path
self._data: Dict[str, Dict[str, Any]] = {}
def load(self) -> None:
self._data = {}
if not os.path.exists(self._path):
return
try:
with open(self._path, "r", encoding="utf-8") as fh:
self._data = json.load(fh)
except (json.JSONDecodeError, OSError) as exc:
logger.warning("Failed to load library cache: %s", exc)
self._data = {}
def save(self) -> None:
self._data = {k: v for k, v in self._data.items()
if os.path.exists(k)}
os.makedirs(os.path.dirname(self._path), exist_ok=True)
try:
with open(self._path, "w", encoding="utf-8") as fh:
json.dump(self._data, fh, ensure_ascii=False)
except OSError as exc:
logger.warning("Failed to save library cache: %s", exc)
def get(self, file_path: str) -> Optional[Dict[str, Any]]:
entry = self._data.get(file_path)
if entry is None:
return None
if not os.path.isfile(file_path):
self._data.pop(file_path, None)
return None
actual_mtime = os.path.getmtime(file_path)
if actual_mtime != entry.get("_mtime", 0):
return None
result = {}
for k, v in entry.items():
if k == "_mtime":
continue
if k == "_content_hash":
result["content_hash"] = v
else:
result[k] = v
return result
def put(self, file_path: str, data: Dict[str, Any]) -> None:
entry = dict(data)
try:
entry["_mtime"] = os.path.getmtime(file_path)
except OSError:
return
entry["_content_hash"] = _content_hash(file_path)
self._data[file_path] = entry
def remove(self, file_path: str) -> None:
self._data.pop(file_path, None)
def clear(self) -> None:
self._data.clear()
if os.path.exists(self._path):
try:
os.remove(self._path)
except OSError:
pass
def __len__(self) -> int:
return len(self._data)
def get_library_cache() -> "LibraryCache": _library_cache: Optional[LibraryCache] = None
def get_library_cache() -> LibraryCache:
global _library_cache global _library_cache
if _library_cache is None: if _library_cache is None:
_library_cache = LibraryCache() _library_cache = LibraryCache()
_library_cache._migrate_old() _library_cache.load()
return _library_cache return _library_cache
@ -112,385 +131,3 @@ def save_library_cache() -> None:
def invalidate_cache_for_path(file_path: str) -> None: def invalidate_cache_for_path(file_path: str) -> None:
cache = get_library_cache() cache = get_library_cache()
cache.remove(file_path) cache.remove(file_path)
# ---------------------------------------------------------------------------
# SQLite-backed cache
# ---------------------------------------------------------------------------
class LibraryCache:
"""Persistent cache backed by a single SQLite database.
Public API (unchanged from the JSON version):
get(path) → dict | None
put(path, data) → None
remove(path) → None
clear() → None
save() → None
__len__() → int
New public API:
iter_entries() → Iterator[(path, dict)]
get_by_content_hash(chash) → dict | None
get_by_uuid(uid) → dict | None
update_play_count(path) → int (increment + return)
"""
def __init__(self, db_path: str | None = None):
if db_path is None:
db_path = os.path.join(xdg_data_path(), _DB_FILE)
self._path = db_path
self._lock = threading.RLock()
self.conn = sqlite3.connect(
db_path, check_same_thread=False,
detect_types=sqlite3.PARSE_DECLTYPES,
)
self.conn.row_factory = sqlite3.Row
self.conn.execute("PRAGMA journal_mode=WAL")
self.conn.execute("PRAGMA busy_timeout=5000")
self.conn.execute("PRAGMA synchronous=NORMAL")
self.conn.execute("PRAGMA foreign_keys=OFF")
self._init_db()
def _init_db(self) -> None:
self._migrate_schema()
self.conn.executescript(_SCHEMA)
self.conn.execute(f"PRAGMA user_version = {_SCHEMA_VERSION}")
self.conn.commit()
def _migrate_schema(self) -> None:
cur_ver = self.conn.execute("PRAGMA user_version").fetchone()[0]
if cur_ver >= 2:
return
if cur_ver == 1:
self.conn.execute("DROP TABLE IF EXISTS playcounts_log")
# ── public read API ────────────────────────────────────────────────
def get(self, file_path: str) -> dict[str, Any] | None:
"""Return cached data for *file_path*, or None if missing or stale."""
with self._lock:
row = self.conn.execute(
"SELECT * FROM library_cache WHERE path=?",
(file_path,),
).fetchone()
if row is None:
return None
if not os.path.isfile(file_path):
return None
actual_mtime = os.path.getmtime(file_path)
if actual_mtime != row["mtime"]:
return None
return self._row_to_dict(row)
def get_by_content_hash(self, content_hash: str) -> dict[str, Any] | None:
"""Look up a track by its content fingerprint."""
with self._lock:
row = self.conn.execute(
"SELECT * FROM library_cache WHERE content_hash=?",
(content_hash,),
).fetchone()
if row is None:
return None
return self._row_to_dict(row)
def get_by_uuid(self, track_uuid: str) -> dict[str, Any] | None:
"""Look up a track by its stable UUID."""
with self._lock:
row = self.conn.execute(
"SELECT * FROM library_cache WHERE uuid=?",
(track_uuid,),
).fetchone()
if row is None:
return None
return self._row_to_dict(row)
def iter_entries(self) -> Iterator[tuple[str, dict[str, Any]]]:
"""Yield (file_path, data_dict) for every row in the database."""
with self._lock:
rows = self.conn.execute(
"SELECT * FROM library_cache"
).fetchall()
for row in rows:
yield (row["path"], self._row_to_dict(row))
def __len__(self) -> int:
with self._lock:
return self.conn.execute(
"SELECT COUNT(*) FROM library_cache"
).fetchone()[0]
# ── public write API ───────────────────────────────────────────────
def put(self, file_path: str, data: dict[str, Any]) -> None:
"""Insert or replace a cache entry for *file_path*."""
try:
mtime = os.path.getmtime(file_path)
except OSError:
return
track_uuid = data.get("_track_uuid") or data.get("uuid") or str(uuid.uuid4())
ch = data.get("_content_hash") or data.get("content_hash") or _content_hash(file_path)
with self._lock:
self.conn.execute(
"""INSERT OR REPLACE INTO library_cache
(path, uuid, title, artist, album, album_artist,
genre, duration_s, size, track_num, source_type,
bitrate, sample_rate, cover_path, thumbnail_url,
play_count, rating, last_played, skip_count,
mtime, content_hash, updated_at)
VALUES (?,?,?,?,?,?,
?,?,?,?,?,
?,?,?,?,
?,?,?,?,
?,?,cast(strftime('%s','now') as int))""",
(
file_path,
track_uuid,
data.get("title", ""),
data.get("artist", ""),
data.get("album", ""),
data.get("album_artist", ""),
data.get("genre", ""),
data.get("duration_s", 0) or data.get("duration", 0),
data.get("size", 0),
data.get("track_num", 0) or data.get("track_number", 0),
data.get("source_type", ""),
data.get("bitrate", 0),
data.get("sample_rate", 44100),
data.get("cover_path", ""),
data.get("thumbnail_url", ""),
data.get("play_count", 0),
data.get("rating", 0),
data.get("last_played", 0),
data.get("skip_count", 0),
mtime,
ch,
),
)
self._backup()
def update_play_stats(self, file_path: str,
play_count: int | None = None,
rating: int | None = None,
last_played: int | None = None,
skip_count: int | None = None) -> None:
"""Update only the play-stat fields for *file_path*."""
fields = []
vals: list[Any] = []
for k, v in [("play_count", play_count), ("rating", rating),
("last_played", last_played), ("skip_count", skip_count)]:
if v is not None:
fields.append(f"{k}=?")
vals.append(v)
if not fields:
return
fields.append("updated_at=cast(strftime('%s','now') as int)")
with self._lock:
self.conn.execute(
f"UPDATE library_cache SET {', '.join(fields)} WHERE path=?",
(*vals, file_path),
)
self._backup()
def remove(self, file_path: str) -> None:
with self._lock:
self.conn.execute(
"DELETE FROM library_cache WHERE path=?", (file_path,),
)
self._backup()
def clear(self) -> None:
with self._lock:
self.conn.execute("DELETE FROM library_cache")
self._backup()
def save(self) -> None:
"""Explicit commit + backup."""
with self._lock:
self.conn.commit()
self._backup()
# ── play-counts delta deduplication ──────────────────────────────────
def get_last_synced_delta(self, firewire_id: str) -> dict[str, dict[str, int]]:
"""Return the delta dict that was last committed for this iPod.
Returns an empty dict if no sync has happened yet.
"""
with self._lock:
row = self.conn.execute(
"SELECT last_synced_delta FROM playcounts_log WHERE firewire_id=?",
(firewire_id,),
).fetchone()
if row and row["last_synced_delta"]:
try:
return json.loads(row["last_synced_delta"])
except json.JSONDecodeError:
pass
return {}
def set_last_synced_delta(self, firewire_id: str,
delta: dict[str, dict[str, int]]) -> None:
"""Store the delta that was committed during the last sync."""
blob = json.dumps(delta, sort_keys=True, separators=(",", ":"))
with self._lock:
self.conn.execute(
"INSERT OR REPLACE INTO playcounts_log (firewire_id, last_synced_delta) "
"VALUES (?, ?)",
(firewire_id, blob),
)
self._backup()
# ── backup rotation ────────────────────────────────────────────────
def _backup(self) -> None:
"""Rotate database backup files after each write.
Commits the current transaction first so the backup connection
can checkpoint the WAL without blocking.
Keeps two historical copies:
library_cache.db ← current
library_cache.db.backup ← previous version
library_cache.db.bak ← version before that
"""
# Commit so the second connection can safely checkpoint.
self.conn.commit()
db = self._path
b1 = db + ".backup"
b2 = db + ".bak"
try:
bak_conn = sqlite3.connect(db)
bak_conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
bak_conn.close()
except Exception as exc:
logger.debug("Checkpoint failed: %s", exc)
if os.path.exists(b1):
try:
os.replace(b1, b2)
except OSError:
pass
try:
import shutil
if os.path.exists(db):
shutil.copy2(db, b1)
except OSError as exc:
logger.debug("Backup rotation failed: %s", exc)
# ── internal helpers ───────────────────────────────────────────────
@staticmethod
def _row_to_dict(row: sqlite3.Row) -> dict[str, Any]:
keys = row.keys()
d = dict(zip(keys, row))
d.pop("updated_at", None)
alias = {
"duration_s": "duration",
"track_num": "track_number",
}
for old_k, new_k in alias.items():
if old_k in d and new_k not in d:
d[new_k] = d[old_k]
return d
# ── JSON migration legacy→SQLite ──────────────────────────────────
def _migrate_old(self) -> None:
"""One-shot migration from the old JSON cache file."""
global _migrated_flag
if _migrated_flag:
return
old_dir = xdg_data_path()
old_path = os.path.join(
os.path.dirname(old_dir), # parent: neo-pod-desktop
".cache", "neo-pod-desktop", _OLD_JSON_FILE,
)
# also try the actual XDG cache dir
from xdg_base import xdg_cache_path
alt_path = os.path.join(xdg_cache_path(), _OLD_JSON_FILE)
for candidate in (old_path, alt_path):
if os.path.exists(candidate):
self._do_migrate_json(candidate)
break
_migrated_flag = True
def _do_migrate_json(self, json_path: str) -> None:
"""Read old JSON and insert all entries."""
try:
with open(json_path, "r", encoding="utf-8") as f:
old_data = json.load(f)
except (json.JSONDecodeError, OSError) as exc:
logger.warning("JSON migration source corrupt: %s", exc)
return
count = 0
for file_path, entry in old_data.items():
if not isinstance(entry, dict):
continue
mtime = entry.get("_mtime", 0)
if not mtime:
try:
mtime = os.path.getmtime(file_path)
except OSError:
continue
entry["_track_uuid"] = entry.get("_track_uuid") or str(uuid.uuid4())
entry["_content_hash"] = entry.get("_content_hash") or _content_hash(file_path)
with self._lock:
self.conn.execute(
"""INSERT OR IGNORE INTO library_cache
(path, uuid, title, artist, album, album_artist,
genre, duration_s, size, track_num, source_type,
bitrate, sample_rate, cover_path, thumbnail_url,
play_count, rating, last_played, skip_count,
mtime, content_hash, updated_at)
VALUES (?,?,?,?,?,?,
?,?,?,?,?,
?,?,?,?,
?,?,?,?,
?,?,cast(strftime('%s','now') as int))""",
(
file_path,
entry.get("_track_uuid", ""),
entry.get("title", ""),
entry.get("artist", ""),
entry.get("album", ""),
entry.get("album_artist", ""),
entry.get("genre", ""),
entry.get("duration_s", 0) or entry.get("duration", 0),
entry.get("size", 0),
entry.get("track_num", 0) or entry.get("track_number", 0),
entry.get("source_type", ""),
entry.get("bitrate", 0),
entry.get("sample_rate", 44100),
entry.get("cover_path", ""),
entry.get("thumbnail_url", ""),
entry.get("play_count", 0),
entry.get("rating", 0),
entry.get("last_played", 0),
entry.get("skip_count", 0),
mtime,
entry.get("_content_hash", ""),
),
)
count += 1
self.conn.commit()
self._backup()
# Rename old file so we don't re-migrate
try:
os.rename(json_path, json_path + ".migrated")
logger.info(
"Migrated %d entries from %s → SQLite",
count, json_path,
)
except OSError:
pass

32
src/metadata_handler.py
View File

@ -17,7 +17,7 @@ import requests
from PIL import Image from PIL import Image
from mutagen import File as MutagenFile from mutagen import File as MutagenFile
from mutagen.mp4 import MP4, MP4Cover from mutagen.mp4 import MP4, MP4Cover
from mutagen.id3 import ID3, APIC, POPM, TIT2, TPE1, TALB, TRCK from mutagen.id3 import ID3, APIC, TIT2, TPE1, TALB, TRCK
from mutagen.easyid3 import EasyID3 from mutagen.easyid3 import EasyID3
from track_info import TrackInfo from track_info import TrackInfo
@ -552,8 +552,7 @@ class MetadataHandler:
file_path: Path to audio file file_path: Path to audio file
tags: Dict of tag_name -> value tags: Dict of tag_name -> value
Supported keys: title, artist, album, album_artist, genre, Supported keys: title, artist, album, album_artist, genre,
date, track_number, track_total, disc_number, disc_total, date, track_number, track_total, disc_number, disc_total, comment
comment, rating (0‑5 stars)
new_cover_path: Path to new cover image, None to keep existing, new_cover_path: Path to new cover image, None to keep existing,
empty string '' to remove cover art empty string '' to remove cover art
@ -638,13 +637,6 @@ class MetadataHandler:
elif 'disk' in audio: elif 'disk' in audio:
del audio['disk'] del audio['disk']
if 'rating' in tags:
val = tags['rating'] * 20 # stars → 0-100
if val > 0:
audio['\xa9rt'] = [val]
elif '\xa9rt' in audio:
del audio['\xa9rt']
if new_cover_path == '': if new_cover_path == '':
if 'covr' in audio: if 'covr' in audio:
del audio['covr'] del audio['covr']
@ -712,16 +704,6 @@ class MetadataHandler:
elif 'discnumber' in audio: elif 'discnumber' in audio:
del audio['discnumber'] del audio['discnumber']
if 'rating' in tags:
stars = tags['rating']
popm_val = [0, 64, 128, 192, 224, 255][min(stars, 5)]
audio = ID3(file_path)
audio.delall("POPM")
if popm_val > 0:
audio.add(POPM(email="", rating=popm_val))
audio.save(file_path)
audio = EasyID3(file_path)
audio.save() audio.save()
if new_cover_path is not None: if new_cover_path is not None:
@ -816,16 +798,6 @@ class MetadataHandler:
except (KeyError, TypeError): except (KeyError, TypeError):
pass pass
if 'rating' in tags:
val = str(tags['rating'] * 20) # stars → "0" … "100"
if int(val) > 0:
audio.tags['RATING'] = val
elif 'RATING' in audio.tags:
try:
del audio.tags['RATING']
except (KeyError, TypeError):
pass
if new_cover_path is not None: if new_cover_path is not None:
self._clear_cover_vorbis(audio) self._clear_cover_vorbis(audio)
if new_cover_path != '': if new_cover_path != '':

20
src/playlist_manager.py
View File

@ -5,11 +5,11 @@ import time
import logging import logging
from typing import List, Dict, Optional from typing import List, Dict, Optional
from xdg_base import xdg_data_path
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
_PLAYLIST_FILE = os.path.join(xdg_data_path(), "playlists.json") _PLAYLIST_FILE = os.path.join(
os.path.expanduser("~"), ".config", "neo-pod-desktop", "playlists.json"
)
class Playlist: class Playlist:
@ -47,22 +47,8 @@ class PlaylistManager:
def __init__(self, filepath: str = _PLAYLIST_FILE): def __init__(self, filepath: str = _PLAYLIST_FILE):
self._filepath = filepath self._filepath = filepath
self._playlists: List[Playlist] = [] self._playlists: List[Playlist] = []
self._migrate_old()
self._load() self._load()
def _migrate_old(self):
"""Migrate from old ~/.config/ location to ~/.local/share/."""
old = os.path.join(
os.path.expanduser("~"), ".config", "neo-pod-desktop", "playlists.json",
)
if os.path.exists(old) and not os.path.exists(self._filepath):
try:
os.makedirs(os.path.dirname(self._filepath), exist_ok=True)
os.rename(old, self._filepath)
logger.info("Migrated playlists from %s → %s", old, self._filepath)
except OSError as exc:
logger.warning("Playlist migration failed: %s", exc)
def _load(self): def _load(self):
try: try:
os.makedirs(os.path.dirname(self._filepath), exist_ok=True) os.makedirs(os.path.dirname(self._filepath), exist_ok=True)

88
src/session_store.py
View File

@ -1,88 +0,0 @@
#!/usr/bin/env python3
"""
Session state store — persistent playback state.
Stored as JSON in ``~/.local/share/neo-pod-desktop/session.json``.
Replaces the ``[Playback]`` section of config.ini — session state
(user's last track, position, volume, playback mode) is user data,
not configuration.
"""
import json
import logging
import os
from typing import Any
from xdg_base import xdg_data_path
logger = logging.getLogger(__name__)
_SESSION_FILE = "session.json"
class SessionStore:
def __init__(self, path: str | None = None):
self._path = path or os.path.join(xdg_data_path(), _SESSION_FILE)
self._data: dict[str, Any] = {}
self._load()
# ── load / save ────────────────────────────────────────────────────
def _load(self) -> None:
if not os.path.exists(self._path):
return
try:
with open(self._path, "r", encoding="utf-8") as f:
self._data = json.load(f)
except (json.JSONDecodeError, OSError) as exc:
logger.warning("Failed to load session: %s", exc)
self._data = {}
def save(self) -> None:
tmp = self._path + ".tmp"
try:
os.makedirs(os.path.dirname(self._path), exist_ok=True)
with open(tmp, "w", encoding="utf-8") as f:
json.dump(self._data, f, ensure_ascii=False)
os.replace(tmp, self._path)
except OSError as exc:
logger.warning("Failed to save session: %s", exc)
# ── typed accessors ────────────────────────────────────────────────
def get_str(self, key: str, fallback: str = "") -> str:
return str(self._data.get(key, fallback))
def get_int(self, key: str, fallback: int = 0) -> int:
return int(self._data.get(key, fallback))
def get_bool(self, key: str, fallback: bool = False) -> bool:
v = self._data.get(key, fallback)
if isinstance(v, bool):
return v
return str(v).lower() in ("true", "1", "yes")
def set(self, key: str, value: Any) -> None:
self._data[key] = value
# ── migrations ─────────────────────────────────────────────────────
def migrate_from_config(self, config_loader: Any) -> None:
"""Import ``[Playback]`` section from the old config loader."""
keys = [
"last_track_path", "last_position_ms", "last_volume",
"last_playing", "shuffle_active", "repeat_mode",
]
changed = False
for k in keys:
if k not in self._data:
v = config_loader.get("Playback", k, fallback="")
self.set(k, v)
changed = True
if changed:
self.save()
# Wipe old Playback section so it isn't read again.
if config_loader.config.has_section("Playback"):
config_loader.config.remove_section("Playback")
config_loader.save()
logger.info("Migrated playback state from config.ini to session.json")

790
src/ui/ipod_tab.py
View File

File diff suppressed because it is too large Load Diff

283
src/ui/library_tab.py
View File

@ -8,8 +8,7 @@ import os
import sys import sys
import time import time
import logging import logging
import shutil from typing import List, Dict, Optional, Tuple, Any
from typing import List, Dict, Optional, Tuple, Any, Callable
from pathlib import Path from pathlib import Path
from PyQt6.QtWidgets import ( from PyQt6.QtWidgets import (
@ -18,8 +17,8 @@ from PyQt6.QtWidgets import (
QTableWidget, QTableWidgetItem, QSlider, QMenu, QInputDialog, QTableWidget, QTableWidgetItem, QSlider, QMenu, QInputDialog,
QStackedWidget, QListWidget, QListWidgetItem, QListView, QStackedWidget, QListWidget, QListWidgetItem, QListView,
) )
from PyQt6.QtCore import Qt, QTimer, pyqtSignal, QSize, QUrl, QMimeData from PyQt6.QtCore import Qt, QTimer, pyqtSignal, QSize
from PyQt6.QtGui import QPixmap, QColor, QIcon, QPainter, QPainterPath, QDrag from PyQt6.QtGui import QPixmap, QColor, QIcon, QPainter, QPainterPath
from track_info import TrackInfo from track_info import TrackInfo
from metadata_handler import MetadataHandler from metadata_handler import MetadataHandler
@ -42,51 +41,6 @@ EQ_FRAMES = [
"\u2585\u2586\u2583", "\u2586\u2584\u2581", "\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): class NumericTableItem(QTableWidgetItem):
def __lt__(self, other): def __lt__(self, other):
_, self_data = self.data(Qt.ItemDataRole.UserRole) or (False, {}) _, self_data = self.data(Qt.ItemDataRole.UserRole) or (False, {})
@ -100,6 +54,7 @@ class LibraryTab(QWidget):
"""Library tab widget with library table, filtering, and transfer workflows.""" """Library tab widget with library table, filtering, and transfer workflows."""
transfer_finished = pyqtSignal() transfer_finished = pyqtSignal()
tab_switch_requested = pyqtSignal(int)
play_requested = pyqtSignal(dict, int, bool) play_requested = pyqtSignal(dict, int, bool)
def __init__(self, config_loader: ConfigLoader, parent=None): def __init__(self, config_loader: ConfigLoader, parent=None):
@ -140,7 +95,7 @@ class LibraryTab(QWidget):
self._setup_library_toolbar(layout) self._setup_library_toolbar(layout)
self.library_table = DraggableLibraryTable() self.library_table = QTableWidget()
self.library_table.setColumnCount(8) self.library_table.setColumnCount(8)
self.library_table.setHorizontalHeaderLabels( self.library_table.setHorizontalHeaderLabels(
["#", "Artist", "Title", "Album", "Duration", "Genre", "Size", "Plays"] ["#", "Artist", "Title", "Album", "Duration", "Genre", "Size", "Plays"]
@ -257,13 +212,6 @@ class LibraryTab(QWidget):
layout.addLayout(toolbar) 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): 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(): if index < 0 or index >= self.library_table.rowCount():
return return
@ -450,8 +398,7 @@ class LibraryTab(QWidget):
return f"{size_bytes / 1024:.0f} KB" return f"{size_bytes / 1024:.0f} KB"
return f"{size_bytes / (1024 * 1024):.1f} MB" return f"{size_bytes / (1024 * 1024):.1f} MB"
def _scan_library(self, force: bool = False, def _scan_library(self, force: bool = False):
on_progress: Callable[[int, int], None] | None = None):
output_dir = self.get_output_dir() output_dir = self.get_output_dir()
if not output_dir or not os.path.isdir(output_dir): if not output_dir or not os.path.isdir(output_dir):
self.library_ready.clear() self.library_ready.clear()
@ -459,23 +406,10 @@ class LibraryTab(QWidget):
self._refresh_library_ui() self._refresh_library_ui()
return return
# Pre-count total scannable files so the overlay can show progress.
total = 0
if on_progress:
for root, _dirs, files in os.walk(output_dir):
for fname in files:
fpath = os.path.join(root, fname)
ext = os.path.splitext(fname)[1].lower()
if os.path.isfile(fpath) and (ext in AUDIO_EXTS or ext in SOURCE_EXTS):
total += 1
if total == 0:
total = 1
ready = [] ready = []
sources = [] sources = []
cache_hits = 0 cache_hits = 0
cache_misses = 0 cache_misses = 0
processed = 0
cache = get_library_cache() cache = get_library_cache()
handler = MetadataHandler() handler = MetadataHandler()
@ -487,7 +421,6 @@ class LibraryTab(QWidget):
if not os.path.isfile(fpath): if not os.path.isfile(fpath):
continue continue
processed += 1
cached = None if force else cache.get(fpath) cached = None if force else cache.get(fpath)
if cached is not None: if cached is not None:
@ -498,8 +431,10 @@ class LibraryTab(QWidget):
elif ext in SOURCE_EXTS: elif ext in SOURCE_EXTS:
cached.pop("_mtime", None) cached.pop("_mtime", None)
sources.append(cached) sources.append(cached)
else: continue
cache_misses += 1 cache_misses += 1
if ext in AUDIO_EXTS: if ext in AUDIO_EXTS:
track_data = self._extract_ready_track(fpath, fname, handler) track_data = self._extract_ready_track(fpath, fname, handler)
ready.append(track_data) ready.append(track_data)
@ -509,9 +444,6 @@ class LibraryTab(QWidget):
sources.append(track_data) sources.append(track_data)
cache.put(fpath, track_data) cache.put(fpath, track_data)
if on_progress:
on_progress(processed, total)
cache.save() cache.save()
if cache_misses: if cache_misses:
@ -685,52 +617,6 @@ class LibraryTab(QWidget):
self._restore_playing_row() self._restore_playing_row()
self._build_album_grid() self._build_album_grid()
def refresh_play_counts_from_cache(self):
"""Reload play-count values from the persistent cache without a
full library rescan.
Called after ``mount_and_load`` completes — the cache already
holds merged iPod + local play stats, but the table was
rendered before the mount and displays stale values.
"""
cache = get_library_cache()
updated = 0
ready_index = {t["path"]: i for i, t in enumerate(self.library_ready)}
for row in range(self.library_table.rowCount()):
item_0 = self.library_table.item(row, 0)
if item_0 is None:
continue
data = item_0.data(Qt.ItemDataRole.UserRole)
if data is None:
continue
_is_ready, track = data
if not _is_ready:
continue
path = track.get("path", "")
if not path:
continue
entry = cache.get(path)
if entry is None:
continue
new_pc = entry.get("play_count", 0)
old_pc = track.get("play_count", 0)
if new_pc != old_pc:
track["play_count"] = new_pc
pc_cell = self.library_table.item(row, 7)
if pc_cell:
pc_cell.setText(str(new_pc))
idx = ready_index.get(path)
if idx is not None:
self.library_ready[idx]["play_count"] = new_pc
updated += 1
if updated:
logger.debug("Refreshed play counts for %d tracks from cache", updated)
def _restore_playing_row(self): def _restore_playing_row(self):
idx = self.current_playback_index idx = self.current_playback_index
if idx < 0: if idx < 0:
@ -816,77 +702,20 @@ class LibraryTab(QWidget):
self._scan_library(force=True) self._scan_library(force=True)
def _on_library_add_files(self): def _on_library_add_files(self):
all_exts = sorted(SOURCE_EXTS | AUDIO_EXTS) source_filter = "Audio Files (*.flac *.wav *.ogg *.opus *.wma *.ape *.aiff *.alac)"
source_filter = f"Audio Files (*{' *'.join(all_exts)})"
files, _ = QFileDialog.getOpenFileNames( files, _ = QFileDialog.getOpenFileNames(
self, "Select Audio Files", "", source_filter self, "Select Audio Files", "", source_filter
) )
new_source_files = [] new_files = []
copied_ready = []
for f in files: for f in files:
ext = os.path.splitext(f)[1].lower() 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: if ext in SOURCE_EXTS and os.path.exists(f) and f not in self.library_source_files:
self.library_source_files.append(f) self.library_source_files.append(f)
new_source_files.append(f) new_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: if new_files:
self._scan_library() 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): def _incremental_update_after_edit(self, file_paths: list):
cache = get_library_cache() cache = get_library_cache()
handler = MetadataHandler() handler = MetadataHandler()
@ -1128,7 +957,7 @@ class LibraryTab(QWidget):
) )
return return
if self._is_worker_running(): if self.worker_thread and self.worker_thread.isRunning():
QMessageBox.information(self, "Info", "Please wait for the current task to finish") QMessageBox.information(self, "Info", "Please wait for the current task to finish")
return return
@ -1196,7 +1025,7 @@ class LibraryTab(QWidget):
QMessageBox.warning(self, "Error", "No iPod device mounted. Go to iPod tab and mount first.") QMessageBox.warning(self, "Error", "No iPod device mounted. Go to iPod tab and mount first.")
return return
if self._is_worker_running(): if self.worker_thread and self.worker_thread.isRunning():
QMessageBox.information(self, "Info", "Please wait for the current task to finish") QMessageBox.information(self, "Info", "Please wait for the current task to finish")
return return
@ -1231,72 +1060,6 @@ class LibraryTab(QWidget):
self._cleanup_worker() 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): def _on_sync_metadata_clicked(self):
selected_tracks = self._get_selected_ready_tracks() selected_tracks = self._get_selected_ready_tracks()
if not selected_tracks: if not selected_tracks:
@ -1307,7 +1070,7 @@ class LibraryTab(QWidget):
QMessageBox.warning(self, "Error", "No iPod device mounted.") QMessageBox.warning(self, "Error", "No iPod device mounted.")
return return
if self._is_worker_running(): if self.worker_thread and self.worker_thread.isRunning():
QMessageBox.information(self, "Info", "Please wait for the current task to finish") QMessageBox.information(self, "Info", "Please wait for the current task to finish")
return return
@ -1377,18 +1140,10 @@ class LibraryTab(QWidget):
def _cleanup_worker(self): def _cleanup_worker(self):
if self.worker_thread: if self.worker_thread:
self.worker_thread.quit() self.worker_thread.wait(3000)
self.worker_thread.wait() self.worker_thread.deleteLater()
self.worker_thread = None 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): def on_device_mounted(self, mount_point: str):
self._current_mount_point = mount_point self._current_mount_point = mount_point
self.library_progress.setVisible(True) self.library_progress.setVisible(True)
@ -1446,7 +1201,7 @@ class LibraryTab(QWidget):
entry = cache.get(path) if path else None entry = cache.get(path) if path else None
visible = False visible = False
if entry and isinstance(entry, dict): if entry and isinstance(entry, dict):
added = entry.get("mtime", 0) or entry.get("added_at", 0) or 0 added = entry.get("_mtime", 0) or entry.get("added_at", 0) or 0
visible = added >= cutoff visible = added >= cutoff
self.library_table.setRowHidden(row, not visible) self.library_table.setRowHidden(row, not visible)

89
src/ui/metadata_editor.py
View File

@ -31,8 +31,6 @@ from PyQt6.QtGui import QPixmap, QDragEnterEvent, QDropEvent
from metadata_handler import MetadataHandler from metadata_handler import MetadataHandler
from artwork.cache import cover_cache from artwork.cache import cover_cache
from services.itunes_search import search_itunes from services.itunes_search import search_itunes
from library_cache import get_library_cache, save_library_cache
from .star_rating import ClickableStarRating
COVER_SIZE = 250 COVER_SIZE = 250
@ -74,8 +72,6 @@ class MetadataEditorDialog(QDialog):
self._original_tags = dict(first_tags) self._original_tags = dict(first_tags)
self._existing_cover_data = self._extract_cover_from_path(self.file_paths[0]) self._existing_cover_data = self._extract_cover_from_path(self.file_paths[0])
self._load_ratings_from_cache()
if self._batch_mode: if self._batch_mode:
all_tags = [first_tags] all_tags = [first_tags]
all_different = False all_different = False
@ -88,31 +84,6 @@ class MetadataEditorDialog(QDialog):
if tags.get(key) != self._original_tags.get(key): if tags.get(key) != self._original_tags.get(key):
self._mixed_fields.add(key) self._mixed_fields.add(key)
def _load_ratings_from_cache(self):
"""Load rating from LibraryCache for all selected files.
Stores the rating (0–5 stars) in ``self._original_tags['rating']``
(only when cache has a value; file‑tag rating takes precedence if
cache is empty). In batch mode, marks ``rating`` as mixed when
values differ.
"""
cache = get_library_cache()
ratings = []
for p in self.file_paths:
entry = cache.get(p)
r = entry.get("rating", 0) if entry else 0
ratings.append(max(0, min(5, round(r / 20))))
cache_val = ratings[0]
if cache_val > 0 or "rating" not in self._original_tags:
self._original_tags["rating"] = cache_val
if self._batch_mode and len(set(ratings)) > 1:
self._mixed_fields.add("rating")
self._ratings = {p: r for p, r in zip(self.file_paths, ratings)}
else:
self._ratings = {}
def _load_single_file(self, file_path: str, skip_cover: bool = False) -> Dict[str, Any]: def _load_single_file(self, file_path: str, skip_cover: bool = False) -> Dict[str, Any]:
"""Load tags from a single file. Returns dict of tag_key -> value.""" """Load tags from a single file. Returns dict of tag_key -> value."""
result: Dict[str, Any] = {} result: Dict[str, Any] = {}
@ -139,40 +110,12 @@ class MetadataEditorDialog(QDialog):
result[our_key] = val result[our_key] = val
result.update(self._load_track_disc(tags)) result.update(self._load_track_disc(tags))
result.update(self._load_rating_from_tags(tags, ext))
except Exception: except Exception:
pass pass
return result return result
@staticmethod
def _load_rating_from_tags(tags, ext: str) -> dict:
"""Read rating (0–5 stars) from audio file tags, if present."""
if tags is None:
return {}
try:
if ext in ('.m4a', '.aac', '.mp4') and '\xa9rt' in tags:
rt = tags['\xa9rt']
if isinstance(rt, list) and rt:
val = int(rt[0]) // 20 # 0-100 → 0-5
return {"rating": max(0, min(5, val))}
elif ext == '.mp3':
for frame in tags.getall('POPM'):
if frame.rating > 0:
pops = [0, 64, 128, 192, 224, 255]
stars = max(i for i, v in enumerate(pops) if v <= frame.rating)
return {"rating": stars}
elif ext in ('.flac', '.ogg', '.opus') and 'RATING' in tags:
raw = tags['RATING']
if isinstance(raw, list):
raw = raw[0]
val = int(str(raw)) // 20
return {"rating": max(0, min(5, val))}
except Exception:
pass
return {}
def _load_track_disc(self, tags) -> Dict[str, Any]: def _load_track_disc(self, tags) -> Dict[str, Any]:
result = {} result = {}
if tags is None: if tags is None:
@ -423,11 +366,6 @@ class MetadataEditorDialog(QDialog):
self._genre_label = QLabel("Genre:") self._genre_label = QLabel("Genre:")
form_layout.addRow(self._genre_label, self.genre_edit) form_layout.addRow(self._genre_label, self.genre_edit)
self.rating_widget = ClickableStarRating()
self.rating_widget.rating_changed.connect(self._on_rating_changed)
self._rating_label = QLabel("Rating:")
form_layout.addRow(self._rating_label, self.rating_widget)
self._year_label = QLabel("Year:") self._year_label = QLabel("Year:")
form_layout.addRow(self._year_label, self.year_spin) form_layout.addRow(self._year_label, self.year_spin)
@ -491,10 +429,6 @@ class MetadataEditorDialog(QDialog):
for key in self._mixed_fields: for key in self._mixed_fields:
self._set_mixed(key) self._set_mixed(key)
if "rating" in self._mixed_fields:
self.rating_widget.set_rating(0)
self._rating_label.setText("Rating: (mixed)")
def _set_mixed(self, key: str): def _set_mixed(self, key: str):
placeholders = { placeholders = {
'title': "(different values)", 'title': "(different values)",
@ -554,12 +488,8 @@ class MetadataEditorDialog(QDialog):
self.disc_num_spin.setValue(t.get('disc_number', 0)) self.disc_num_spin.setValue(t.get('disc_number', 0))
self.disc_total_spin.setValue(t.get('disc_total', 0)) self.disc_total_spin.setValue(t.get('disc_total', 0))
self.rating_widget.set_rating(t.get("rating", 0))
self._update_cover_display() self._update_cover_display()
def _on_rating_changed(self, stars: int) -> None:
pass # placeholder for future live-preview
def _update_cover_display(self): def _update_cover_display(self):
if self._cover_removed: if self._cover_removed:
self.cover_label.setText("No cover") self.cover_label.setText("No cover")
@ -699,7 +629,6 @@ class MetadataEditorDialog(QDialog):
'disc_number': self.disc_num_spin.value(), 'disc_number': self.disc_num_spin.value(),
'disc_total': self.disc_total_spin.value(), 'disc_total': self.disc_total_spin.value(),
'comment': self.comment_edit.text().strip(), 'comment': self.comment_edit.text().strip(),
'rating': self.rating_widget.rating(),
} }
def _collect_changes(self) -> Dict[str, Any]: def _collect_changes(self) -> Dict[str, Any]:
@ -780,9 +709,6 @@ class MetadataEditorDialog(QDialog):
else: else:
self._invalidate_cover_cache(path, changes, cover_arg) self._invalidate_cover_cache(path, changes, cover_arg)
if not failed:
self._persist_rating(paths, changes)
if failed == len(paths): if failed == len(paths):
QMessageBox.warning(self, "Error", "Failed to save metadata changes.") QMessageBox.warning(self, "Error", "Failed to save metadata changes.")
elif failed: elif failed:
@ -796,21 +722,6 @@ class MetadataEditorDialog(QDialog):
except Exception as e: except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to save metadata:\n{e}") QMessageBox.critical(self, "Error", f"Failed to save metadata:\n{e}")
def _persist_rating(self, paths: list[str], changes: dict[str, Any]) -> None:
if "rating" not in changes:
return
try:
cache = get_library_cache()
for p in paths:
entry = cache.get(p)
if entry is None:
continue
entry["rating"] = changes["rating"] * 20
cache.put(p, entry)
save_library_cache()
except Exception:
logger.exception("Failed to persist rating to LibraryCache")
def _invalidate_cover_cache(self, file_path: str, changes: Dict[str, Any], def _invalidate_cover_cache(self, file_path: str, changes: Dict[str, Any],
cover_arg: Optional[str]) -> None: cover_arg: Optional[str]) -> None:
"""Update cover cache after successful save.""" """Update cover cache after successful save."""

49
src/ui/player_header.py
View File

@ -1,5 +1,5 @@
from PyQt6.QtWidgets import ( from PyQt6.QtWidgets import (
QWidget, QHBoxLayout, QVBoxLayout, QPushButton, QLabel, QSlider, QGroupBox, QStyle, QProxyStyle, QWidget, QHBoxLayout, QVBoxLayout, QPushButton, QLabel, QSlider, QGroupBox, QStyle,
) )
from PyQt6.QtCore import Qt, pyqtSignal from PyQt6.QtCore import Qt, pyqtSignal
from PyQt6.QtGui import QPixmap, QColor from PyQt6.QtGui import QPixmap, QColor
@ -11,19 +11,10 @@ def _fmt_time(ms: int) -> str:
return f"{m}:{s:02d}" return f"{m}:{s:02d}"
class JumpStyle(QProxyStyle):
def styleHint(self, hint, opt=None, widget=None, returnData=None):
if hint == QStyle.StyleHint.SH_Slider_AbsoluteSetButtons:
return Qt.MouseButton.LeftButton.value
return super().styleHint(hint, opt, widget, returnData)
class PlayerHeader(QGroupBox): class PlayerHeader(QGroupBox):
prev_requested = pyqtSignal() prev_requested = pyqtSignal()
play_pause_requested = pyqtSignal() play_pause_requested = pyqtSignal()
next_requested = pyqtSignal() next_requested = pyqtSignal()
shuffle_requested = pyqtSignal()
repeat_requested = pyqtSignal()
position_changed_by_user = pyqtSignal(int) position_changed_by_user = pyqtSignal(int)
volume_changed = pyqtSignal(int) volume_changed = pyqtSignal(int)
@ -54,28 +45,15 @@ class PlayerHeader(QGroupBox):
self.next_btn.setToolTip("Next") self.next_btn.setToolTip("Next")
self.next_btn.clicked.connect(self.next_requested) self.next_btn.clicked.connect(self.next_requested)
self.shuffle_btn = QPushButton("\u21C4")
self.shuffle_btn.setFixedWidth(36)
self.shuffle_btn.setToolTip("Shuffle")
self.shuffle_btn.clicked.connect(self.shuffle_requested)
self.repeat_btn = QPushButton("\u21BB")
self.repeat_btn.setFixedWidth(36)
self.repeat_btn.setToolTip("Repeat: Off")
self.repeat_btn.clicked.connect(self.repeat_requested)
controls_row.addWidget(self.shuffle_btn)
controls_row.addWidget(self.prev_btn) controls_row.addWidget(self.prev_btn)
controls_row.addWidget(self.play_btn) controls_row.addWidget(self.play_btn)
controls_row.addWidget(self.next_btn) controls_row.addWidget(self.next_btn)
controls_row.addWidget(self.repeat_btn)
self.time_label_start = QLabel("0:00") self.time_label_start = QLabel("0:00")
self.time_label_start.setFixedWidth(40) self.time_label_start.setFixedWidth(40)
self.time_label_start.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter) self.time_label_start.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter)
self.position_slider = QSlider(Qt.Orientation.Horizontal) self.position_slider = QSlider(Qt.Orientation.Horizontal)
self.position_slider.setStyle(JumpStyle(self.position_slider.style()))
self.position_slider.setMinimum(0) self.position_slider.setMinimum(0)
self.position_slider.setMaximum(1000) self.position_slider.setMaximum(1000)
self.position_slider.setValue(0) self.position_slider.setValue(0)
@ -116,30 +94,9 @@ class PlayerHeader(QGroupBox):
now_playing_layout.addWidget(self.now_playing_label, stretch=1) now_playing_layout.addWidget(self.now_playing_label, stretch=1)
layout.addLayout(now_playing_layout) layout.addLayout(now_playing_layout)
def set_shuffle_active(self, active: bool):
if active:
self.shuffle_btn.setStyleSheet("color: palette(highlight); font-weight: bold;")
self.shuffle_btn.setToolTip("Shuffle: On")
else:
self.shuffle_btn.setStyleSheet("")
self.shuffle_btn.setToolTip("Shuffle: Off")
def set_repeat_mode(self, mode: str):
if mode == "repeat_all":
self.repeat_btn.setText("\u21BB")
self.repeat_btn.setStyleSheet("color: palette(highlight); font-weight: bold;")
self.repeat_btn.setToolTip("Repeat: All")
elif mode == "repeat_one":
self.repeat_btn.setText("\u21BB\u00B9")
self.repeat_btn.setStyleSheet("color: palette(highlight); font-weight: bold;")
self.repeat_btn.setToolTip("Repeat: One")
else:
self.repeat_btn.setText("\u21BB")
self.repeat_btn.setStyleSheet("")
self.repeat_btn.setToolTip("Repeat: Off")
def _on_slider_user_action(self): def _on_slider_user_action(self):
self.position_changed_by_user.emit(self.position_slider.value()) frac = self.position_slider.value() / 1000.0
self.position_changed_by_user.emit(int(frac))
def set_cover(self, pixmap: QPixmap = None, tooltip: str = ""): def set_cover(self, pixmap: QPixmap = None, tooltip: str = ""):
if pixmap: if pixmap:

199
src/ui/sidebar_widget.py
View File

@ -1,5 +1,4 @@
import logging import logging
import os
from typing import List, Optional from typing import List, Optional
from PyQt6.QtWidgets import ( from PyQt6.QtWidgets import (
@ -7,9 +6,8 @@ from PyQt6.QtWidgets import (
QListWidget, QListWidgetItem, QMenu, QInputDialog, QMessageBox, QListWidget, QListWidgetItem, QMenu, QInputDialog, QMessageBox,
QFrame, QFrame,
) )
from PyQt6.QtCore import Qt, pyqtSignal, QUrl from PyQt6.QtCore import Qt, pyqtSignal
from PyQt6.QtGui import QColor from PyQt6.QtGui import QColor
from PyQt6.QtGui import QDragEnterEvent, QDragMoveEvent, QDragLeaveEvent, QDropEvent
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -18,184 +16,9 @@ SIDEBAR_MAX_WIDTH = 240
SIDEBAR_DEFAULT_WIDTH = 200 SIDEBAR_DEFAULT_WIDTH = 200
class PlaylistDropListWidget(QListWidget):
"""QListWidget that accepts external track drops onto playlist items."""
tracks_dropped_on_playlist = pyqtSignal(str, list)
def __init__(self, parent=None):
super().__init__(parent)
self.setAcceptDrops(True)
self._hovered_item = None
def mimeData(self, items):
mime = super().mimeData(items)
if items:
playlist_id = items[0].data(Qt.ItemDataRole.UserRole)
if playlist_id:
mime.setData("application/x-neopod-playlist", playlist_id.encode())
return mime
def dragEnterEvent(self, event: QDragEnterEvent):
if event.source() is not self and event.mimeData().hasUrls():
event.acceptProposedAction()
else:
super().dragEnterEvent(event)
def dragMoveEvent(self, event: QDragMoveEvent):
if event.source() is not self and event.mimeData().hasUrls():
pos = event.position().toPoint()
item = self.itemAt(pos)
if item != self._hovered_item:
self._clear_hover()
if item:
self._hovered_item = item
bg = self.palette().color(self.palette().ColorRole.Highlight)
c = bg.name()
item.setBackground(bg)
event.acceptProposedAction()
else:
super().dragMoveEvent(event)
def dragLeaveEvent(self, event: QDragLeaveEvent):
self._clear_hover()
super().dragLeaveEvent(event)
def dropEvent(self, event: QDropEvent):
self._clear_hover()
if event.source() is not self and event.mimeData().hasUrls():
pos = event.position().toPoint()
item = self.itemAt(pos)
if item is None:
event.ignore()
return
playlist_id = item.data(Qt.ItemDataRole.UserRole)
if not playlist_id:
event.ignore()
return
paths = [url.toLocalFile() for url in event.mimeData().urls()]
paths = [p for p in paths if os.path.exists(p)]
if paths:
self.tracks_dropped_on_playlist.emit(playlist_id, paths)
event.acceptProposedAction()
else:
event.ignore()
else:
super().dropEvent(event)
def _clear_hover(self):
if self._hovered_item:
self._hovered_item.setData(Qt.ItemDataRole.BackgroundRole, None)
self._hovered_item = None
class NavDropListWidget(QListWidget):
"""QListWidget that accepts external track drops on the iPod nav item."""
tracks_dropped_for_transfer = pyqtSignal(list)
playlist_dropped_for_transfer = pyqtSignal(str)
def __init__(self, parent=None):
super().__init__(parent)
self.setAcceptDrops(True)
self._hovered_item = None
def _target_item_at(self, pos):
item = self.itemAt(pos)
if item and item.data(Qt.ItemDataRole.UserRole + 2) == "ipod":
return item
return None
def _repolish(self):
self.style().unpolish(self)
self.style().polish(self)
def dragEnterEvent(self, event: QDragEnterEvent):
if event.mimeData().hasFormat("application/x-neopod-playlist"):
self.setProperty("dndActive", True)
self._repolish()
pos = event.position().toPoint()
if self._target_item_at(pos):
event.acceptProposedAction()
return
# Let default handler accept so dragMoveEvent still fires
if event.source() is not self and event.mimeData().hasUrls():
pos = event.position().toPoint()
if self._target_item_at(pos):
event.acceptProposedAction()
return
super().dragEnterEvent(event)
def dragMoveEvent(self, event: QDragMoveEvent):
accept = False
if event.source() is not self and event.mimeData().hasUrls():
pos = event.position().toPoint()
item = self._target_item_at(pos)
if item != self._hovered_item:
self._clear_hover()
if item:
self._hovered_item = item
bg = self.palette().color(self.palette().ColorRole.Highlight)
item.setBackground(bg)
if item:
event.acceptProposedAction()
accept = True
if not accept and event.mimeData().hasFormat("application/x-neopod-playlist"):
pos = event.position().toPoint()
item = self._target_item_at(pos)
if item != self._hovered_item:
self._clear_hover()
if item:
self._hovered_item = item
bg = self.palette().color(self.palette().ColorRole.Highlight)
item.setBackground(bg)
if item:
event.acceptProposedAction()
accept = True
if not accept:
event.ignore()
def dragLeaveEvent(self, event: QDragLeaveEvent):
self._clear_hover()
if self.property("dndActive"):
self.setProperty("dndActive", False)
self._repolish()
super().dragLeaveEvent(event)
def dropEvent(self, event: QDropEvent):
self._clear_hover()
if self.property("dndActive"):
self.setProperty("dndActive", False)
self._repolish()
if event.source() is not self and event.mimeData().hasUrls():
pos = event.position().toPoint()
if self._target_item_at(pos):
paths = [url.toLocalFile() for url in event.mimeData().urls()]
paths = [p for p in paths if os.path.exists(p)]
if paths:
self.tracks_dropped_for_transfer.emit(paths)
event.acceptProposedAction()
return
if event.mimeData().hasFormat("application/x-neopod-playlist"):
pos = event.position().toPoint()
if self._target_item_at(pos):
playlist_id = bytes(event.mimeData().data("application/x-neopod-playlist")).decode()
if playlist_id:
self.playlist_dropped_for_transfer.emit(playlist_id)
event.acceptProposedAction()
return
super().dropEvent(event)
def _clear_hover(self):
if self._hovered_item:
self._hovered_item.setData(Qt.ItemDataRole.BackgroundRole, None)
self._hovered_item = None
class SidebarWidget(QWidget): class SidebarWidget(QWidget):
item_selected = pyqtSignal(str, str) item_selected = pyqtSignal(str, str)
playlist_selected = pyqtSignal(str) playlist_selected = pyqtSignal(str)
tracks_dropped_on_playlist = pyqtSignal(str, list)
tracks_dropped_for_nav_transfer = pyqtSignal(list)
playlist_dropped_for_nav_transfer = pyqtSignal(str)
def __init__(self, parent=None): def __init__(self, parent=None):
super().__init__(parent) super().__init__(parent)
@ -255,9 +78,6 @@ class SidebarWidget(QWidget):
QListWidget::item:hover:!selected {{ QListWidget::item:hover:!selected {{
background: {hl_light_name}; background: {hl_light_name};
}} }}
#navList[dndActive="true"]::item:hover:!selected {{
background: transparent;
}}
""") """)
def _build_ui(self): def _build_ui(self):
@ -306,7 +126,7 @@ class SidebarWidget(QWidget):
self._playlists_header.setStyleSheet(section_style) self._playlists_header.setStyleSheet(section_style)
layout.addWidget(self._playlists_header) layout.addWidget(self._playlists_header)
self._playlist_list = PlaylistDropListWidget() self._playlist_list = QListWidget()
self._playlist_list.setDragDropMode(QListWidget.DragDropMode.InternalMove) self._playlist_list.setDragDropMode(QListWidget.DragDropMode.InternalMove)
self._playlist_list.setDefaultDropAction(Qt.DropAction.MoveAction) self._playlist_list.setDefaultDropAction(Qt.DropAction.MoveAction)
self._playlist_list.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu) self._playlist_list.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
@ -315,7 +135,6 @@ class SidebarWidget(QWidget):
) )
self._playlist_list.itemClicked.connect(self._on_playlist_clicked) self._playlist_list.itemClicked.connect(self._on_playlist_clicked)
self._playlist_list.model().rowsMoved.connect(self._on_playlists_reordered) self._playlist_list.model().rowsMoved.connect(self._on_playlists_reordered)
self._playlist_list.tracks_dropped_on_playlist.connect(self._on_tracks_dropped_on_playlist)
layout.addWidget(self._playlist_list, stretch=1) layout.addWidget(self._playlist_list, stretch=1)
btn_style = f""" btn_style = f"""
@ -344,8 +163,7 @@ class SidebarWidget(QWidget):
line2.setFixedHeight(1) line2.setFixedHeight(1)
layout.addWidget(line2) layout.addWidget(line2)
self._nav_list = NavDropListWidget() self._nav_list = QListWidget()
self._nav_list.setObjectName("navList")
items = [ items = [
("🎸 iPod", "ipod"), ("🎸 iPod", "ipod"),
("⚙️ Settings", "settings"), ("⚙️ Settings", "settings"),
@ -356,8 +174,6 @@ class SidebarWidget(QWidget):
item.setData(Qt.ItemDataRole.UserRole + 2, tab_id) item.setData(Qt.ItemDataRole.UserRole + 2, tab_id)
self._nav_list.addItem(item) self._nav_list.addItem(item)
self._nav_list.itemClicked.connect(self._on_nav_item_clicked) self._nav_list.itemClicked.connect(self._on_nav_item_clicked)
self._nav_list.tracks_dropped_for_transfer.connect(self._on_tracks_dropped_for_nav_transfer)
self._nav_list.playlist_dropped_for_transfer.connect(self._on_playlist_dropped_for_transfer)
layout.addWidget(self._nav_list) layout.addWidget(self._nav_list)
self._playlist_items: dict = {} self._playlist_items: dict = {}
@ -445,15 +261,6 @@ class SidebarWidget(QWidget):
ids.append(item.data(Qt.ItemDataRole.UserRole)) ids.append(item.data(Qt.ItemDataRole.UserRole))
return ids return ids
def _on_tracks_dropped_on_playlist(self, playlist_id: str, paths: list):
self.tracks_dropped_on_playlist.emit(playlist_id, paths)
def _on_tracks_dropped_for_nav_transfer(self, paths: list):
self.tracks_dropped_for_nav_transfer.emit(paths)
def _on_playlist_dropped_for_transfer(self, playlist_id: str):
self.playlist_dropped_for_nav_transfer.emit(playlist_id)
def _on_playlists_reordered(self): def _on_playlists_reordered(self):
order = self.get_playlist_order_ids() order = self.get_playlist_order_ids()
self.item_selected.emit("playlist_reorder", "|".join(order)) self.item_selected.emit("playlist_reorder", "|".join(order))

118
src/ui/splash_overlay.py
View File

@ -1,118 +0,0 @@
#!/usr/bin/env python3
"""Splash overlay shown while scanning the music library for the first time."""
from PyQt6.QtWidgets import QDialog, QVBoxLayout, QLabel, QWidget, QApplication
from PyQt6.QtCore import Qt, QTimer
from PyQt6.QtGui import QPainter, QPen, QColor, QFont
class _Spinner(QWidget):
"""A spinning arc — like a circular progress indicator."""
def __init__(self, parent=None):
super().__init__(parent)
self.setFixedSize(64, 64)
self._angle = 0.0
self._timer = QTimer(self)
self._timer.timeout.connect(self._tick)
self._timer.start(30)
def _tick(self):
self._angle = (self._angle + 6) % 360
self.update()
def paintEvent(self, event):
p = QPainter(self)
p.setRenderHint(QPainter.RenderHint.Antialiasing)
r = self.rect().adjusted(4, 4, -4, -4)
p.setPen(QPen(QColor("#e0e0e0"), 3, Qt.PenStyle.SolidLine, Qt.PenCapStyle.RoundCap))
p.drawEllipse(r)
p.setPen(QPen(QColor("#3b82f6"), 4, Qt.PenStyle.SolidLine, Qt.PenCapStyle.RoundCap))
p.drawArc(r, int(self._angle * 16), 60 * 16)
class SplashOverlay(QDialog):
"""Modal overlay with a spinning animation + progress counter.
Usage::
splash = SplashOverlay(window)
splash.show()
# … scan …
splash.set_progress(current, total)
splash.close()
"""
def __init__(self, parent=None):
super().__init__(parent)
self.setWindowFlags(
Qt.WindowType.FramelessWindowHint | Qt.WindowType.Dialog
)
self.setAttribute(Qt.WidgetAttribute.WA_TranslucentBackground)
self.setAttribute(Qt.WidgetAttribute.WA_ShowWithoutActivating)
self.setModal(True)
self._card = QWidget(self)
self._card.setObjectName("card")
self._card.setFixedSize(340, 240)
layout = QVBoxLayout(self._card)
layout.setAlignment(Qt.AlignmentFlag.AlignCenter)
layout.setSpacing(16)
self._spinner = _Spinner()
layout.addWidget(self._spinner, alignment=Qt.AlignmentFlag.AlignCenter)
self._title = QLabel("Создаём музыкальную\nмедиатеку…")
self._title.setAlignment(Qt.AlignmentFlag.AlignCenter)
f = self._title.font()
f.setPointSize(16)
f.setBold(True)
self._title.setFont(f)
layout.addWidget(self._title)
self._count = QLabel("")
self._count.setAlignment(Qt.AlignmentFlag.AlignCenter)
f2 = self._count.font()
f2.setPointSize(12)
self._count.setFont(f2)
layout.addWidget(self._count)
self._card.setStyleSheet("""
#card {
background: white;
border-radius: 16px;
}
QLabel {
color: #333;
background: transparent;
}
""")
def _recenter(self):
if not self.parent():
return
pr = self.parent().rect()
self.setGeometry(pr)
cx = (pr.width() - self._card.width()) // 2
cy = (pr.height() - self._card.height()) // 2
self._card.move(cx, cy)
def showEvent(self, event):
self._recenter()
super().showEvent(event)
QApplication.processEvents()
def resizeEvent(self, event):
self._recenter()
super().resizeEvent(event)
def set_progress(self, current: int, total: int) -> None:
"""Update the file counter shown on the overlay."""
if total > 0:
self._count.setText(f"Файл {current} из {total}")
else:
self._count.setText(f"Файл {current}")
QApplication.processEvents()

107
src/ui/star_rating.py
View File

@ -1,107 +0,0 @@
#!/usr/bin/env python3
"""
Clickable 5‑star rating widget.
Scale: 0 (unrated) … 5 (★★★★★). Stored internally as stars;
callers convert to/from the iPod 0–100 scale as needed.
"""
import math
from PyQt6.QtCore import Qt, QRectF, pyqtSignal
from PyQt6.QtGui import QPainter, QColor, QPainterPath, QBrush, QPen
from PyQt6.QtWidgets import QWidget
class ClickableStarRating(QWidget):
"""Five clickable stars with hover preview and toggle-to-unrate."""
rating_changed = pyqtSignal(int)
STAR_COUNT = 5
STAR_SIZE = 22 # diameter of each star
PADDING = 4
FILLED_COLOR = QColor("#ffb347") # warm amber
EMPTY_COLOR = QColor("#555555")
HOVER_COLOR = QColor("#ffd700") # gold
def __init__(self, parent=None):
super().__init__(parent)
self._rating = 0
self._hover_idx = -1
w = self.STAR_COUNT * self.STAR_SIZE + (self.STAR_COUNT - 1) * self.PADDING
self.setFixedSize(w, self.STAR_SIZE + 8)
self.setMouseTracking(True)
# ── public API ────────────────────────────────────────────────
def set_rating(self, stars: int) -> None:
self._rating = max(0, min(self.STAR_COUNT, stars))
self.update()
def rating(self) -> int:
return self._rating
# ── painting ───────────────────────────────────────────────────
def paintEvent(self, event): # noqa: N802
painter = QPainter(self)
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
painter.setPen(Qt.PenStyle.NoPen)
hovering = self._hover_idx >= 0
for i in range(self.STAR_COUNT):
if hovering:
filled = i <= self._hover_idx
else:
filled = i < self._rating
color = (
self.HOVER_COLOR if hovering and filled
else self.FILLED_COLOR if filled
else self.EMPTY_COLOR
)
painter.setBrush(QBrush(color))
cx = self.STAR_SIZE // 2 + i * (self.STAR_SIZE + self.PADDING)
cy = self.height() // 2
self._draw_star(painter, cx, cy, self.STAR_SIZE // 2 - 1)
@staticmethod
def _draw_star(painter: QPainter, cx: int, cy: int, r: int) -> None:
"""Draw a five-pointed star centred at (cx, cy) with outer radius *r*."""
path = QPainterPath()
inner_r = r * 0.4
points = []
for i in range(10):
angle = math.pi / 2 + i * math.pi / 5
radius = r if i % 2 == 0 else inner_r
x = cx + radius * math.cos(angle)
y = cy - radius * math.sin(angle)
points.append((x, y))
path.moveTo(points[0][0], points[0][1])
for x, y in points[1:]:
path.lineTo(x, y)
path.closeSubpath()
painter.drawPath(path)
# ── mouse interaction ──────────────────────────────────────────
def _star_at(self, x: int) -> int:
idx = x // (self.STAR_SIZE + self.PADDING)
return min(max(idx, 0), self.STAR_COUNT - 1)
def mousePressEvent(self, event): # noqa: N802
star = self._star_at(int(event.position().x()))
new_val = star + 1 if self._rating != star + 1 else 0
self._rating = new_val
self.rating_changed.emit(new_val)
self.update()
def mouseMoveEvent(self, event): # noqa: N802
self._hover_idx = self._star_at(int(event.position().x()))
self.update()
def leaveEvent(self, event): # noqa: N802
self._hover_idx = -1
self.update()

178
src/worker.py
View File

@ -24,15 +24,11 @@ SOURCE_EXTS = {'.flac', '.wav', '.ogg', '.opus', '.wma', '.ape', '.aiff', '.alac
class WorkerThread(QThread): class WorkerThread(QThread):
"""Worker thread for background tasks""" """Worker thread for background tasks"""
_processed_fwids: set[str] = set()
_auto_synced_fwids: set[str] = set()
_pending_commit_fwids: set[str] = set()
progress_signal = pyqtSignal(int, str) progress_signal = pyqtSignal(int, str)
finished_signal = pyqtSignal(bool, str, object) finished_signal = pyqtSignal(bool, str, object)
def __init__(self, task_type: str, parent=None, **kwargs): def __init__(self, task_type: str, **kwargs):
super().__init__(parent) super().__init__()
self.task_type = task_type self.task_type = task_type
self.kwargs = kwargs self.kwargs = kwargs
self.is_running = True self.is_running = True
@ -48,8 +44,6 @@ class WorkerThread(QThread):
"delete_tracks": self._run_delete_tracks, "delete_tracks": self._run_delete_tracks,
"remove_duplicates": self._run_remove_duplicates, "remove_duplicates": self._run_remove_duplicates,
"sync_metadata": self._run_sync_metadata, "sync_metadata": self._run_sync_metadata,
"create_playlist": self._run_create_playlist,
"auto_sync_playcounts": self._run_auto_sync_playcounts,
} }
handler = handlers.get(self.task_type) handler = handlers.get(self.task_type)
if handler: if handler:
@ -210,14 +204,11 @@ class WorkerThread(QThread):
self.progress_signal.emit(60, "Syncing play counts from iPod...") self.progress_signal.emit(60, "Syncing play counts from iPod...")
from library_cache import get_library_cache, save_library_cache from library_cache import get_library_cache
cache = get_library_cache() cache = get_library_cache()
ipod_base, _ipod_delta = db.read_play_stats() ipod_base, ipod_delta = db.read_play_stats()
fwid = db.firewire_id.hex()
WorkerThread._processed_fwids.add(fwid)
# ── build content_hash → iPod location map ──────────────────────
hash_to_location: dict[str, str] = {} hash_to_location: dict[str, str] = {}
for t in tracks: for t in tracks:
ch = t.get("content_hash") ch = t.get("content_hash")
@ -230,49 +221,37 @@ class WorkerThread(QThread):
loc = f":iPod_Control:Music:{rel.replace(os.sep, ':')}" loc = f":iPod_Control:Music:{rel.replace(os.sep, ':')}"
hash_to_location[ch] = loc hash_to_location[ch] = loc
# ── merge into local cache ──────────────────────────────────────
updated_cache = 0 updated_cache = 0
for file_path, entry in cache.iter_entries(): for file_path, entry in list(cache._data.items()):
if not os.path.exists(file_path): if not os.path.exists(file_path):
continue continue
ch = entry.get("content_hash", "") ch = entry.get("_content_hash", "")
if not ch or ch not in hash_to_location: if not ch or ch not in hash_to_location:
continue continue
loc = hash_to_location[ch] loc = hash_to_location[ch]
bs = ipod_base.get(loc, {}) bs = ipod_base.get(loc, {})
dd = ipod_delta.get(loc, {})
new_play_count = max( entry["play_count"] = max(
entry.get("play_count", 0), bs.get("play_count", 0), entry.get("play_count", 0), bs.get("play_count", 0),
) ) + dd.get("play_count", 0)
new_skip_count = max(
entry.get("skip_count", 0), bs.get("skip_count", 0),
)
new_rating = bs.get("rating", 0) or entry.get("rating", 0)
new_last_played = max(
entry.get("last_played", 0), bs.get("last_played", 0),
)
cache.update_play_stats( entry["skip_count"] = max(
file_path, entry.get("skip_count", 0), bs.get("skip_count", 0),
play_count=new_play_count, ) + dd.get("skip_count", 0)
rating=new_rating,
last_played=new_last_played, entry["rating"] = dd.get("rating", 0) or bs.get("rating", 0) or entry.get("rating", 0)
skip_count=new_skip_count,
entry["last_played"] = max(
entry.get("last_played", 0), bs.get("last_played", 0),
) )
updated_cache += 1 updated_cache += 1
if updated_cache: if updated_cache:
save_library_cache() cache.save()
logger.info("Updated play counts for %d local tracks from iPod", updated_cache) logger.info("Updated play counts for %d local tracks from iPod", updated_cache)
# ── deferred iTunesDB commit on stable USB ────────────────── db.consume_play_deltas()
if fwid in WorkerThread._pending_commit_fwids and os.access(mount_point, os.W_OK):
if db.sync_itunescdb():
WorkerThread._pending_commit_fwids.discard(fwid)
logger.info("Deferred iTunesDB commit for %s complete", fwid)
else:
logger.debug("Deferred commit for %s failed (read-only filesystem)", fwid)
self.progress_signal.emit(80, "Loading complete") self.progress_signal.emit(80, "Loading complete")
result = { result = {
@ -285,105 +264,6 @@ class WorkerThread(QThread):
True, f"Loaded {len(tracks)} tracks from iPod", result True, f"Loaded {len(tracks)} tracks from iPod", result
) )
def _run_auto_sync_playcounts(self):
mount_point = self.kwargs.get("mount_point")
if not mount_point:
self.finished_signal.emit(False, "No mount point for auto-sync", None)
return
ipod_ctrl = os.path.join(mount_point, "iPod_Control")
if not os.path.exists(ipod_ctrl):
self.finished_signal.emit(True, "Auto-sync skipped (device gone)", None)
return
from ipod_nano7_db import Nano7Database
db = Nano7Database(mount_point)
fwid = db.firewire_id.hex()
if fwid in WorkerThread._auto_synced_fwids:
self.finished_signal.emit(True, "Auto-sync already done", None)
return
ipod_base, ipod_delta = db.read_play_stats()
if not ipod_base and not ipod_delta:
import time
time.sleep(3)
ipod_base, ipod_delta = db.read_play_stats()
if not ipod_base and not ipod_delta:
ipod_delta = db.read_play_counts_delta()
if not ipod_delta:
self.finished_signal.emit(True, "Auto-sync: no delta", None)
return
from library_cache import get_library_cache, save_library_cache
cache = get_library_cache()
stored = cache.get_last_synced_delta(fwid)
effective: dict[str, dict[str, int]] = {}
for loc, dd in ipod_delta.items():
sd = stored.get(loc, {})
eff_pc = max(0, dd.get("play_count", 0) - sd.get("play_count", 0))
eff_sc = max(0, dd.get("skip_count", 0) - sd.get("skip_count", 0))
if eff_pc > 0 or eff_sc > 0:
effective[loc] = {"play_count": eff_pc, "skip_count": eff_sc}
dd_rating = dd.get("rating", 0)
if dd_rating > 0:
effective.setdefault(loc, {})["rating"] = dd_rating
cache.set_last_synced_delta(fwid, ipod_delta)
if effective:
tracks = db.get_all_tracks()
hash_to_loc: dict[str, str] = {}
loc_to_hash: dict[str, str] = {}
for t in tracks:
ch = t.get("content_hash", "")
fpath = t.get("file_path", "")
if ch and fpath:
rel = os.path.relpath(
fpath, os.path.join(db.mountpoint, "iPod_Control", "Music"),
)
loc = f":iPod_Control:Music:{rel.replace(os.sep, ':')}"
loc_to_hash[loc] = ch
hash_to_loc[ch] = loc
self.progress_signal.emit(40, "Merging play counts...")
updated = 0
for file_path, entry in cache.iter_entries():
ch = entry.get("content_hash", "")
if ch not in hash_to_loc:
continue
loc = hash_to_loc[ch]
eff = effective.get(loc)
if not eff:
continue
bs = ipod_base.get(loc, {})
new_pc = max(entry.get("play_count", 0), bs.get("play_count", 0)) + eff.get(
"play_count", 0,
)
new_sc = max(entry.get("skip_count", 0), bs.get("skip_count", 0)) + eff.get(
"skip_count", 0,
)
new_rt = eff.get("rating", 0) or bs.get("rating", 0) or entry.get("rating", 0)
new_lp = max(entry.get("last_played", 0), bs.get("last_played", 0))
cache.update_play_stats(
file_path,
play_count=new_pc,
rating=new_rt,
last_played=new_lp,
skip_count=new_sc,
)
updated += 1
if updated:
save_library_cache()
logger.info("Auto-sync: applied %d effective deltas", updated)
self.progress_signal.emit(100, "Auto-sync complete")
WorkerThread._pending_commit_fwids.add(fwid)
WorkerThread._auto_synced_fwids.add(fwid)
self.finished_signal.emit(True, "Auto-sync complete", mount_point)
def _run_export_ipod(self): def _run_export_ipod(self):
tracks = self.kwargs.get("tracks", []) tracks = self.kwargs.get("tracks", [])
dest_dir = self.kwargs.get("dest_dir", "") dest_dir = self.kwargs.get("dest_dir", "")
@ -481,26 +361,6 @@ class WorkerThread(QThread):
self.progress_signal.emit(100, msg) self.progress_signal.emit(100, msg)
self.finished_signal.emit(synced > 0, msg, {"synced": synced, "errors": errors}) self.finished_signal.emit(synced > 0, msg, {"synced": synced, "errors": errors})
def _run_create_playlist(self):
mount_point = self.kwargs.get("mount_point")
name = self.kwargs.get("playlist_name", "")
track_paths = self.kwargs.get("track_paths", [])
if not mount_point or not name:
self.finished_signal.emit(False, "Missing parameters for playlist creation", None)
return
from ipod_nano7_db import Nano7Database
db = Nano7Database(mount_point)
self.progress_signal.emit(0, f"Creating playlist '{name}' on iPod...")
success = db.create_playlist(name, track_paths)
if success:
self.finished_signal.emit(True, f"Playlist '{name}' created on iPod", None)
else:
self.finished_signal.emit(False, f"Failed to create playlist '{name}'", None)
def stop(self): def stop(self):
self.is_running = False self.is_running = False
self.wait() self.wait()

54
src/xdg_base.py
View File

@ -1,54 +0,0 @@
#!/usr/bin/env python3
"""
XDG Base Directory helpers.
Follows the XDG Base Directory Specification:
~/.config/ — configuration files
~/.local/share/ — persistent user data
~/.cache/ — temporary cache data
"""
import os
_DATA_DIR: str | None = None
_CONFIG_DIR: str | None = None
_CACHE_DIR: str | None = None
def xdg_config_path(subdir: str = "") -> str:
global _CONFIG_DIR
if _CONFIG_DIR is None:
base = os.environ.get(
"XDG_CONFIG_HOME",
os.path.join(os.path.expanduser("~"), ".config"),
)
_CONFIG_DIR = os.path.join(base, "neo-pod-desktop")
path = os.path.join(_CONFIG_DIR, subdir) if subdir else _CONFIG_DIR
os.makedirs(path, exist_ok=True)
return path
def xdg_data_path(subdir: str = "") -> str:
global _DATA_DIR
if _DATA_DIR is None:
base = os.environ.get(
"XDG_DATA_HOME",
os.path.join(os.path.expanduser("~"), ".local", "share"),
)
_DATA_DIR = os.path.join(base, "neo-pod-desktop")
path = os.path.join(_DATA_DIR, subdir) if subdir else _DATA_DIR
os.makedirs(path, exist_ok=True)
return path
def xdg_cache_path(subdir: str = "") -> str:
global _CACHE_DIR
if _CACHE_DIR is None:
base = os.environ.get(
"XDG_CACHE_HOME",
os.path.join(os.path.expanduser("~"), ".cache"),
)
_CACHE_DIR = os.path.join(base, "neo-pod-desktop")
path = os.path.join(_CACHE_DIR, subdir) if subdir else _CACHE_DIR
os.makedirs(path, exist_ok=True)
return path