travandClaude Opus 5.5
27abdbdb63
Opening Connections called find_device() on the GUI thread, which lists the gvfs-MTP mount; on a wedged mount that blocks forever in an uninterruptible FUSE wait and the app had to be force-quit. DeviceWatcher now probes on a daemon thread (one at a time, STUCK after 3 s, late answers still land), the menu reads its cached answer, and a stuck mount shows "Rabbit isn't responding…" with a Reset Connection button (kill gvfsd-mtp, remount via gio). Tests can no longer reach the real mount (conftest patches find_device), which also takes the suite from ~200 s to ~22 s. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1852 lines
82 KiB
Python
1852 lines
82 KiB
Python
import shutil
|
|
import threading
|
|
from pathlib import Path
|
|
|
|
from PyQt6.QtWidgets import (
|
|
QMainWindow, QSplitter, QStackedWidget, QWidget, QVBoxLayout, QLineEdit,
|
|
QPlainTextEdit, QTextEdit, QAbstractSpinBox, QComboBox, QApplication,
|
|
QLabel, QMessageBox, QFileDialog, QProgressBar, QPushButton,
|
|
)
|
|
from PyQt6.QtCore import Qt, QEvent, QObject, QTimer, pyqtSignal
|
|
from PyQt6.QtGui import QAction, QKeySequence
|
|
|
|
from lintunes import device_sync, music_folder, theme, url_import
|
|
from lintunes.andtunes import layout as andtunes_layout
|
|
from lintunes.andtunes import install as andtunes_install
|
|
from lintunes.andtunes import sync as andtunes_sync
|
|
from lintunes.art_search import AlbumArtFetcher
|
|
from lintunes.cast.controller import CastController
|
|
from lintunes.eventlog import log_control
|
|
from lintunes.config import load_config, save_config
|
|
from lintunes.export import exporter
|
|
from lintunes.fingerprint import TrackIdentifier, fpcalc_available
|
|
from lintunes.inhibit import INHIBIT_LOGOUT, INHIBIT_SUSPEND, SleepInhibitor
|
|
from lintunes.player import Player
|
|
from lintunes.importers import file_importer
|
|
from lintunes.gui.album_art_dialog import AlbumArtDialog
|
|
from lintunes.gui.art_ops import embed_artwork
|
|
from lintunes.gui.cast_dialog import ChromecastDialog
|
|
from lintunes.gui.cassette_ui import CassetteUi
|
|
from lintunes.gui import sounds
|
|
from lintunes.gui.device_sync_dialog import (
|
|
DeviceSyncSettingsDialog, resolve_sync_playlists,
|
|
)
|
|
from lintunes.gui.export_dialog import ExportKindDialog, WebMixDialog
|
|
from lintunes.gui.identify_dialog import IdentifyDialog
|
|
from lintunes.gui.sidebar import SIDEBAR_INSET, SidebarPanel
|
|
from lintunes.gui.library_view import LibraryView
|
|
from lintunes.gui.playlist_view import PlaylistView
|
|
from lintunes.gui.transport import TransportBar
|
|
from lintunes.gui.url_import_dialog import UrlImportDialog
|
|
from lintunes.gui.info_dialog import InfoDialog
|
|
from lintunes.gui.music_folder_dialog import MusicFolderDialog
|
|
from lintunes.gui.preferences_dialog import PreferencesDialog
|
|
from lintunes.gui.track_table import format_total_time
|
|
from lintunes.gui.version_button import VersionButton
|
|
from lintunes.updater import Updater
|
|
|
|
|
|
def album_tracks(library, artist: str, album: str) -> list:
|
|
"""Every library track on the given album (case-insensitive match on
|
|
album name + album artist, falling back to track artist — the same
|
|
grouping the art download uses)."""
|
|
artist_cf, album_cf = artist.casefold(), album.casefold()
|
|
return [t for t in library.tracks.values()
|
|
if t.album.casefold() == album_cf
|
|
and (t.album_artist or t.artist).casefold() == artist_cf]
|
|
|
|
|
|
class _CancelSyncButton(QPushButton):
|
|
"""Compact "✕" in the status bar's right corner that expands to say what
|
|
it does ("cancel transfer") while hovered, then shrinks back. Expanding
|
|
pushes the progress bar and label left; the button itself stays put."""
|
|
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
self.setFlat(True)
|
|
self.setFocusPolicy(Qt.FocusPolicy.NoFocus)
|
|
self.setCursor(Qt.CursorShape.PointingHandCursor)
|
|
self.setToolTip("Cancel the transfer")
|
|
self.setStyleSheet("QPushButton { border: none; padding: 0 4px; }")
|
|
self._apply("✕")
|
|
|
|
def _apply(self, text: str):
|
|
self.setText(text)
|
|
self.setFixedWidth(self.fontMetrics().horizontalAdvance(text) + 14)
|
|
|
|
def enterEvent(self, event):
|
|
self._apply("cancel transfer")
|
|
super().enterEvent(event)
|
|
|
|
def leaveEvent(self, event):
|
|
self._apply("✕")
|
|
super().leaveEvent(event)
|
|
|
|
|
|
class _ResetRelay(QObject):
|
|
"""Carries reset_mtp_connection's answer back to the GUI thread."""
|
|
done = pyqtSignal(object)
|
|
|
|
|
|
class MainWindow(QMainWindow):
|
|
def __init__(self, manager, prefs, lastfm=None, parent=None):
|
|
super().__init__(parent)
|
|
self._manager = manager
|
|
self._prefs = prefs
|
|
self._lastfm = lastfm
|
|
self._prefs_dialog = None
|
|
self.player = Player(manager, self)
|
|
# Owns the cast session, if any. Built before the transport bar, which
|
|
# takes it to drive the little cast indicator under the volume slider.
|
|
self._cast = CastController(self.player, self)
|
|
# Where the Rabbit is, found off the GUI thread: a wedged gvfs-MTP
|
|
# mount blocks any listing of it forever (Round 71). Nothing probes
|
|
# until the Connections menu opens.
|
|
self._devices = device_sync.DeviceWatcher(parent=self)
|
|
self._devices.changed.connect(self._refresh_device_actions)
|
|
# Keep the machine awake while audio is actually playing.
|
|
self._inhibitor = SleepInhibitor()
|
|
# Context playback started from: "library" or "playlist:<pid>". Scopes
|
|
# the now-playing speaker icon to where the song is actually playing.
|
|
self._now_playing_context = ""
|
|
# Friend sharing's menu group and dialogs. Its background service is
|
|
# started by run_gui (host machine only), never by constructing the
|
|
# window — tests build windows by the dozen.
|
|
self._cassette = CassetteUi(self, prefs, manager)
|
|
|
|
self.setWindowTitle("LinTunes")
|
|
self.resize(1280, 800)
|
|
# Keep the window's minimum well under typical sizes so the WM never
|
|
# refuses an interactive shrink
|
|
self.setMinimumSize(560, 420)
|
|
self.setAcceptDrops(True)
|
|
|
|
central = QWidget()
|
|
layout = QVBoxLayout(central)
|
|
layout.setContentsMargins(0, 0, 0, 0)
|
|
layout.setSpacing(0)
|
|
self.setCentralWidget(central)
|
|
|
|
self._transport = TransportBar(self.player, manager, self._prefs,
|
|
cast=self._cast)
|
|
layout.addWidget(self._transport)
|
|
|
|
splitter = QSplitter(Qt.Orientation.Horizontal)
|
|
# The handle is the gap between the playlists and the songs: as wide
|
|
# as the strip left of the playlists, so they sit evenly spaced.
|
|
splitter.setHandleWidth(SIDEBAR_INSET)
|
|
layout.addWidget(splitter, stretch=1)
|
|
|
|
self._sidebar = SidebarPanel(manager, self.player)
|
|
self._sidebar.setMinimumWidth(140)
|
|
splitter.addWidget(self._sidebar)
|
|
|
|
self._content = QStackedWidget()
|
|
# The same strip of bare window on the right as the sidebar keeps on
|
|
# the left, so the lists sit symmetrically in the window.
|
|
self._content.setContentsMargins(0, 0, SIDEBAR_INSET, 0)
|
|
splitter.addWidget(self._content)
|
|
splitter.setSizes([220, 1060])
|
|
splitter.setCollapsible(1, False)
|
|
|
|
self._library_view = LibraryView(manager)
|
|
self._content.addWidget(self._library_view)
|
|
self._playlist_view = PlaylistView(manager)
|
|
self._content.addWidget(self._playlist_view)
|
|
self._content.setCurrentWidget(self._library_view)
|
|
|
|
# The header row under the transport boxes: the Library button over
|
|
# the sidebar, then whichever view is showing's own header (search
|
|
# box, playlist name), then the seek row.
|
|
self._transport.set_sidebar_header(self._sidebar.header)
|
|
self._transport.track_split(self._sidebar.list_area, self._content)
|
|
self._content.currentChanged.connect(self._mount_view_header)
|
|
self._mount_view_header()
|
|
# Moving the Library button out of the sidebar put it last in the
|
|
# focus chain, and a playlist tree that takes the first focus selects
|
|
# its first playlist: the window would open on that, not the library.
|
|
self._sidebar.library_button.setFocus()
|
|
|
|
# Bottom status bar, left to right: version/self-update button, then a
|
|
# centered totals readout. Both are non-permanent widgets, so a
|
|
# transient showMessage (import/scrobble/error/update text) replaces
|
|
# them for its duration and they come back — a permanent widget with
|
|
# stretch would squeeze the message area to zero width and swallow
|
|
# every message.
|
|
self.restart_requested = False # run_gui re-execs when True on quit
|
|
self._missing_folder_prompted = False
|
|
self._updater = Updater(self)
|
|
self._version_button = VersionButton(self._updater)
|
|
self.statusBar().addWidget(self._version_button)
|
|
self._totals_label = QLabel()
|
|
self._totals_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
|
self.statusBar().addWidget(self._totals_label, 1)
|
|
self._version_button.status_message.connect(
|
|
lambda msg: self.statusBar().showMessage(msg, 8000))
|
|
self._updater.update_applied.connect(self._restart_for_update)
|
|
self._updater.start_checking()
|
|
# Device-sync progress, right-justified: permanent widgets sit on the
|
|
# status bar's right and survive showMessage. No stretch (see above)
|
|
# and hidden while idle so it costs no space. Text lives in its own
|
|
# label — QProgressBar's overlay can't fit a sentence at this width.
|
|
# The group is laid out left to right but justified right, so whatever
|
|
# is added *last* owns the corner and never moves. The cancel button
|
|
# goes there deliberately: it is the one widget the user aims at, and
|
|
# the label beside it is as wide as the song title of the moment.
|
|
self._sync_label = QLabel()
|
|
self._sync_label.hide()
|
|
self.statusBar().addPermanentWidget(self._sync_label)
|
|
self._sync_progress = QProgressBar()
|
|
self._sync_progress.setFixedWidth(160)
|
|
self._sync_progress.setTextVisible(False)
|
|
self._sync_progress.hide()
|
|
self.statusBar().addPermanentWidget(self._sync_progress)
|
|
self._sync_cancel = _CancelSyncButton()
|
|
self._sync_cancel.hide()
|
|
self._sync_cancel.clicked.connect(self._confirm_cancel_sync)
|
|
self.statusBar().addPermanentWidget(self._sync_cancel)
|
|
self._andtunes_worker = None
|
|
self._andtunes_planner = None
|
|
self._install_worker = None
|
|
self._sync_device_name = ""
|
|
# Export shares the status-bar progress widgets with device sync —
|
|
# only one of the two may run at a time (each guards on the other).
|
|
self._export_worker = None
|
|
# Identify Track… works through its selection one at a time, each
|
|
# behind its own review dialog.
|
|
self._identify_queue = []
|
|
self._identify_total = 0
|
|
self._identifier = None
|
|
# True from the first lookup until the queue drains, so tracks queued
|
|
# mid-run (Import from URL) wait their turn instead of starting a
|
|
# second lookup alongside.
|
|
self._identify_running = False
|
|
# Import from URL: also shares the status-bar progress widgets.
|
|
self._url_worker = None
|
|
self._url_target = None # url_import.ImportTarget, or None
|
|
self._url_inserted = 0 # songs added to the target so far
|
|
self._url_imported = 0
|
|
self._url_identify = True # False after "Stop Identifying"
|
|
self._url_note = ""
|
|
# Separate instance from the playback inhibitor: pausing music
|
|
# mid-sync must not drop the sync's hold. Logout flag included so
|
|
# GNOME's shutdown/restart dialog names the transfer as the blocker.
|
|
self._sync_inhibitor = SleepInhibitor(
|
|
reason="Syncing a playlist to a device",
|
|
flags=INHIBIT_LOGOUT | INHIBIT_SUSPEND)
|
|
self._export_inhibitor = SleepInhibitor(
|
|
reason="Exporting a playlist",
|
|
flags=INHIBIT_LOGOUT | INHIBIT_SUSPEND)
|
|
|
|
# Wiring
|
|
self._transport.play_clicked.connect(self.play_pause)
|
|
self._sidebar.library_selected.connect(self._show_library)
|
|
self._sidebar.friend_selector_clicked.connect(
|
|
self._cassette.show_friend_menu)
|
|
self._sidebar.playlist_selected.connect(self._show_playlist)
|
|
self._sidebar.tree.export_requested.connect(self._export_playlist)
|
|
self._sidebar.art.download_art_requested.connect(
|
|
lambda tid: self._download_album_art([tid]))
|
|
self._content.currentChanged.connect(self._update_totals)
|
|
for view in (self._library_view, self._playlist_view):
|
|
view.play_requested.connect(
|
|
lambda ids, start, v=view: self._play_from_view(v, ids, start))
|
|
view.info_requested.connect(self._show_info)
|
|
view.show_in_playlist_requested.connect(self._show_track_in_playlist)
|
|
view.table.tracks_changed.connect(self._update_totals)
|
|
view.table.download_art_requested.connect(self._download_album_art)
|
|
view.table.identify_requested.connect(self._identify_tracks)
|
|
view.table.remove_from_library_requested.connect(
|
|
lambda ids: self._delete_tracks(ids, delete_files=False))
|
|
view.table.delete_from_library_requested.connect(
|
|
lambda ids: self._delete_tracks(ids, delete_files=True))
|
|
self._playlist_view.files_dropped.connect(self._import_files_to_playlist)
|
|
manager.tracks_removed.connect(self._on_tracks_removed)
|
|
manager.tracks_restored.connect(self._on_tracks_restored)
|
|
self.player.error_occurred.connect(
|
|
lambda msg: QMessageBox.warning(self, "Can't play track", msg))
|
|
self.player.track_missing.connect(self._on_track_missing)
|
|
self.player.track_changed.connect(self._on_track_changed)
|
|
self.player.playing_changed.connect(self._update_now_playing_icon)
|
|
self.player.playing_changed.connect(self._on_playing_changed)
|
|
manager.track_updated.connect(self._on_track_meta_updated)
|
|
manager.tag_write_failed.connect(
|
|
lambda name, err: self.statusBar().showMessage(
|
|
f"Couldn't write tags to “{name}”: {err}", 8000))
|
|
manager.file_move_failed.connect(
|
|
lambda name, err: self.statusBar().showMessage(
|
|
f"Couldn't move the file for “{name}”: {err}", 8000))
|
|
self.player.track_unplayable.connect(self._on_track_unplayable)
|
|
self._cast.connecting.connect(self._on_cast_connecting)
|
|
self._cast.connected.connect(self._on_cast_connected)
|
|
self._cast.disconnected.connect(self._on_cast_disconnected)
|
|
self._cast.failed.connect(
|
|
lambda msg: QMessageBox.warning(self, "Chromecast", msg))
|
|
manager.library_reloaded.connect(self._on_library_reloaded)
|
|
manager.conflict_resolved.connect(self._on_conflict_resolved)
|
|
prefs.changed.connect(self._on_prefs_changed)
|
|
self._conflict_dialog = None
|
|
if lastfm is not None:
|
|
self.player.track_changed.connect(lastfm.handle_track_started)
|
|
self.player.track_finished.connect(lastfm.handle_track_finished)
|
|
lastfm.status_message.connect(
|
|
lambda msg: self.statusBar().showMessage(msg, 4000))
|
|
|
|
self._build_menus()
|
|
self.apply_ui_metrics()
|
|
self._update_totals()
|
|
|
|
def _restart_for_update(self):
|
|
# The pull already succeeded. Route through close() — not a bare
|
|
# quit() — so closeEvent runs: it flushes the library, tears down the
|
|
# player, and can veto the restart while a device sync is copying.
|
|
self.restart_requested = True
|
|
if self.close():
|
|
QApplication.instance().quit()
|
|
else:
|
|
self.restart_requested = False
|
|
|
|
# ---- status bar totals ----
|
|
|
|
def _update_totals(self):
|
|
view = self._content.currentWidget()
|
|
count, total_ms = view.table.total_stats()
|
|
text = ""
|
|
if count:
|
|
noun = "track" if count == 1 else "tracks"
|
|
text = f"{count:,} {noun} · {format_total_time(total_ms)}"
|
|
# A friend's followed playlist also says when their folder last synced.
|
|
suffix = getattr(view, "status_suffix", None)
|
|
if suffix is not None and suffix():
|
|
text = f"{text} · {suffix()}" if text else suffix()
|
|
self._totals_label.setText(text)
|
|
|
|
# ---- preferences / theming ----
|
|
|
|
def _on_prefs_changed(self):
|
|
theme.apply_theme(QApplication.instance(), self._prefs)
|
|
self.apply_ui_metrics()
|
|
self._transport.refresh_theme()
|
|
self._version_button.refresh_theme()
|
|
self._sidebar.refresh_style()
|
|
|
|
def _mount_view_header(self, _index=None):
|
|
view = self._content.currentWidget()
|
|
self._transport.set_view_header(getattr(view, "header", None))
|
|
|
|
def apply_ui_metrics(self):
|
|
metrics = theme.scale_metrics(self._prefs)
|
|
for view in (self._library_view, self._playlist_view):
|
|
view.table.verticalHeader().setDefaultSectionSize(metrics["row_height"])
|
|
self._sidebar.apply_metrics(metrics["header_strip"])
|
|
self._transport.set_header_height(metrics["header_strip"])
|
|
self.statusBar().setMinimumHeight(metrics["status_height"])
|
|
|
|
def _show_preferences(self):
|
|
if self._prefs_dialog is None:
|
|
self._prefs_dialog = PreferencesDialog(
|
|
self._prefs, self._lastfm, self,
|
|
music_folder=self._music_folder_summary,
|
|
on_music_folder_change=self._change_music_folder)
|
|
else:
|
|
self._prefs_dialog.refresh_music_folder()
|
|
self._prefs_dialog.show()
|
|
self._prefs_dialog.raise_()
|
|
self._prefs_dialog.activateWindow()
|
|
|
|
def _build_menus(self):
|
|
bar = self.menuBar()
|
|
|
|
file_menu = bar.addMenu("&File")
|
|
self._add_action(file_menu, "New Playlist", "Ctrl+N",
|
|
lambda: self._sidebar.tree.create_playlist_interactive())
|
|
self._add_action(file_menu, "New Smart Playlist", "Ctrl+Alt+N",
|
|
lambda: self._sidebar.tree.create_smart_playlist_interactive())
|
|
self._add_action(file_menu, "New Playlist Folder", "Ctrl+Shift+N",
|
|
lambda: self._manager.create_folder("untitled folder"))
|
|
file_menu.addSeparator()
|
|
self._add_action(file_menu, "Add Files to Library…", "Ctrl+O",
|
|
self._open_files_dialog)
|
|
self._url_import_action = self._add_action(
|
|
file_menu, "Import from URL…", "", self._import_from_url)
|
|
self._export_action = self._add_action(
|
|
file_menu, "Export Playlist…", "", self._export_playlist)
|
|
file_menu.aboutToShow.connect(self._refresh_export_action)
|
|
file_menu.aboutToShow.connect(self._refresh_url_import_action)
|
|
file_menu.addSeparator()
|
|
self._add_action(file_menu, "Quit", "Ctrl+Q", self.close)
|
|
|
|
edit_menu = bar.addMenu("&Edit")
|
|
self._undo_action = self._add_action(
|
|
edit_menu, "Undo", QKeySequence.StandardKey.Undo,
|
|
self._manager.undo_stack.undo)
|
|
self._redo_action = self._add_action(
|
|
edit_menu, "Redo", QKeySequence.StandardKey.Redo,
|
|
self._manager.undo_stack.redo)
|
|
# Accept both the platform default (Ctrl+Y) and Ctrl+Shift+Z for redo.
|
|
self._redo_action.setShortcuts([
|
|
QKeySequence(QKeySequence.StandardKey.Redo),
|
|
QKeySequence("Ctrl+Shift+Z"),
|
|
])
|
|
edit_menu.addSeparator()
|
|
self._add_action(edit_menu, "Preferences…", "Ctrl+,", self._show_preferences)
|
|
self._manager.undo_stack.changed.connect(self._refresh_undo_actions)
|
|
self._refresh_undo_actions()
|
|
|
|
controls = bar.addMenu("&Controls")
|
|
self._add_action(controls, "Play/Pause", "", self.play_pause)
|
|
self._add_action(controls, "Next", "", self.player.next)
|
|
self._add_action(controls, "Previous", "", self.player.previous)
|
|
|
|
view_menu = bar.addMenu("&View")
|
|
self._add_action(view_menu, "Toggle Column Browser", "Ctrl+B",
|
|
self._library_view.toggle_browser)
|
|
|
|
track_menu = bar.addMenu("&Track")
|
|
self._add_action(track_menu, "Get Info", "Ctrl+I", self._info_for_current_view)
|
|
self._add_action(track_menu, "Go to Current Song", "Ctrl+L",
|
|
self._go_to_current_song)
|
|
|
|
# Mnemonic on the "n": Alt+C already belongs to &Controls.
|
|
connections_menu = bar.addMenu("Co&nnections")
|
|
self._andtunes_action = self._add_action(
|
|
connections_menu, "Sync to Rabbit", "", self._sync_to_andtunes)
|
|
self._andtunes_action.setEnabled(False)
|
|
self._add_action(connections_menu, "Rabbit Sync Settings…", "",
|
|
self._show_device_sync_settings)
|
|
self._install_action = self._add_action(
|
|
connections_menu, "Install andTunes on Rabbit…", "",
|
|
self._install_andtunes)
|
|
# "Sync Playlist to Rabbit (Auxio)" was retired in Round 51, once
|
|
# andTunes could play what it syncs. device_sync's helpers stay: the
|
|
# export and andTunes code import them.
|
|
# Friend sharing (Cassette) sits between two separators of its own.
|
|
self._cassette.build_menu(connections_menu)
|
|
self._cast_action = self._add_action(
|
|
connections_menu, "Connect to Chromecast…", "", self._toggle_cast)
|
|
# Re-checked every time the menu opens. The device part is a
|
|
# background probe; its answer updates the open menu when it lands.
|
|
connections_menu.aboutToShow.connect(self._refresh_connection_actions)
|
|
|
|
@property
|
|
def cassette(self) -> CassetteUi:
|
|
return self._cassette
|
|
|
|
# What friend mode swaps in and out.
|
|
@property
|
|
def content_stack(self):
|
|
return self._content
|
|
|
|
@property
|
|
def sidebar(self):
|
|
return self._sidebar
|
|
|
|
def play_keys(self, keys, start, context):
|
|
"""Play a followed friend playlist (queue of friend-track keys)."""
|
|
self._now_playing_context = context
|
|
self.player.play_queue(keys, start)
|
|
|
|
@property
|
|
def now_playing_context(self) -> str:
|
|
return self._now_playing_context
|
|
|
|
def update_totals(self):
|
|
self._update_totals()
|
|
|
|
def refresh_library_view(self, _track_ids=None):
|
|
"""Songs arrived from somewhere other than an import dialog (a friend's
|
|
delivery): show them."""
|
|
self._library_view.reload()
|
|
|
|
def show_my_library(self):
|
|
self._sidebar.reflect_library()
|
|
self._show_library()
|
|
|
|
def _add_action(self, menu, text, shortcut, slot):
|
|
action = QAction(text, self)
|
|
if shortcut:
|
|
action.setShortcut(QKeySequence(shortcut))
|
|
action.triggered.connect(slot)
|
|
menu.addAction(action)
|
|
return action
|
|
|
|
def _refresh_undo_actions(self):
|
|
stack = self._manager.undo_stack
|
|
can_undo = stack.can_undo()
|
|
can_redo = stack.can_redo()
|
|
self._undo_action.setEnabled(can_undo)
|
|
self._redo_action.setEnabled(can_redo)
|
|
self._undo_action.setText(
|
|
f"Undo {stack.undo_label()}" if can_undo else "Undo")
|
|
self._redo_action.setText(
|
|
f"Redo {stack.redo_label()}" if can_redo else "Redo")
|
|
|
|
# ---- connections: device sync ----
|
|
|
|
def _refresh_connection_actions(self):
|
|
# An export in flight owns the same status-bar widgets, so it blocks
|
|
# a sync just as another sync would.
|
|
self._devices.probe()
|
|
self._refresh_device_actions()
|
|
self._cassette.refresh_menu()
|
|
# Enabled whenever there's an APK to offer; the click finds the
|
|
# Rabbit (adb or MTP) and says so if it can't — cheaper than asking
|
|
# adb on every menu open.
|
|
self._install_action.setEnabled(
|
|
andtunes_install.bundled_apk() is not None
|
|
and not (self._install_worker is not None
|
|
and self._install_worker.busy()))
|
|
|
|
if self._cast.is_connected():
|
|
self._cast_action.setText(
|
|
f"Disconnect from {self._cast.device_name()}")
|
|
self._cast_action.setEnabled(True)
|
|
self._cast_action.setToolTip("")
|
|
return
|
|
self._cast_action.setText("Connect to Chromecast…")
|
|
available, why = self._cast.is_available()
|
|
self._cast_action.setEnabled(available)
|
|
# The self-updater ships code but not dependencies, so a machine that
|
|
# has pulled this round hasn't necessarily installed pychromecast yet.
|
|
self._cast_action.setToolTip(
|
|
"" if available else f"{why} — run: pip install -e .")
|
|
|
|
# ---- connections: casting ----
|
|
|
|
def _toggle_cast(self):
|
|
if self._cast.is_connected():
|
|
self._cast.disconnect()
|
|
return
|
|
available, why = self._cast.is_available()
|
|
if not available:
|
|
QMessageBox.information(
|
|
self, "Chromecast support not installed",
|
|
f"{why}.\n\nRun “pip install -e .” in the lintunes checkout, "
|
|
"then restart LinTunes.")
|
|
return
|
|
dialog = ChromecastDialog(self._cast.discovery(), self)
|
|
if dialog.exec() and dialog.selected_device is not None:
|
|
self._cast.connect_to(dialog.selected_device)
|
|
|
|
def _on_cast_connecting(self, device):
|
|
self.statusBar().showMessage(f"Connecting to {device.name}…", 8000)
|
|
|
|
def _on_cast_connected(self, name: str):
|
|
self.statusBar().showMessage(f"Casting to {name}", 6000)
|
|
|
|
def _on_cast_disconnected(self, reason: str):
|
|
self.statusBar().showMessage(
|
|
reason or "Stopped casting — playing here again", 8000)
|
|
|
|
def _on_track_unplayable(self, track, reason: str):
|
|
self.statusBar().showMessage(
|
|
f"Skipped “{track.name}” — {reason}", 6000)
|
|
|
|
def _hide_sync_widgets(self):
|
|
self._sync_cancel.hide()
|
|
self._sync_label.hide()
|
|
self._sync_progress.hide()
|
|
|
|
def _busy_worker(self):
|
|
"""The transfer currently owning the status-bar progress widgets."""
|
|
for worker in (self._andtunes_planner, self._andtunes_worker,
|
|
self._export_worker, self._url_worker):
|
|
if worker is not None and worker.busy():
|
|
return worker
|
|
return None
|
|
|
|
def _confirm_cancel_sync(self):
|
|
if self._url_worker is not None and self._url_worker.busy():
|
|
self._confirm_cancel_url_import()
|
|
return
|
|
if self._export_worker is not None and self._export_worker.busy():
|
|
self._confirm_cancel_export()
|
|
return
|
|
worker = self._andtunes_worker
|
|
if worker is None or not worker.busy():
|
|
return
|
|
box = QMessageBox(self)
|
|
box.setWindowTitle("Cancel transfer")
|
|
box.setText(f"Cancel the transfer to the {self._sync_device_name}?\n"
|
|
"Tracks already copied stay on the device; syncing again "
|
|
"later picks up where this left off.")
|
|
cancel_button = box.addButton(
|
|
"Cancel Transfer", QMessageBox.ButtonRole.DestructiveRole)
|
|
keep = box.addButton("Keep Copying", QMessageBox.ButtonRole.RejectRole)
|
|
box.setDefaultButton(keep)
|
|
box.exec()
|
|
# The transfer keeps running while the dialog is up — it may well
|
|
# have finished by now, in which case there is nothing to cancel.
|
|
if box.clickedButton() is cancel_button and worker is not None:
|
|
worker.cancel()
|
|
|
|
def _on_sync_cancelled(self, summary):
|
|
self._sync_inhibitor.release()
|
|
self._hide_sync_widgets()
|
|
self.statusBar().showMessage(
|
|
f"Sync cancelled — {summary['copied']} of {summary['total']} "
|
|
"tracks copied; sync again to finish.", 8000)
|
|
if summary.get("plays"):
|
|
# Counts are brought back before anything is copied, so even a
|
|
# cancelled sync has news for the journals.
|
|
self._manager.reload_from_disk()
|
|
|
|
def _refresh_device_actions(self):
|
|
"""Sync to Rabbit from the watcher's cached answer — never a fresh
|
|
look at the mount, which is what froze the menu."""
|
|
busy = self._busy_worker() is not None
|
|
if self._devices.state == device_sync.STUCK:
|
|
# Enabled: the click explains and offers the reset.
|
|
self._andtunes_action.setText("Rabbit isn't responding…")
|
|
self._andtunes_action.setEnabled(True)
|
|
return
|
|
self._andtunes_action.setText("Sync to Rabbit")
|
|
# Whole-library sync doesn't care which view is open — only that a
|
|
# device is plugged in and at least one playlist is ticked.
|
|
self._andtunes_action.setEnabled(
|
|
not busy and self._devices.device is not None)
|
|
|
|
def _rabbit_stuck(self):
|
|
"""The mount stopped answering: say so, and offer the CLAUDE.md
|
|
recovery (kill gvfsd-mtp, remount) as one button."""
|
|
box = QMessageBox(self)
|
|
box.setIcon(QMessageBox.Icon.Warning)
|
|
box.setWindowTitle("Rabbit isn't responding")
|
|
box.setText("The Rabbit's USB connection has stopped answering, so "
|
|
"LinTunes can't reach it.")
|
|
box.setInformativeText(
|
|
"Reset Connection restarts the desktop's MTP service and "
|
|
"remounts the Rabbit. If that doesn't do it, unplug the Rabbit "
|
|
"and plug it back in.")
|
|
reset = box.addButton("Reset Connection",
|
|
QMessageBox.ButtonRole.AcceptRole)
|
|
box.addButton(QMessageBox.StandardButton.Cancel)
|
|
box.setDefaultButton(reset)
|
|
box.exec()
|
|
if box.clickedButton() is not reset:
|
|
return
|
|
self.statusBar().showMessage("Resetting the Rabbit connection…")
|
|
relay = self._reset_relay = _ResetRelay(self)
|
|
relay.done.connect(self._on_rabbit_reset)
|
|
threading.Thread(
|
|
target=lambda: relay.done.emit(device_sync.reset_mtp_connection()),
|
|
daemon=True).start()
|
|
|
|
def _on_rabbit_reset(self, error):
|
|
if error:
|
|
self.statusBar().clearMessage()
|
|
QMessageBox.warning(self, "Rabbit isn't responding",
|
|
f"The reset didn't work: {error}\n\n"
|
|
"Unplug the Rabbit and plug it back in.")
|
|
return
|
|
self.statusBar().showMessage("Rabbit connection reset", 6000)
|
|
self._devices.probe()
|
|
|
|
def _on_sync_failed(self, message):
|
|
self._sync_inhibitor.release()
|
|
self._hide_sync_widgets()
|
|
QMessageBox.warning(self, "Sync failed", message)
|
|
|
|
# ---- connections: andTunes (whole-library sync) ----
|
|
|
|
def _show_device_sync_settings(self):
|
|
device = self._devices.device # only for its name
|
|
dialog = DeviceSyncSettingsDialog(
|
|
self._prefs, self._manager.library, self,
|
|
device_name=device.name if device else "Rabbit")
|
|
dialog.exec()
|
|
|
|
def _selected_sync_playlists(self) -> list:
|
|
"""The ticked playlists (and everything in ticked folders)."""
|
|
return resolve_sync_playlists(self._manager.library,
|
|
self._prefs.device_sync)
|
|
|
|
def _sync_to_andtunes(self):
|
|
# Everything may have changed since the menu opened — re-verify.
|
|
if self._busy_worker() is not None:
|
|
return
|
|
self._devices.when_ready(self._sync_to_device)
|
|
|
|
def _sync_to_device(self, device, state):
|
|
if state == device_sync.STUCK:
|
|
self._rabbit_stuck()
|
|
return
|
|
if device is None or self._busy_worker() is not None:
|
|
return
|
|
playlists = self._selected_sync_playlists()
|
|
if not playlists:
|
|
QMessageBox.information(
|
|
self, "Nothing to sync",
|
|
"No playlists are set to sync yet.\n\n"
|
|
"Pick them in Connections → Rabbit Sync Settings…")
|
|
return
|
|
|
|
# Planning walks the whole device over MTP — off the GUI thread, or
|
|
# the window freezes and GNOME offers to kill it mid-sync.
|
|
root = andtunes_layout.root_for(device)
|
|
self._andtunes_planner = andtunes_sync.AndTunesPlanWorker(
|
|
playlists, self._manager.library.tracks, root, device,
|
|
copy_art=bool(self._prefs.device_sync.get("copy_art", True)),
|
|
data_dir=self._manager.data_dir, parent=self)
|
|
self._andtunes_planner.planned.connect(self._on_andtunes_planned)
|
|
self._andtunes_planner.failed.connect(self._on_sync_failed)
|
|
self._sync_device_name = device.name
|
|
self._sync_label.setText(f"Checking the {device.name}")
|
|
self._sync_progress.setRange(0, 0) # busy bar: no total to show yet
|
|
self._sync_label.show()
|
|
self._sync_progress.show()
|
|
self._andtunes_planner.start()
|
|
|
|
def _on_andtunes_planned(self, plan, free):
|
|
device = plan.device
|
|
playlists = plan.playlists
|
|
self._hide_sync_widgets()
|
|
if plan.needs_ffmpeg and not exporter.web_support.ffmpeg_available():
|
|
conversions = plan.conversions
|
|
names = "\n".join(f" • {device_sync.track_display(i.track)}"
|
|
for i in conversions[:8])
|
|
more = (f"\n … and {len(conversions) - 8} more"
|
|
if len(conversions) > 8 else "")
|
|
answer = QMessageBox.question(
|
|
self, "FFmpeg not found",
|
|
f"{len(conversions)} song(s) need converting to FLAC before "
|
|
f"the {device.name} can play them, but the ffmpeg command "
|
|
f"isn't installed:\n\n{names}{more}\n\nSync without them?")
|
|
if answer != QMessageBox.StandardButton.Yes:
|
|
return
|
|
plan.drop_conversions("it needs ffmpeg to convert")
|
|
needed = plan.bytes_to_copy + device_sync.SPACE_MARGIN
|
|
if free + plan.bytes_freed < needed:
|
|
QMessageBox.warning(
|
|
self, "Not enough space",
|
|
f"Syncing {len(playlists)} playlist(s) needs "
|
|
f"{device_sync.format_bytes(needed - plan.bytes_freed)}, but "
|
|
f"the {device.name} only has "
|
|
f"{device_sync.format_bytes(free)} free.")
|
|
return
|
|
|
|
self._andtunes_worker = andtunes_sync.AndTunesSyncWorker(plan, self)
|
|
self._andtunes_worker.progress.connect(self._on_andtunes_progress)
|
|
self._andtunes_worker.finished.connect(self._on_andtunes_finished)
|
|
self._andtunes_worker.cancelled.connect(self._on_sync_cancelled)
|
|
self._andtunes_worker.failed.connect(self._on_sync_failed)
|
|
self._sync_device_name = device.name
|
|
self._sync_label.setText(f"Copying to {device.name}")
|
|
self._sync_progress.setRange(0, max(plan.bytes_to_copy // 1024, 1))
|
|
self._sync_progress.setValue(0)
|
|
self._sync_cancel.show()
|
|
self._sync_label.show()
|
|
self._sync_progress.show()
|
|
self._sync_inhibitor.inhibit()
|
|
self._andtunes_worker.start()
|
|
|
|
def _on_andtunes_progress(self, done_kib, total_kib, label):
|
|
# Unlike the per-playlist sync this run has phases that move no audio
|
|
# bytes (album art, then the index), so the label carries the story.
|
|
self._sync_progress.setRange(0, total_kib)
|
|
self._sync_progress.setValue(done_kib)
|
|
self._sync_label.setText(f"{self._sync_device_name} · {label}")
|
|
|
|
def _on_andtunes_finished(self, summary):
|
|
self._sync_inhibitor.release()
|
|
self._hide_sync_widgets()
|
|
msg = (f"Synced {summary['playlists']} playlist(s) to the "
|
|
f"{summary['device']}: {summary['tracks']} tracks "
|
|
f"({summary['copied']} copied, {summary['kept']} up to date), "
|
|
f"{summary['removed']} removed")
|
|
if summary.get("covers_copied"):
|
|
msg += f", {summary['covers_copied']} covers updated"
|
|
if summary["skipped"]:
|
|
msg += f", {summary['skipped']} skipped (no local file)"
|
|
if summary.get("refused"):
|
|
msg += f", {summary['refused']} left out (can't play there)"
|
|
if summary.get("unconverted"):
|
|
msg += f", {summary['unconverted']} couldn't be converted"
|
|
unwritable = summary.get("unwritable") or []
|
|
if unwritable:
|
|
msg += f", {len(unwritable)} the device wouldn't take"
|
|
if summary.get("plays"):
|
|
msg += f"; play counts back for {summary['plays']} song(s)"
|
|
# The journal landed in the data dir; fold it in now rather than
|
|
# waiting for the file watcher.
|
|
self._manager.reload_from_disk()
|
|
self.statusBar().showMessage(msg, 10000)
|
|
sounds.play_done(self._prefs)
|
|
if unwritable:
|
|
# Worth a dialog: these songs are *not* in the library the app
|
|
# reads, so they'd otherwise just be quietly absent from the
|
|
# playlist. A stale gvfs mount is the usual cause.
|
|
names = "\n".join(f" • {name} — {why}" for name, why in unwritable[:8])
|
|
more = (f"\n … and {len(unwritable) - 8} more"
|
|
if len(unwritable) > 8 else "")
|
|
QMessageBox.warning(
|
|
self, "Some songs wouldn't copy",
|
|
f"The {summary['device']} refused "
|
|
f"{len(unwritable)} song(s):\n\n{names}{more}\n\n"
|
|
"They've been left out of the library on the device, so the "
|
|
"app won't list a song it can't play. Unplugging the "
|
|
"device and plugging it back in usually clears this.")
|
|
|
|
# ---- connections: install the andTunes app ----
|
|
|
|
def _install_andtunes(self):
|
|
if self._install_worker is not None and self._install_worker.busy():
|
|
return
|
|
apk = andtunes_install.bundled_apk()
|
|
if apk is None:
|
|
return
|
|
self._devices.when_ready(
|
|
lambda device, state: self._install_andtunes_to(apk, device, state))
|
|
|
|
def _install_andtunes_to(self, apk, device, state):
|
|
adb = andtunes_install.adb_path()
|
|
serial = andtunes_install.rabbit_serial(adb) if adb else None
|
|
if serial is None and state == device_sync.STUCK:
|
|
self._rabbit_stuck()
|
|
return
|
|
if serial is None and device is None:
|
|
QMessageBox.information(
|
|
self, "Rabbit not found",
|
|
"Plug the Rabbit in with USB debugging on, or in "
|
|
"File transfer mode, then try again.")
|
|
return
|
|
|
|
ours = andtunes_install.bundled_version()
|
|
ours_name = ours.get("version_name", "?")
|
|
if serial:
|
|
have = andtunes_install.installed_version(adb, serial)
|
|
if have is None:
|
|
verb = "Install"
|
|
detail = f"Install andTunes {ours_name} on the Rabbit?"
|
|
elif have["version_code"] < ours.get("version_code", 0):
|
|
verb = "Update"
|
|
detail = (f"The Rabbit has andTunes {have['version_name']}; "
|
|
f"this LinTunes carries {ours_name}. Update it?")
|
|
else:
|
|
verb = "Reinstall"
|
|
detail = (f"The Rabbit already has andTunes "
|
|
f"{have['version_name']}. Reinstall {ours_name}?")
|
|
else:
|
|
verb = "Copy"
|
|
detail = ("adb isn't available, so the app will be copied to "
|
|
"Download/ on the Rabbit for you to open in the Files "
|
|
"app and install by hand. Copy it?")
|
|
box = QMessageBox(self)
|
|
box.setWindowTitle(f"{verb} andTunes")
|
|
box.setText(detail)
|
|
go = box.addButton(verb, QMessageBox.ButtonRole.AcceptRole)
|
|
box.addButton(QMessageBox.StandardButton.Cancel)
|
|
box.setDefaultButton(go)
|
|
box.exec()
|
|
if box.clickedButton() is not go:
|
|
return
|
|
|
|
self._install_worker = andtunes_install.InstallWorker(
|
|
apk, adb, serial, device, self)
|
|
self._install_worker.finished.connect(self._on_install_finished)
|
|
self._install_worker.failed.connect(
|
|
lambda message: QMessageBox.warning(self, "andTunes", message))
|
|
self.statusBar().showMessage("Installing andTunes on the Rabbit…")
|
|
self._install_worker.start()
|
|
|
|
def _on_install_finished(self, result):
|
|
if result["method"] == "copied":
|
|
self.statusBar().clearMessage()
|
|
QMessageBox.information(
|
|
self, "andTunes copied",
|
|
f"Copied to {result['path']}.\n\nOn the Rabbit, open Files → "
|
|
"Download → andTunes.apk to install it.")
|
|
return
|
|
msg = "andTunes is installed on the Rabbit"
|
|
if not result.get("granted", True):
|
|
msg += " — open it once to allow file access"
|
|
self.statusBar().showMessage(msg, 8000)
|
|
|
|
# ---- export ----
|
|
|
|
def _refresh_export_action(self):
|
|
"""Export needs a playlist in view and no transfer already running."""
|
|
self._export_action.setEnabled(
|
|
self._busy_worker() is None
|
|
and self._content.currentWidget() is self._playlist_view
|
|
and bool(self._playlist_view.playlist_id))
|
|
|
|
def _export_playlist(self, pid: str = ""):
|
|
# Reachable from the File menu (no argument) and from the sidebar
|
|
# context menu (which names the playlist that was right-clicked).
|
|
if self._busy_worker() is not None:
|
|
return
|
|
pid = pid or self._playlist_view.playlist_id
|
|
playlist = self._manager.library.playlists.get(pid)
|
|
if playlist is None:
|
|
return
|
|
tracks = [self._manager.library.tracks[tid]
|
|
for tid in playlist.track_ids
|
|
if tid in self._manager.library.tracks]
|
|
if not tracks:
|
|
QMessageBox.information(
|
|
self, "Export Playlist",
|
|
f"“{playlist.name}” is empty — there is nothing to export.")
|
|
return
|
|
|
|
kind_dialog = ExportKindDialog(playlist.name, self)
|
|
if kind_dialog.exec() != ExportKindDialog.DialogCode.Accepted:
|
|
return
|
|
kind = kind_dialog.kind()
|
|
|
|
details = None
|
|
if kind == exporter.WEB:
|
|
web_dialog = WebMixDialog(playlist.name, self)
|
|
if web_dialog.exec() != WebMixDialog.DialogCode.Accepted:
|
|
return
|
|
details = web_dialog.values()
|
|
|
|
start_dir = self._music_import_dir()
|
|
parent_dir = QFileDialog.getExistingDirectory(
|
|
self, "Export To",
|
|
str(start_dir.parent) if start_dir else str(Path.home()))
|
|
if not parent_dir:
|
|
return
|
|
|
|
try:
|
|
plan = exporter.plan_export(
|
|
playlist.name, tracks, Path(parent_dir), kind)
|
|
except OSError as e:
|
|
QMessageBox.warning(self, "Export failed",
|
|
f"Couldn't read the playlist's files: {e}")
|
|
return
|
|
if details is not None:
|
|
plan.title = details["title"] or playlist.name
|
|
plan.description = details["description"]
|
|
plan.image = details["image"]
|
|
plan.accent = details["accent"]
|
|
|
|
if not self._confirm_export(plan):
|
|
return
|
|
|
|
self._export_worker = exporter.ExportWorker(plan, self)
|
|
self._export_worker.progress.connect(self._on_export_progress)
|
|
self._export_worker.finished.connect(self._on_export_finished)
|
|
self._export_worker.cancelled.connect(self._on_export_cancelled)
|
|
self._export_worker.failed.connect(self._on_export_failed)
|
|
self._sync_label.setText("Exporting")
|
|
self._sync_progress.setRange(0, max(plan.bytes_to_copy // 1024, 1))
|
|
self._sync_progress.setValue(0)
|
|
self._sync_cancel.show()
|
|
self._sync_label.show()
|
|
self._sync_progress.show()
|
|
self._export_inhibitor.inhibit()
|
|
self._export_worker.start()
|
|
|
|
def _confirm_export(self, plan) -> bool:
|
|
"""Last look before writing: overwrite, ffmpeg, and skipped tracks."""
|
|
if plan.needs_ffmpeg and not exporter.web_support.ffmpeg_available():
|
|
names = "\n".join(f" • {i.display}" for i in plan.conversions[:8])
|
|
more = (f"\n … and {len(plan.conversions) - 8} more"
|
|
if len(plan.conversions) > 8 else "")
|
|
answer = QMessageBox.question(
|
|
self, "FFmpeg not found",
|
|
f"{len(plan.conversions)} track(s) need converting before a "
|
|
f"browser can play them, but the ffmpeg command isn't "
|
|
f"installed:\n\n{names}{more}\n\nExport without them?")
|
|
if answer != QMessageBox.StandardButton.Yes:
|
|
return False
|
|
plan.skipped += [(i.display, "it needs ffmpeg to convert")
|
|
for i in plan.conversions]
|
|
plan.items = [i for i in plan.items if not i.convert_to]
|
|
plan.bytes_to_copy = sum(i.size for i in plan.items)
|
|
if not plan.items:
|
|
QMessageBox.information(
|
|
self, "Export Playlist",
|
|
"Nothing left to export once those are left out.")
|
|
return False
|
|
|
|
if plan.skipped:
|
|
names = "\n".join(f" • {d} — {why}" for d, why in plan.skipped[:8])
|
|
more = (f"\n … and {len(plan.skipped) - 8} more"
|
|
if len(plan.skipped) > 8 else "")
|
|
answer = QMessageBox.question(
|
|
self, "Some tracks can't be exported",
|
|
f"{len(plan.skipped)} of {len(plan.skipped) + len(plan.items)} "
|
|
f"track(s) will be left out:\n\n{names}{more}\n\nExport the "
|
|
f"other {len(plan.items)}?")
|
|
if answer != QMessageBox.StandardButton.Yes:
|
|
return False
|
|
|
|
if plan.dest_dir.exists():
|
|
answer = QMessageBox.question(
|
|
self, "Folder exists",
|
|
f"“{plan.dest_dir.name}” already exists in that folder. "
|
|
f"Files with the same names will be overwritten.\n\nContinue?")
|
|
if answer != QMessageBox.StandardButton.Yes:
|
|
return False
|
|
return True
|
|
|
|
def _confirm_cancel_export(self):
|
|
if self._export_worker is None or not self._export_worker.busy():
|
|
return
|
|
box = QMessageBox(self)
|
|
box.setWindowTitle("Cancel export")
|
|
box.setText("Cancel the export?\n"
|
|
"The folder is left as-is, without a playlist file — "
|
|
"exporting again starts it over.")
|
|
cancel_button = box.addButton(
|
|
"Cancel Export", QMessageBox.ButtonRole.DestructiveRole)
|
|
keep = box.addButton("Keep Exporting", QMessageBox.ButtonRole.RejectRole)
|
|
box.setDefaultButton(keep)
|
|
box.exec()
|
|
# The export keeps running while the dialog is up — it may well have
|
|
# finished by now, in which case there is nothing to cancel.
|
|
if (box.clickedButton() is cancel_button
|
|
and self._export_worker is not None):
|
|
self._export_worker.cancel()
|
|
|
|
def _on_export_progress(self, done_kib, total_kib, label):
|
|
self._sync_progress.setRange(0, total_kib)
|
|
self._sync_progress.setValue(done_kib)
|
|
self._sync_label.setText(
|
|
f"Exporting · {device_sync.format_bytes(done_kib * 1024)} / "
|
|
f"{device_sync.format_bytes(total_kib * 1024)}")
|
|
|
|
def _on_export_finished(self, summary):
|
|
self._export_inhibitor.release()
|
|
self._hide_sync_widgets()
|
|
what = "web mix" if summary["kind"] == exporter.WEB else "folder"
|
|
msg = (f"Exported “{summary['playlist']}” as a {what}: "
|
|
f"{summary['exported']} tracks → {summary['dest']}")
|
|
if summary["converted"]:
|
|
msg += f", {summary['converted']} converted to FLAC"
|
|
if summary["skipped"]:
|
|
msg += f", {summary['skipped']} skipped"
|
|
if summary["vanished"]:
|
|
msg += (f", {summary['vanished']} changed under us "
|
|
"(export again to pick them up)")
|
|
self.statusBar().showMessage(msg, 12000)
|
|
sounds.play_done(self._prefs)
|
|
|
|
def _on_export_cancelled(self, summary):
|
|
self._export_inhibitor.release()
|
|
self._hide_sync_widgets()
|
|
self.statusBar().showMessage(
|
|
f"Export cancelled — {summary['exported']} of {summary['total']} "
|
|
f"tracks written to {summary['dest']}.", 8000)
|
|
|
|
def _on_export_failed(self, message):
|
|
self._export_inhibitor.release()
|
|
self._hide_sync_widgets()
|
|
QMessageBox.warning(self, "Export failed", message)
|
|
|
|
# ---- view switching ----
|
|
|
|
def _show_library(self):
|
|
self._content.setCurrentWidget(self._library_view)
|
|
self._update_now_playing_icon()
|
|
|
|
def _show_playlist(self, persistent_id: str):
|
|
playlist = self._manager.library.playlists.get(persistent_id)
|
|
if playlist:
|
|
self._playlist_view.show_playlist(playlist)
|
|
self._content.setCurrentWidget(self._playlist_view)
|
|
self._update_now_playing_icon()
|
|
|
|
def _go_to_current_song(self):
|
|
"""Jump to wherever the current song is playing from, select it, and
|
|
scroll it into view. No-op when nothing is cued."""
|
|
track = self.player.current_track
|
|
if track is None:
|
|
return
|
|
ctx = self._now_playing_context
|
|
if ctx == "library":
|
|
self._show_library()
|
|
self._sidebar.reflect_library()
|
|
self._library_view.reveal_track(track.track_id)
|
|
elif ctx.startswith("playlist:"):
|
|
pid = ctx.split(":", 1)[1]
|
|
if pid in self._manager.library.playlists:
|
|
self._show_playlist(pid)
|
|
self._sidebar.reflect_playlist(pid)
|
|
self._playlist_view.reveal_track(track.track_id)
|
|
|
|
def _show_track_in_playlist(self, track_id: int, pid: str):
|
|
"""Jump to a playlist and highlight the first instance of the track,
|
|
reusing the Go-to-Current-Song navigation pattern."""
|
|
if pid in self._manager.library.playlists:
|
|
self._show_playlist(pid)
|
|
self._sidebar.reflect_playlist(pid)
|
|
self._playlist_view.reveal_track(track_id)
|
|
|
|
# ---- multi-machine sync ----
|
|
|
|
def _on_library_reloaded(self):
|
|
"""Another machine's changes were reconciled in. Refresh the library
|
|
view (the sidebar and open playlist refresh via playlists_changed),
|
|
keeping the user's scroll and selection. Never touches the player."""
|
|
table = self._library_view.table
|
|
selected = table.selected_track_ids()
|
|
scroll = table.verticalScrollBar().value()
|
|
self._library_view.reload()
|
|
if selected:
|
|
table.reveal_track(selected[0])
|
|
table.verticalScrollBar().setValue(scroll)
|
|
|
|
def _on_conflict_resolved(self, summaries):
|
|
self.statusBar().showMessage(
|
|
f"Merged changes from your other machine "
|
|
f"({len(summaries)} item(s))", 8000)
|
|
self.show_conflict_summary(summaries)
|
|
|
|
def show_conflict_summary(self, summaries):
|
|
"""Show (or extend) the one merge window. Syncthing can deliver conflicts
|
|
every few minutes; each merge becomes another entry in the open window
|
|
rather than another window.
|
|
|
|
Only raise it when the new batch has something to say at the level on
|
|
screen — otherwise the entry is folded in quietly and is there when the
|
|
user dials the filter down."""
|
|
from lintunes.gui.conflict_dialog import ConflictSummaryDialog
|
|
if self._conflict_dialog is not None:
|
|
if self._conflict_dialog.add_event(summaries):
|
|
self._conflict_dialog.raise_()
|
|
return
|
|
# Modeless so a merge never blocks playback or what you're doing.
|
|
self._conflict_dialog = ConflictSummaryDialog(
|
|
summaries, self._manager, self)
|
|
self._conflict_dialog.finished.connect(self._on_conflict_dialog_closed)
|
|
self._conflict_dialog.show()
|
|
self._conflict_dialog.raise_()
|
|
|
|
def _on_conflict_dialog_closed(self, _result):
|
|
"""Drop the reference (and the widget) so the next merge opens a fresh
|
|
window instead of touching a deleted C++ object."""
|
|
dialog, self._conflict_dialog = self._conflict_dialog, None
|
|
if dialog is not None:
|
|
dialog.deleteLater()
|
|
|
|
# ---- info dialog ----
|
|
|
|
def _show_info(self, track_ids: list[int]):
|
|
if track_ids:
|
|
InfoDialog(self._manager, track_ids, self).exec()
|
|
|
|
def _info_for_current_view(self):
|
|
view = self._content.currentWidget()
|
|
ids = view.table.selected_track_ids()
|
|
self._show_info(ids)
|
|
|
|
# ---- music folder ----
|
|
|
|
def check_music_folder(self):
|
|
"""Ask where the music goes, if we don't already know.
|
|
|
|
Called once shortly after the window appears, and again lazily if an
|
|
import happens while no folder is set. Never blocks startup: by the
|
|
time this runs the main window is already up, so the dialog is a normal
|
|
child dialog rather than a popup in front of an empty screen.
|
|
"""
|
|
state = self._manager.music_folder_state()
|
|
if state.path is not None:
|
|
return
|
|
if state.source == "unset":
|
|
if music_folder.prompt_seen():
|
|
return
|
|
elif self._missing_folder_prompted:
|
|
return # asked once this session; don't nag on every import
|
|
|
|
dialog = MusicFolderDialog(
|
|
self, missing_path=state.stored if state.source == "missing" else "")
|
|
dialog.exec()
|
|
chosen = dialog.chosen_folder()
|
|
|
|
# Remember we asked, whatever the answer — including "Not Now".
|
|
music_folder.mark_prompt_seen()
|
|
self._missing_folder_prompted = True
|
|
|
|
if chosen is not None:
|
|
# A folder that exists elsewhere but not here is this machine's own
|
|
# problem to solve; don't overwrite what the other machine set.
|
|
self._manager.set_music_folder(
|
|
chosen, machine_only=(state.source == "missing"))
|
|
self.statusBar().showMessage(
|
|
f"Music folder set to {self._manager.organize_root()}", 6000)
|
|
|
|
def _change_music_folder(self):
|
|
"""Preferences ▸ Music folder ▸ Change… — pick a new one."""
|
|
state = self._manager.music_folder_state()
|
|
start = str(state.path) if state.path else str(Path.home())
|
|
chosen = QFileDialog.getExistingDirectory(
|
|
self, "Choose Music Folder", start)
|
|
if not chosen:
|
|
return
|
|
|
|
media = music_folder.normalize_media_folder(chosen)
|
|
old_root = self._manager.organize_root()
|
|
if old_root is not None and old_root.exists() and \
|
|
media != Path(state.path or ""):
|
|
# Say the quiet part out loud: nothing moves, and from here on only
|
|
# new music goes to the new place. Otherwise the change looks total
|
|
# but silently applies to future additions only.
|
|
answer = QMessageBox.question(
|
|
self, "Change music folder?",
|
|
f"Your existing songs will stay where they are:\n{old_root}\n\n"
|
|
f"Only music you add from now on will go to:\n"
|
|
f"{music_folder.organize_root_for(media)}\n\n"
|
|
"Change the folder?",
|
|
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
|
|
QMessageBox.StandardButton.No)
|
|
if answer != QMessageBox.StandardButton.Yes:
|
|
return
|
|
|
|
self._manager.set_music_folder(
|
|
media, machine_only=state.is_machine_local)
|
|
self.statusBar().showMessage(
|
|
f"Music folder set to {self._manager.organize_root()}", 6000)
|
|
|
|
def _music_folder_summary(self) -> tuple[str, str]:
|
|
"""(folder, note) for the Preferences row."""
|
|
state = self._manager.music_folder_state()
|
|
if state.path is None:
|
|
return ("", "Not set yet — pick a folder to start adding music.")
|
|
if state.is_machine_local:
|
|
return (str(state.path), "Used on this computer only.")
|
|
return (str(state.path), "Shared with your other computers.")
|
|
|
|
# ---- file import ----
|
|
|
|
def _music_import_dir(self) -> Path | None:
|
|
"""Where imports are filed, or None when no music folder is usable.
|
|
|
|
Deliberately no fallback. The old `or Path("Music")` was *relative*, so
|
|
it resolved against the process working directory — which GNOME's dash
|
|
does not set predictably (see packaging/install-desktop.sh), scattering
|
|
a new user's music somewhere they'd never find it.
|
|
"""
|
|
return self._manager.organize_root()
|
|
|
|
def _open_files_dialog(self):
|
|
paths, _filter = QFileDialog.getOpenFileNames(
|
|
self, "Add Files to Library", "",
|
|
"Audio files (*.mp3 *.m4a *.flac *.aac *.wav *.aiff)")
|
|
if paths:
|
|
self.import_files([Path(p) for p in paths])
|
|
|
|
def _ensure_music_dir(self) -> Path | None:
|
|
music_dir = self._music_import_dir()
|
|
if music_dir is None:
|
|
# Ask rather than guess — better one question than music copied
|
|
# somewhere unfindable.
|
|
self.check_music_folder()
|
|
music_dir = self._music_import_dir()
|
|
if music_dir is None:
|
|
self.statusBar().showMessage(
|
|
"Choose where to keep your music first — Edit ▸ Preferences.",
|
|
8000)
|
|
return music_dir
|
|
|
|
def import_files(self, paths, playlist_pid: str = "",
|
|
position: int | None = None) -> list:
|
|
music_dir = self._ensure_music_dir()
|
|
if music_dir is None:
|
|
return []
|
|
self.statusBar().showMessage(f"Importing {len(paths)} item(s)…")
|
|
imported = file_importer.import_paths(
|
|
paths, music_dir, self._manager, playlist_pid,
|
|
position)
|
|
self.statusBar().showMessage(f"Imported {len(imported)} track(s)", 5000)
|
|
if imported:
|
|
self._library_view.reload()
|
|
return imported
|
|
|
|
def _import_files_to_playlist(self, paths, pid, position=None):
|
|
self.import_files([Path(p) for p in paths], pid, position)
|
|
|
|
# ---- importing from a link (trav's `song` helper, in-process) ----
|
|
|
|
def _refresh_url_import_action(self):
|
|
"""One transfer owns the status-bar progress widgets at a time."""
|
|
self._url_import_action.setEnabled(self._busy_worker() is None)
|
|
|
|
def _url_import_target(self):
|
|
shown, row = "", None
|
|
if self._content.currentWidget() is self._playlist_view:
|
|
shown = self._playlist_view.playlist_id
|
|
rows = self._playlist_view.table.selected_source_rows()
|
|
row = rows[0] if rows else None
|
|
return url_import.resolve_target(
|
|
self._manager.library, shown, row, self._now_playing_context)
|
|
|
|
def _import_from_url(self):
|
|
if self._busy_worker() is not None:
|
|
return
|
|
missing = [name for name, ok in (
|
|
("yt-dlp", url_import.ytdlp_available()),
|
|
("ffmpeg", exporter.web_support.ffmpeg_available())) if not ok]
|
|
if missing:
|
|
QMessageBox.warning(
|
|
self, "Import from URL",
|
|
f"Importing from a link needs {' and '.join(missing)}, "
|
|
f"which {'is' if len(missing) == 1 else 'are'} not "
|
|
f"installed.\n\n"
|
|
"yt-dlp: pip install --user yt-dlp (or your distro's "
|
|
"package)\nffmpeg: your distro's “ffmpeg” package")
|
|
return
|
|
target = self._url_import_target()
|
|
dialog = UrlImportDialog(target.description if target else None, self)
|
|
if not dialog.exec():
|
|
return
|
|
if self._ensure_music_dir() is None:
|
|
return
|
|
self._url_target = target if dialog.add_to_playlist() else None
|
|
self._url_inserted = 0
|
|
self._url_imported = 0
|
|
self._url_identify = True
|
|
problem = self._identify_setup_problem()
|
|
self._url_note = (f"tags proposed from filenames only ({problem})"
|
|
if problem else "")
|
|
|
|
worker = url_import.UrlImportWorker(
|
|
dialog.url(), self,
|
|
cookies_browser=load_config().get(url_import.COOKIES_KEY))
|
|
worker.cookies_used.connect(self._remember_cookies_browser)
|
|
worker.item_started.connect(self._on_url_item_started)
|
|
worker.progress.connect(self._sync_progress.setValue)
|
|
worker.downloaded.connect(self._on_url_downloaded)
|
|
worker.finished.connect(self._on_url_finished)
|
|
worker.failed.connect(self._on_url_failed)
|
|
self._url_worker = worker
|
|
self._sync_label.setText("Downloading…")
|
|
self._sync_progress.setRange(0, 100)
|
|
self._sync_progress.setValue(0)
|
|
self._sync_cancel.show()
|
|
self._sync_label.show()
|
|
self._sync_progress.show()
|
|
worker.start()
|
|
|
|
def _remember_cookies_browser(self, browser: str):
|
|
"""A bot-checked link got through on this browser's cookies: send
|
|
them from the start next time. config.json, not preferences — the
|
|
browser holding a YouTube login is this machine's business."""
|
|
config = load_config()
|
|
if config.get(url_import.COOKIES_KEY) != browser:
|
|
config[url_import.COOKIES_KEY] = browser
|
|
save_config(config)
|
|
|
|
def _on_url_item_started(self, index: int, count: int, title: str):
|
|
short = title if len(title) <= 40 else title[:39] + "…"
|
|
where = f"{index} of {count} · " if count > 1 else ""
|
|
self._sync_label.setText(f"Downloading {where}{short}")
|
|
self._sync_progress.setValue(0)
|
|
|
|
def _on_url_downloaded(self, path: str):
|
|
"""One song is off the network: file it and identify it now, so the
|
|
first dialog can be up while the rest are still downloading."""
|
|
target = self._url_target
|
|
pid, position = "", None
|
|
if target is not None and target.pid in self._manager.library.playlists:
|
|
pid = target.pid
|
|
if target.position is not None:
|
|
# Each song goes after the ones already added, so the batch
|
|
# keeps the link's order, all of it above the selected song.
|
|
position = target.position + self._url_inserted
|
|
imported = self.import_files([Path(path)], pid, position)
|
|
if pid:
|
|
self._url_inserted += len(imported)
|
|
self._url_imported += len(imported)
|
|
if imported and self._url_identify:
|
|
self._enqueue_identify(imported)
|
|
|
|
def _finish_url_import(self):
|
|
if self._url_worker is not None:
|
|
self._url_worker.cleanup()
|
|
self._hide_sync_widgets()
|
|
|
|
def _on_url_finished(self, summary: dict):
|
|
self._finish_url_import()
|
|
count = self._url_imported
|
|
if summary["cancelled"]:
|
|
msg = f"Import cancelled — {count} song(s) imported"
|
|
else:
|
|
msg = f"Imported {count} song(s) from the link"
|
|
sounds.play_done(self._prefs)
|
|
target = self._url_target
|
|
playlist = (self._manager.library.playlists.get(target.pid)
|
|
if target is not None else None)
|
|
if playlist is not None and count:
|
|
msg += f" into “{playlist.name}”"
|
|
if summary["errors"]:
|
|
msg += f" · {len(summary['errors'])} couldn't be downloaded"
|
|
if self._url_note and count and self._url_identify:
|
|
msg += f" · {self._url_note}"
|
|
self.statusBar().showMessage(msg, 12000)
|
|
|
|
def _on_url_failed(self, message: str):
|
|
self._finish_url_import()
|
|
QMessageBox.warning(self, "Import from URL",
|
|
f"Couldn't import from that link:\n\n{message}")
|
|
|
|
def _confirm_cancel_url_import(self):
|
|
box = QMessageBox(self)
|
|
box.setWindowTitle("Stop downloading")
|
|
box.setText("Stop downloading from the link?\n"
|
|
"Songs already imported stay in your library.")
|
|
stop = box.addButton("Stop Downloading",
|
|
QMessageBox.ButtonRole.DestructiveRole)
|
|
keep = box.addButton("Keep Downloading",
|
|
QMessageBox.ButtonRole.RejectRole)
|
|
box.setDefaultButton(keep)
|
|
box.exec()
|
|
if box.clickedButton() is stop and self._url_worker is not None:
|
|
self._url_worker.cancel()
|
|
|
|
# ---- deleting tracks ----
|
|
|
|
def _delete_tracks(self, track_ids: list[int], delete_files: bool):
|
|
tracks = [self._manager.library.tracks[tid] for tid in track_ids
|
|
if tid in self._manager.library.tracks]
|
|
if not tracks or not self._confirm_delete(tracks, delete_files):
|
|
return
|
|
removed, failures = self._manager.delete_tracks(
|
|
[t.track_id for t in tracks], delete_files=delete_files)
|
|
|
|
if removed:
|
|
where = "moved to Trash" if delete_files else "file kept"
|
|
self.statusBar().showMessage(
|
|
f"Removed {len(removed)} song(s) from your library "
|
|
f"({where}) — undo with Ctrl+Z", 8000)
|
|
if failures:
|
|
QMessageBox.warning(
|
|
self, "Couldn't delete",
|
|
f"{len(failures)} file(s) could not be moved to the Trash, so "
|
|
"those songs are still in your library:\n\n"
|
|
+ "\n".join(f"• “{name}”: {err}" for name, err in failures[:5]))
|
|
|
|
def _confirm_delete(self, tracks: list, delete_files: bool) -> bool:
|
|
if len(tracks) == 1:
|
|
track = tracks[0]
|
|
who = f" by {track.artist}" if track.artist else ""
|
|
question = f"Delete “{track.name}”{who} from your library?"
|
|
else:
|
|
question = f"Delete {len(tracks)} songs from your library?"
|
|
|
|
if delete_files:
|
|
fate = ("The file will be moved to your Trash, so you can put it "
|
|
"back later." if len(tracks) == 1 else
|
|
"The files will be moved to your Trash, so you can put "
|
|
"them back later.")
|
|
else:
|
|
fate = ("The file stays where it is on disk."
|
|
if len(tracks) == 1 else
|
|
"The files stay where they are on disk.")
|
|
|
|
doomed = {t.track_id for t in tracks}
|
|
playlists = {pid for pid, _name in
|
|
[entry for t in tracks
|
|
for entry in self._manager.playlists_containing(t.track_id)]}
|
|
lines = [question, fate]
|
|
if playlists:
|
|
lines.append(f"{'It' if len(doomed) == 1 else 'They'} will also be "
|
|
f"removed from {len(playlists)} playlist(s).")
|
|
|
|
box = QMessageBox(self)
|
|
box.setIcon(QMessageBox.Icon.Warning)
|
|
box.setWindowTitle("Delete from Library")
|
|
box.setText("\n".join(lines))
|
|
delete_button = box.addButton(
|
|
"Delete Song" if len(tracks) == 1 else "Delete Songs",
|
|
QMessageBox.ButtonRole.DestructiveRole)
|
|
cancel = box.addButton("Cancel", QMessageBox.ButtonRole.RejectRole)
|
|
box.setDefaultButton(cancel)
|
|
box.exec()
|
|
return box.clickedButton() is delete_button
|
|
|
|
def _on_tracks_removed(self, track_ids: list):
|
|
"""Tracks left the library (deleted here, or undone/redone). Keep the
|
|
player off dead ids and rebuild the library view."""
|
|
self.player.drop_tracks(set(track_ids))
|
|
self._on_library_reloaded()
|
|
self._update_totals()
|
|
|
|
def _on_tracks_restored(self, track_ids: list):
|
|
"""An undo put tracks back — show them again."""
|
|
self._on_library_reloaded()
|
|
self._update_totals()
|
|
|
|
# ---- album art download ----
|
|
|
|
def _download_album_art(self, track_ids: list[int]):
|
|
"""Fetch cover art from the iTunes Search API for the selection,
|
|
grouped per album, each behind its own confirmation dialog."""
|
|
groups: dict[tuple, dict] = {}
|
|
skipped = 0
|
|
for tid in track_ids:
|
|
track = self._manager.library.tracks.get(tid)
|
|
if track is None:
|
|
continue
|
|
artist = track.album_artist or track.artist
|
|
if not artist or not track.album:
|
|
skipped += 1
|
|
continue
|
|
key = (artist.casefold(), track.album.casefold())
|
|
group = groups.setdefault(
|
|
key, {"artist": artist, "album": track.album, "tracks": []})
|
|
group["tracks"].append(track)
|
|
if skipped:
|
|
self.statusBar().showMessage(
|
|
f"Skipped {skipped} track(s) with no artist/album to search by",
|
|
6000)
|
|
self._art_groups = list(groups.values())
|
|
self._art_next_group()
|
|
|
|
def _art_next_group(self):
|
|
if not self._art_groups:
|
|
return
|
|
group = self._art_groups.pop(0)
|
|
self.statusBar().showMessage(
|
|
f"Searching artwork for {group['artist']} — {group['album']}…")
|
|
# Kept as an attribute so the fetcher (and its thread's signal source)
|
|
# outlives this method.
|
|
self._art_fetcher = AlbumArtFetcher()
|
|
self._art_fetcher.search_finished.connect(
|
|
lambda result, g=group: self._on_art_search(g, result))
|
|
self._art_fetcher.search(group["artist"], group["album"])
|
|
|
|
def _on_art_search(self, group, result):
|
|
if "error" in result:
|
|
self.statusBar().showMessage(
|
|
f"Artwork search failed: {result['error']}", 6000)
|
|
elif not result["candidates"]:
|
|
self.statusBar().showMessage(
|
|
f"No artwork found for {group['artist']} — {group['album']}",
|
|
6000)
|
|
else:
|
|
self.statusBar().clearMessage()
|
|
whole_album = album_tracks(self._manager.library,
|
|
group["artist"], group["album"])
|
|
dialog = AlbumArtDialog(result["candidates"],
|
|
len(group["tracks"]),
|
|
album_count=len(whole_album), parent=self)
|
|
if dialog.exec() and dialog.selected_image:
|
|
targets = whole_album if dialog.apply_to_album \
|
|
else group["tracks"]
|
|
self._embed_album_art(targets, dialog.selected_image,
|
|
dialog.selected_mime)
|
|
self._art_next_group()
|
|
|
|
def _embed_album_art(self, tracks, image: bytes, mime: str):
|
|
"""Embed confirmed artwork into each track's file (not undoable —
|
|
see ``art_ops``)."""
|
|
done, failures = embed_artwork(self._manager, tracks, image, mime)
|
|
if failures:
|
|
QMessageBox.warning(
|
|
self, "Download Album Art",
|
|
"Couldn't embed artwork in some files:\n\n"
|
|
+ "\n".join(failures[:10]))
|
|
if done:
|
|
# Normally track_updated already did this, but only when the
|
|
# file's size changed.
|
|
self._sidebar.art.refresh()
|
|
if done and not failures:
|
|
noun = "track" if done == 1 else "tracks"
|
|
self.statusBar().showMessage(
|
|
f"Artwork embedded in {done} {noun}", 5000)
|
|
|
|
# ---- track identification (fingerprint lookup) ----
|
|
|
|
def _identify_tracks(self, track_ids: list[int]):
|
|
"""Fingerprint the selection and look each track up on AcoustID, one
|
|
at a time, each behind its own proposal dialog."""
|
|
if not fpcalc_available():
|
|
QMessageBox.warning(
|
|
self, "Identify Track",
|
|
"Track identification needs the fpcalc tool from Chromaprint.\n\n"
|
|
"Install it (Fedora: chromaprint-tools, Debian: "
|
|
"libchromaprint-tools) and try again.")
|
|
return
|
|
api_key = self._prefs.acoustid.get("api_key", "").strip()
|
|
if not api_key:
|
|
QMessageBox.information(
|
|
self, "Identify Track",
|
|
"AcoustID needs a free application key.\n\n"
|
|
"Register one at acoustid.org/new-application, then paste it "
|
|
"into Preferences under Track Identification.")
|
|
return
|
|
tracks = []
|
|
skipped = 0
|
|
for tid in track_ids:
|
|
track = self._manager.library.tracks.get(tid)
|
|
if track is None:
|
|
continue
|
|
if not track.location:
|
|
skipped += 1
|
|
continue
|
|
tracks.append(track)
|
|
if skipped:
|
|
self.statusBar().showMessage(
|
|
f"Skipped {skipped} track(s) with no file to fingerprint", 6000)
|
|
if not tracks:
|
|
return
|
|
self._enqueue_identify(tracks)
|
|
|
|
def _identify_setup_problem(self) -> str:
|
|
"""Why an AcoustID lookup can't run on this machine, or ''."""
|
|
if not fpcalc_available():
|
|
return "fpcalc isn't installed"
|
|
if not self._prefs.acoustid.get("api_key", "").strip():
|
|
return "no AcoustID key in Preferences"
|
|
return ""
|
|
|
|
def _enqueue_identify(self, tracks: list):
|
|
"""Queue tracks for Identify and start the queue if it's idle. An
|
|
Import from URL adds songs one at a time while an earlier one is still
|
|
behind its dialog, so the queue can grow mid-run."""
|
|
if not self._identify_running:
|
|
self._identify_queue = []
|
|
self._identify_total = 0
|
|
self._identify_queue.extend(tracks)
|
|
self._identify_total += len(tracks)
|
|
if not self._identify_running:
|
|
self._identify_next()
|
|
|
|
def _identify_next(self):
|
|
if not self._identify_queue:
|
|
self._identify_running = False
|
|
return
|
|
self._identify_running = True
|
|
track = self._identify_queue.pop(0)
|
|
if track.track_id not in self._manager.library.tracks:
|
|
self._identify_next() # deleted while it waited its turn
|
|
return
|
|
done = self._identify_total - len(self._identify_queue)
|
|
counter = (f" ({done} of {self._identify_total})"
|
|
if self._identify_total > 1 else "")
|
|
self.statusBar().showMessage(
|
|
f"Identifying “{track.name}”{counter}…")
|
|
# Kept as an attribute so the identifier (and its thread's signal
|
|
# source) outlives this method.
|
|
self._identifier = TrackIdentifier()
|
|
self._identifier.finished.connect(
|
|
lambda result, t=track: self._on_identified(t, result))
|
|
# Read fresh for every track. With no key (only an Import from URL
|
|
# gets this far without one) the identifier proposes from the
|
|
# filename alone.
|
|
self._identifier.identify(
|
|
track, self._prefs.acoustid.get("api_key", "").strip())
|
|
|
|
def _on_identified(self, track, result: dict):
|
|
if "error" in result:
|
|
self.statusBar().showMessage(
|
|
f"Couldn't identify “{track.name}”: {result['error']}", 6000)
|
|
elif not result["candidates"]:
|
|
self.statusBar().showMessage(
|
|
f"Nothing to propose for “{track.name}” — no database match "
|
|
f"and the filename says nothing", 6000)
|
|
else:
|
|
self.statusBar().clearMessage()
|
|
dialog = IdentifyDialog(track, result["candidates"],
|
|
remaining=len(self._identify_queue),
|
|
parent=self)
|
|
accepted = dialog.exec()
|
|
if accepted:
|
|
fields = dialog.result_fields()
|
|
if fields:
|
|
# The edit funnel writes the tags, may relocate the file
|
|
# and records one undo step — see LibraryManager.
|
|
self._manager.edit_track_fields(track.track_id, fields)
|
|
self.statusBar().showMessage(
|
|
f"Updated tags for “{track.name}”", 5000)
|
|
elif dialog.cancel_all:
|
|
self._identify_queue.clear()
|
|
self._identify_running = False
|
|
# Also covers songs a URL import has yet to download: they
|
|
# still import, they just don't get a dialog each.
|
|
self._url_identify = False
|
|
self.statusBar().clearMessage()
|
|
return
|
|
self._identify_next()
|
|
|
|
# ---- relocating a moved/renamed file ----
|
|
|
|
def _existing_ancestor(self, path: str) -> str:
|
|
"""Deepest existing directory at or above `path`'s parent, so the file
|
|
picker opens somewhere real even if the exact folder is gone."""
|
|
if path:
|
|
p = Path(path).parent
|
|
while not p.exists() and p != p.parent:
|
|
p = p.parent
|
|
if p.exists():
|
|
return str(p)
|
|
music = self._music_import_dir()
|
|
return str(music) if music and music.exists() else str(Path.home())
|
|
|
|
def _on_track_missing(self, track):
|
|
"""A track's file is gone (moved/renamed outside lintunes). Offer to
|
|
relocate it manually; on success, repoint the track and resume."""
|
|
box = QMessageBox(self)
|
|
box.setIcon(QMessageBox.Icon.Warning)
|
|
box.setWindowTitle("Can't play track")
|
|
box.setText(f"Can't play “{track.name}”.")
|
|
box.setInformativeText(
|
|
f"File not found:\n{track.location or '(no location)'}")
|
|
locate_btn = box.addButton("Locate File…", QMessageBox.ButtonRole.ActionRole)
|
|
box.addButton(QMessageBox.StandardButton.Cancel)
|
|
box.exec()
|
|
if box.clickedButton() is not locate_btn:
|
|
return
|
|
chosen, _filter = QFileDialog.getOpenFileName(
|
|
self, f"Locate “{track.name}”", self._existing_ancestor(track.location),
|
|
"Audio files (*.mp3 *.m4a *.flac *.aac *.wav *.aiff *.aif);;All files (*)")
|
|
if not chosen:
|
|
return
|
|
self._manager.set_track_location(track.track_id, chosen)
|
|
# Defer the retry so it runs after this modal handler unwinds.
|
|
QTimer.singleShot(0, self.player.retry_current)
|
|
|
|
# ---- window-level drops (import anywhere) ----
|
|
|
|
def dragEnterEvent(self, event):
|
|
if event.mimeData().hasUrls():
|
|
event.acceptProposedAction()
|
|
|
|
def dropEvent(self, event):
|
|
paths = [Path(u.toLocalFile()) for u in event.mimeData().urls()
|
|
if u.isLocalFile()]
|
|
if paths:
|
|
self.import_files(paths)
|
|
event.acceptProposedAction()
|
|
|
|
# ---- playback ----
|
|
|
|
def play_pause(self):
|
|
"""Play/pause like iTunes: with nothing cued, start the visible
|
|
view's tracks (from the selected row, else the top)."""
|
|
if self.player.current_track is not None:
|
|
self.player.toggle_play()
|
|
return
|
|
view = self._content.currentWidget()
|
|
ids = view.table.view_order_track_ids()
|
|
if not ids:
|
|
return
|
|
selected = view.table.selectionModel().selectedRows()
|
|
start = selected[0].row() if selected else 0
|
|
self._play_from_view(view, ids, start)
|
|
|
|
def _play_from_view(self, view, ids, start):
|
|
"""Start playback, recording which view it came from so the now-playing
|
|
icon (and Go to Current Song) can find the right context."""
|
|
self._now_playing_context = view.table.context_id()
|
|
self.player.play_queue(ids, start)
|
|
|
|
def _on_track_changed(self, track):
|
|
if track is None:
|
|
self.setWindowTitle("LinTunes")
|
|
else:
|
|
self.setWindowTitle(f"{track.name} — {track.artist} · LinTunes")
|
|
self._update_now_playing_icon()
|
|
|
|
def _on_track_meta_updated(self, track_id: int):
|
|
current = self.player.current_track
|
|
if current is not None and current.track_id == track_id:
|
|
self._sidebar.art.refresh()
|
|
|
|
def _update_now_playing_icon(self, _playing=None):
|
|
track = self.player.current_track
|
|
track_id = track.track_id if track else None
|
|
playing = self.player.is_playing()
|
|
# Only the table showing the context playback started from gets the icon,
|
|
# so the same song appearing in another playlist/library isn't flagged.
|
|
for view in (self._library_view, self._playlist_view):
|
|
match = (track_id is not None
|
|
and view.table.context_id() == self._now_playing_context)
|
|
view.table.model_.set_now_playing(track_id if match else None,
|
|
playing)
|
|
|
|
# ---- global media keys (spacebar / arrows), per spec detail 8 ----
|
|
|
|
def eventFilter(self, obj, event):
|
|
if event.type() == QEvent.Type.KeyPress and self.isActiveWindow():
|
|
focus = QApplication.focusWidget()
|
|
if isinstance(focus, (QLineEdit, QPlainTextEdit, QTextEdit,
|
|
QAbstractSpinBox, QComboBox)):
|
|
return False
|
|
key = event.key()
|
|
if key in (Qt.Key.Key_Space, Qt.Key.Key_Right, Qt.Key.Key_Left) \
|
|
and event.isAutoRepeat():
|
|
# On Wayland key repeat is generated client-side until the
|
|
# compositor delivers the release — and a compositor busy
|
|
# with our own MPRIS churn delivers it late, so one held
|
|
# arrow became 1,676 track skips (Round 56). A transport key
|
|
# acts once per press; the repeats are swallowed so they
|
|
# don't fall through and move the table's cursor either.
|
|
return True
|
|
if key == Qt.Key.Key_Space:
|
|
self._log_key(event, "space")
|
|
self.play_pause()
|
|
return True
|
|
if key == Qt.Key.Key_Right:
|
|
self._log_key(event, "right")
|
|
self.player.next()
|
|
return True
|
|
if key == Qt.Key.Key_Left:
|
|
self._log_key(event, "left")
|
|
self.player.previous()
|
|
return True
|
|
return super().eventFilter(obj, event)
|
|
|
|
@staticmethod
|
|
def _log_key(event, name: str):
|
|
"""Provenance for keys the filter acts on: which input device sent
|
|
the press (compositor permitting) — see eventlog module docstring."""
|
|
detail = ""
|
|
try:
|
|
device = event.device()
|
|
if device is not None:
|
|
detail = f"device={device.name()!r}"
|
|
except Exception:
|
|
pass
|
|
log_control("keyfilter", name, detail)
|
|
|
|
def _on_playing_changed(self, playing: bool):
|
|
if playing:
|
|
self._inhibitor.inhibit()
|
|
else:
|
|
self._inhibitor.release()
|
|
|
|
def closeEvent(self, event):
|
|
# Unsaved cassette choices in friend mode: Save / Don't Save / Cancel.
|
|
if not self._cassette.confirm_quit():
|
|
event.ignore()
|
|
return
|
|
# Quitting mid-transfer abandons a partial file on the device (it
|
|
# self-heals next sync, but silently) — make it a deliberate choice.
|
|
# A link import has nothing half-written on a device to warn about:
|
|
# it is simply stopped below.
|
|
busy = self._busy_worker()
|
|
if busy is not None and busy not in (self._export_worker,
|
|
self._url_worker):
|
|
box = QMessageBox(self)
|
|
box.setWindowTitle("Sync in progress")
|
|
box.setText(
|
|
f"Still copying to the {self._sync_device_name}.\n"
|
|
"If you quit now, the unfinished track will be re-copied "
|
|
"on the next sync.")
|
|
quit_button = box.addButton(
|
|
"Quit Anyway", QMessageBox.ButtonRole.DestructiveRole)
|
|
keep = box.addButton(
|
|
"Keep Syncing", QMessageBox.ButtonRole.RejectRole)
|
|
box.setDefaultButton(keep)
|
|
box.exec()
|
|
if box.clickedButton() is not quit_button:
|
|
event.ignore()
|
|
return
|
|
if self._url_worker is not None:
|
|
self._url_worker.cancel()
|
|
self._url_worker.cleanup()
|
|
self._inhibitor.release()
|
|
self._sync_inhibitor.release()
|
|
# Before the player: this stops the device and kills the HTTP server it
|
|
# is fetching from. Skipped, the Chromecast sits buffering a dead URL.
|
|
self._cassette.shutdown()
|
|
self._cast.shutdown()
|
|
self.player.shutdown()
|
|
self._manager.flush()
|
|
super().closeEvent(event)
|