diff --git a/CLAUDE.md b/CLAUDE.md index 7e1192b..5b0b6ce 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -378,6 +378,25 @@ persistence) → GUI (Qt widgets that read the manager and connect to its signal track number into the mp3 **before** emitting `downloaded`, so the import files it under the right artist and Identify ranks by that artist. If the embed page changes shape, `parse_embed` raises `SpotifyError`. + **Several songs are identified as one batch** (Round 77, + `identify_batch.py` + `gui/identify_batch_dialog.py`): an import of more + than one song never pops a dialog mid-download. `IdentifyCollector` looks + each up quietly as it lands, and only after `finished`/`failed` *and* the + last lookup does one review window open, with all the songs as rows in the + app's columns. `find_common_album` makes it "one album" when every song + naming an album can be on the same one (its own tags vote too, which is + what carries a Spotify album through AcoustID's compilations). Then the + album-wide fields are unified to the most common value, shown once above + the table, and the cover comes from the same `AlbumArtFetcher` search. + Ticked rows go through `LibraryManager.edit_many_track_fields` as **one** + undo step. Unticked rows get the ordinary `IdentifyDialog` afterwards, + from the lookups already made. **No popup may land on top of another + app**: `MainWindow._when_app_active` shows it at once only while LinTunes + is the active application, and otherwise asks for attention + (`QApplication.alert`) and waits for `applicationStateChanged`. A window + test that downloads several songs must stub `_show_batch_review` and + cancel `_url_batch` on teardown, or a late lookup opens a modal dialog + in the next test. - **`lintunes/mpris.py`** — registers `org.mpris.MediaPlayer2.lintunes` over D-Bus so the desktop's media keys / now-playing popup control playback. Spacebar and diff --git a/lintunes/__init__.py b/lintunes/__init__.py index 0a27803..732ebae 100644 --- a/lintunes/__init__.py +++ b/lintunes/__init__.py @@ -1,3 +1,3 @@ """LinTunes — iTunes-style music library manager and player for Linux.""" -__version__ = "0.37.0" +__version__ = "0.38.0" diff --git a/lintunes/gui/identify_batch_dialog.py b/lintunes/gui/identify_batch_dialog.py new file mode 100644 index 0000000..810fa19 --- /dev/null +++ b/lintunes/gui/identify_batch_dialog.py @@ -0,0 +1,384 @@ +"""Review a whole import's proposed tags at once (Round 77). + +The rows are the songs, the columns the app's own (Name, Artist, Album …), +and every cell is editable. A value that would change is **bold**, and its +tooltip says what it was. Ticked rows are applied together, as one undo +step. Unticked rows come back afterwards one at a time in the ordinary +Identify dialog, and **Review Each Song** sends all of them that way. + +When ``identify_batch.analyze`` decided the songs are one album, the +album-wide fields sit above the table, once. Editing them edits every row. +Beside them is the album's cover from the same search Download Album Art +uses, with Next Artwork to cycle through the results. + +Passive like IdentifyDialog: nothing is written until the caller reads +``edits()``, ``unchecked_rows()`` and ``selected_image`` after an accept. +""" +import re + +from PyQt6.QtCore import Qt +from PyQt6.QtGui import QFont, QPixmap +from PyQt6.QtWidgets import ( + QAbstractItemView, QCheckBox, QDialog, QDialogButtonBox, QFormLayout, + QHBoxLayout, QHeaderView, QLabel, QLineEdit, QPushButton, QSpinBox, + QTableWidget, QTableWidgetItem, QVBoxLayout, +) + +from lintunes.art_search import AlbumArtFetcher + +ART_PX = 180 + +# (header, fields shown in the column) +COLUMNS = [ + ("Name", ("name",)), + ("Artist", ("artist",)), + ("Album Artist", ("album_artist",)), + ("Album", ("album",)), + ("Year", ("year",)), + ("Track", ("track_number", "track_count")), + ("Disc", ("disc_number",)), +] +CHECK_COL = 0 +FIRST_FIELD_COL = 1 +SOURCE_COL = FIRST_FIELD_COL + len(COLUMNS) + +_TRACK = re.compile(r"^\s*(\d+)?\s*(?:(?:of|/)\s*(\d+))?\s*$") + + +def format_cell(fields: tuple, values: dict) -> str: + if fields == ("track_number", "track_count"): + number, count = values.get("track_number"), values.get("track_count") + if number and count: + return f"{number} of {count}" + if number: + return str(number) + return f"of {count}" if count else "" + value = values.get(fields[0]) + return "" if value in (None, 0) else str(value) + + +def parse_cell(fields: tuple, text: str) -> dict: + """A cell's text back into fields (None where it's blank or unreadable, + which means "leave it as it is").""" + text = (text or "").strip() + if fields == ("track_number", "track_count"): + m = _TRACK.match(text) + if not m: + return {"track_number": None, "track_count": None} + return {"track_number": int(m.group(1)) if m.group(1) else None, + "track_count": int(m.group(2)) if m.group(2) else None} + name = fields[0] + if name in ("year", "disc_number"): + return {name: int(text) if text.isdigit() and int(text) else None} + return {name: text or None} + + +class IdentifyBatchDialog(QDialog): + APPLY, REVIEW_ALL = 1, 2 + + def __init__(self, analysis, art_query=None, parent=None): + super().__init__(parent) + self.setWindowTitle("Identify Songs") + self._analysis = analysis + self._rows = list(analysis.rows) + if analysis.is_album: + self._rows.sort(key=lambda r: (r.proposal.get("disc_number") or 0, + r.proposal.get("track_number") + or 999)) + self.outcome = 0 + self.selected_image: bytes | None = None + self.selected_mime = "image/jpeg" + self._art_candidates = [] + self._art_index = 0 + self._art_image = None + self._art_mime = "image/jpeg" + self._fetcher = None + self._filling = False + + layout = QVBoxLayout(self) + n = len(self._rows) + header = QLabel( + f"These {n} songs look like one album." if analysis.is_album + else f"Proposed tags for {n} songs.") + bold = header.font() + bold.setBold(True) + header.setFont(bold) + layout.addWidget(header) + + self._album_edit = self._album_artist_edit = self._year_edit = None + self._embed = None + if analysis.is_album: + layout.addLayout(self._album_panel(art_query)) + + self._table = QTableWidget(n, SOURCE_COL + 1) + self._table.setHorizontalHeaderLabels( + [""] + [h for h, _f in COLUMNS] + ["Found by"]) + self._table.verticalHeader().hide() + self._table.setSelectionBehavior( + QAbstractItemView.SelectionBehavior.SelectRows) + self._table.setEditTriggers( + QAbstractItemView.EditTrigger.DoubleClicked + | QAbstractItemView.EditTrigger.EditKeyPressed + | QAbstractItemView.EditTrigger.AnyKeyPressed) + head = self._table.horizontalHeader() + head.setSectionResizeMode(QHeaderView.ResizeMode.Interactive) + head.setSectionResizeMode(CHECK_COL, + QHeaderView.ResizeMode.ResizeToContents) + head.setStretchLastSection(True) + self._fill_table() + self._table.itemChanged.connect(self._on_item_changed) + for col, width in zip(range(FIRST_FIELD_COL, SOURCE_COL), + (200, 150, 120, 170, 55, 70, 45)): + self._table.setColumnWidth(col, width) + layout.addWidget(self._table, 1) + + hint = QLabel("Bold is what changes (hover to see what it was). " + "Double-click to edit. Unticked songs are shown one " + "at a time afterwards.") + hint.setEnabled(False) + hint.setWordWrap(True) + layout.addWidget(hint) + + buttons = QDialogButtonBox() + self._apply_btn = QPushButton() + self._apply_btn.setDefault(True) + buttons.addButton(self._apply_btn, + QDialogButtonBox.ButtonRole.AcceptRole) + review = QPushButton("Review Each Song…") + review.setToolTip("Skip this list and confirm every song in its own " + "window") + buttons.addButton(review, QDialogButtonBox.ButtonRole.ActionRole) + cancel = buttons.addButton(QDialogButtonBox.StandardButton.Cancel) + cancel.setToolTip("Stop identifying: the songs keep the tags they " + "have") + self._apply_btn.clicked.connect(self._on_apply) + review.clicked.connect(self._on_review_all) + buttons.rejected.connect(self.reject) + layout.addWidget(buttons) + self._update_apply() + self._table.setFocus() # not the Album box: nothing to type yet + self.resize(1000, min(260 + 30 * n + (ART_PX if analysis.is_album + else 0), 820)) + + # ---- the album panel ---- + + def _album_panel(self, art_query): + a = self._analysis + row = QHBoxLayout() + self._art = QLabel("Searching…" if art_query else "No artwork search") + self._art.setFixedSize(ART_PX, ART_PX) + self._art.setAlignment(Qt.AlignmentFlag.AlignCenter) + self._art.setStyleSheet("border: 1px solid palette(mid);") + self._art.setWordWrap(True) + row.addWidget(self._art) + + side = QVBoxLayout() + form = QFormLayout() + self._album_edit = QLineEdit(a.album or "") + self._album_artist_edit = QLineEdit(a.album_artist or "") + self._album_artist_edit.setPlaceholderText("(none)") + self._year_edit = QSpinBox() + self._year_edit.setRange(0, 9999) + self._year_edit.setSpecialValueText(" ") + self._year_edit.setValue(a.year or 0) + form.addRow("Album", self._album_edit) + form.addRow("Album Artist", self._album_artist_edit) + form.addRow("Year", self._year_edit) + self._album_edit.textEdited.connect( + lambda t: self._set_column(("album",), t)) + self._album_artist_edit.textEdited.connect( + lambda t: self._set_column(("album_artist",), t)) + self._year_edit.valueChanged.connect( + lambda v: self._set_column(("year",), str(v) if v else "")) + side.addLayout(form) + + self._art_caption = QLabel() + self._art_caption.setWordWrap(True) + self._art_caption.setEnabled(False) + side.addWidget(self._art_caption) + art_row = QHBoxLayout() + self._embed = QCheckBox(f"Embed this artwork in all " + f"{len(self._rows)} songs") + self._embed.setEnabled(False) + self._next_art = QPushButton("Next Artwork") + self._next_art.setEnabled(False) + self._next_art.clicked.connect(self._on_next_art) + art_row.addWidget(self._embed) + art_row.addWidget(self._next_art) + art_row.addStretch(1) + side.addLayout(art_row) + side.addStretch(1) + row.addLayout(side, 1) + + if art_query: + self._fetcher = AlbumArtFetcher() + self._fetcher.search_finished.connect(self._on_art_search) + self._fetcher.image_finished.connect(self._on_art_image) + self._fetcher.search(*art_query) + return row + + def _on_art_search(self, result: dict): + self._art_candidates = result.get("candidates") or [] + if not self._art_candidates: + self._art.setText("No artwork found" if "error" not in result + else "Artwork search failed") + self._art_caption.setText(result.get("error", "")) + return + self._next_art.setEnabled(len(self._art_candidates) > 1) + self._load_art() + + def _load_art(self): + candidate = self._art_candidates[self._art_index] + n = len(self._art_candidates) + self._art_caption.setText( + f"{candidate.artist} — {candidate.album}, via {candidate.source}" + + (f" ({self._art_index + 1}/{n})" if n > 1 else "")) + self._art_image = None + self._embed.setEnabled(False) + self._art.setPixmap(QPixmap()) + self._art.setText("Loading…") + self._fetcher.fetch(candidate) + + def _on_next_art(self): + self._art_index = (self._art_index + 1) % len(self._art_candidates) + self._load_art() + + def _on_art_image(self, result: dict): + if not self._art_candidates or result.get("candidate") is not \ + self._art_candidates[self._art_index]: + if "candidate" in result: + return # a stale download after Next Artwork + if "error" in result: + self._art.setText("Couldn't load the image") + return + pixmap = QPixmap() + if not pixmap.loadFromData(result["image"]): + self._art.setText("Couldn't decode the image") + return + self._art_image = result["image"] + self._art_mime = result["mime"] + self._art.setPixmap(pixmap.scaled( + self._art.size(), Qt.AspectRatioMode.KeepAspectRatio, + Qt.TransformationMode.SmoothTransformation)) + first = not self._embed.isEnabled() and self._art_index == 0 + self._embed.setEnabled(True) + if first: + self._embed.setChecked(True) + + # ---- the table ---- + + def _fill_table(self): + self._filling = True + for r, row in enumerate(self._rows): + check = QTableWidgetItem() + check.setFlags(Qt.ItemFlag.ItemIsUserCheckable + | Qt.ItemFlag.ItemIsEnabled + | Qt.ItemFlag.ItemIsSelectable) + # Ticked whether or not it changes: unticking is what asks for + # a song's own window, and a song with nothing to change + # shouldn't get one unasked. + check.setCheckState(Qt.CheckState.Checked) + self._table.setItem(r, CHECK_COL, check) + shown = {f: (row.proposal.get(f) if row.proposal.get(f) + not in (None, "", 0) else row.current(f)) + for f in ("name", "artist", "album_artist", "album", + "year", "track_number", "track_count", + "disc_number")} + for c, (_h, fields) in enumerate(COLUMNS, FIRST_FIELD_COL): + self._table.setItem(r, c, + QTableWidgetItem(format_cell(fields, + shown))) + self._mark(r, c) + source = QTableWidgetItem(row.source) + source.setFlags(Qt.ItemFlag.ItemIsEnabled + | Qt.ItemFlag.ItemIsSelectable) + self._table.setItem(r, SOURCE_COL, source) + self._filling = False + + def _mark(self, r: int, c: int): + """Bold when the cell would change the song; the tooltip says what + it was.""" + fields = COLUMNS[c - FIRST_FIELD_COL][1] + item = self._table.item(r, c) + row = self._rows[r] + current = {f: row.current(f) for f in fields} + was = format_cell(fields, current) + proposed = parse_cell(fields, item.text()) + differs = any(v is not None and v != (current[f] or None) + for f, v in proposed.items()) + font = QFont(item.font()) + font.setBold(differs) + item.setFont(font) + item.setToolTip(f"Was: {was or '(blank)'}" if differs else "") + + def _set_column(self, fields: tuple, text: str): + col = FIRST_FIELD_COL + [f for _h, f in COLUMNS].index(fields) + self._table.blockSignals(True) + for r in range(len(self._rows)): + self._table.item(r, col).setText(text) + self._mark(r, col) + self._table.blockSignals(False) + + def _on_item_changed(self, item): + if self._filling: + return + if item.column() == CHECK_COL: + self._update_apply() + elif FIRST_FIELD_COL <= item.column() < SOURCE_COL: + self._table.blockSignals(True) + self._mark(item.row(), item.column()) + self._table.blockSignals(False) + + def _checked(self, r: int) -> bool: + return (self._table.item(r, CHECK_COL).checkState() + == Qt.CheckState.Checked) + + def _update_apply(self): + n = sum(1 for r in range(len(self._rows)) if self._checked(r)) + self._apply_btn.setText(f"Apply to {n} Song{'' if n == 1 else 's'}" + if n != len(self._rows) else "Apply to All") + self._apply_btn.setEnabled(True) # none ticked = review them all + + # ---- what the caller reads ---- + + def row_fields(self, r: int) -> dict: + """Row ``r``'s changes as the cells now read: blank or unchanged + cells propose nothing.""" + row = self._rows[r] + out = {} + for c, (_h, fields) in enumerate(COLUMNS, FIRST_FIELD_COL): + for name, value in parse_cell( + fields, self._table.item(r, c).text()).items(): + if value is not None and value != (row.current(name) + or None): + out[name] = value + return out + + def edits(self) -> dict: + """{track id: fields} for the ticked songs that change.""" + out = {} + for r, row in enumerate(self._rows): + if self._checked(r): + fields = self.row_fields(r) + if fields: + out[row.track.track_id] = fields + return out + + def unchecked_rows(self) -> list: + return [row for r, row in enumerate(self._rows) + if not self._checked(r)] + + def all_rows(self) -> list: + return list(self._rows) + + def _on_apply(self): + self.outcome = self.APPLY + if self._embed is not None and self._embed.isChecked() \ + and self._art_image: + self.selected_image = self._art_image + self.selected_mime = self._art_mime + self.accept() + + def _on_review_all(self): + self.outcome = self.REVIEW_ALL + self.accept() diff --git a/lintunes/gui/main_window.py b/lintunes/gui/main_window.py index 2f98348..38df4ff 100644 --- a/lintunes/gui/main_window.py +++ b/lintunes/gui/main_window.py @@ -33,7 +33,11 @@ from lintunes.gui.device_sync_dialog import ( DeviceSyncSettingsDialog, resolve_sync_playlists, ) from lintunes.gui.export_dialog import ExportKindDialog, WebMixDialog +from lintunes.gui.identify_batch_dialog import IdentifyBatchDialog from lintunes.gui.identify_dialog import IdentifyDialog +from lintunes.identify_batch import ( + IdentifyCollector, album_art_query, analyze, +) from lintunes.gui.sidebar import SIDEBAR_INSET, SidebarPanel from lintunes.gui.library_view import LibraryView from lintunes.gui.playlist_view import PlaylistView @@ -250,6 +254,13 @@ class MainWindow(QMainWindow): self._url_identify = True # False after "Stop Identifying" self._url_note = "" self._url_progress = None # UrlImportProgress, for several songs + # Several songs are looked up quietly while they download, and + # reviewed together once the last one has landed. + self._url_batch = None # identify_batch.IdentifyCollector + # Popups waiting for LinTunes to be the app in front (they must + # never land on top of another app). + self._popups_waiting: list = [] + self._popups_hooked = False # 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. @@ -1341,6 +1352,15 @@ class MainWindow(QMainWindow): if problem else "") spotify = dialog.spotify_tracks() + if self._url_batch is not None: + self._url_batch.cancel() + self._url_batch = None + if max(len(spotify), len(chosen)) > 1: + batch = IdentifyCollector( + self._prefs.acoustid.get("api_key", "").strip(), self) + batch.progress.connect(self._on_url_batch_progress) + batch.ready.connect(self._on_url_batch_ready) + self._url_batch = batch if spotify: # Spotify won't hand over audio: each song is found on YouTube # and arrives already wearing Spotify's tags. @@ -1421,12 +1441,113 @@ class MainWindow(QMainWindow): if self._url_progress is not None: self._url_progress.landed(bool(imported)) if imported and self._url_identify: - self._enqueue_identify(imported) + if self._url_batch is not None: + self._url_batch.add(imported) # reviewed when all are in + else: + self._enqueue_identify(imported) def _finish_url_import(self): if self._url_worker is not None: self._url_worker.cleanup() self._hide_sync_widgets() + if self._url_batch is not None: + self._url_batch.close() # no more songs are coming + + def _on_url_batch_progress(self, done: int, total: int): + worker = self._url_worker + if worker is None or not worker.busy(): + self.statusBar().showMessage( + f"Identifying the imported songs: {done} of {total}…") + + def _on_url_batch_ready(self, entries: list): + """Every song is downloaded and looked up: one window for them all + (one song, if that's all that landed, gets the ordinary one).""" + if self.sender() is not self._url_batch: + return + self._url_batch = None + library = self._manager.library + entries = [(t, c) for t, c in entries if t.track_id in library.tracks] + if not entries or not self._url_identify: + return + self.statusBar().clearMessage() + if len(entries) == 1: + analysis = analyze(entries) + self._when_app_active( + lambda: self._review_individually(analysis.rows)) + return + analysis = analyze(entries) + self._when_app_active(lambda: self._show_batch_review(analysis)) + + def _show_batch_review(self, analysis): + dialog = IdentifyBatchDialog(analysis, album_art_query(analysis), + self) + if not dialog.exec(): + self.statusBar().showMessage( + "Identify cancelled — the songs keep the tags they have", + 6000) + return + if dialog.outcome == dialog.REVIEW_ALL: + self._review_individually(dialog.all_rows()) + return + edits = dialog.edits() + if edits: + # The edit funnel writes the tags, may relocate the files, and + # records the whole list as one undo step. + self._manager.edit_many_track_fields(edits) + if dialog.selected_image: + library = self._manager.library + tracks = [library.tracks[r.track.track_id] + for r in dialog.all_rows() + if r.track.track_id in library.tracks] + self._embed_album_art(tracks, dialog.selected_image, + dialog.selected_mime) + if edits: + self.statusBar().showMessage( + f"Updated tags for {len(edits)} song(s)", 6000) + self._review_individually(dialog.unchecked_rows()) + + def _review_individually(self, rows: list): + """The ordinary Identify window, one song at a time, from lookups + already made.""" + rows = [r for r in rows if r.candidates + and r.track.track_id in self._manager.library.tracks] + for i, row in enumerate(rows): + track = self._manager.library.tracks[row.track.track_id] + dialog = IdentifyDialog(track, row.candidates, + remaining=len(rows) - i - 1, parent=self) + if dialog.exec(): + fields = dialog.result_fields() + if fields: + self._manager.edit_track_fields(track.track_id, fields) + elif dialog.cancel_all: + return + + def _when_app_active(self, show): + """Run ``show`` (which opens a window over this one) now if LinTunes + is the app in front, else once it is. A dialog parented to the main + window only ever stacks over LinTunes, but one opened while another + app has focus would still jump in front of it. Meanwhile the window + asks for attention instead.""" + app = QApplication.instance() + if app.applicationState() == Qt.ApplicationState.ApplicationActive: + show() + return + self._popups_waiting.append(show) + QApplication.alert(self) + if not self._popups_hooked: + self._popups_hooked = True + app.applicationStateChanged.connect(self._on_app_state_changed) + + def _on_app_state_changed(self, state): + if (state == Qt.ApplicationState.ApplicationActive + and self._popups_waiting): + waiting, self._popups_waiting = self._popups_waiting, [] + + def run(): + for show in waiting: + show() + # After the activation event has finished being delivered. + QTimer.singleShot(0, run) def _on_url_finished(self, summary: dict): self._finish_url_import() @@ -1453,8 +1574,9 @@ class MainWindow(QMainWindow): self._finish_url_import() if self._url_progress is not None: self._url_progress.finish(False) - QMessageBox.warning(self, "Import from URL", - f"Couldn't import from that link:\n\n{message}") + self._when_app_active(lambda: 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) @@ -1778,7 +1900,14 @@ class MainWindow(QMainWindow): f"and the filename says nothing", 6000) else: self.statusBar().clearMessage() - dialog = IdentifyDialog(track, result["candidates"], + self._when_app_active( + lambda: self._show_identify(track, result["candidates"])) + return + self._identify_next() + + def _show_identify(self, track, candidates): + if track.track_id in self._manager.library.tracks: + dialog = IdentifyDialog(track, candidates, remaining=len(self._identify_queue), parent=self) accepted = dialog.exec() @@ -1796,6 +1925,9 @@ class MainWindow(QMainWindow): # 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 + if self._url_batch is not None: + self._url_batch.cancel() + self._url_batch = None self.statusBar().clearMessage() return self._identify_next() diff --git a/lintunes/identify_batch.py b/lintunes/identify_batch.py new file mode 100644 index 0000000..0e61b2e --- /dev/null +++ b/lintunes/identify_batch.py @@ -0,0 +1,250 @@ +"""Identify several songs at once (Round 77): the analysis behind the batch +review window an Import from URL of more than one song ends with. + +One dialog per song was fine for a song and a chore for an album: twelve +"Apply Tags" clicks for twelve songs that obviously belong together. So a +multi-song import is fingerprinted quietly while it downloads, and only +once *everything* has landed is it shown, all at once, as rows. + +``find_common_album`` is the "these are all one album" call. Every song +votes for each album it could be on: any of its lookup candidates, plus +its own tags (a Spotify album import already carries the right album, even +when AcoustID only knows the song from a compilation). An album named by +every song that names one, covering at least ``ALBUM_SHARE`` of them, is +the album. Each song then takes its best candidate *on that album*, and the +album-wide fields (album, album artist, year, track count) are unified to +the most common value, so one stray pressing can't leave a single song +filed under a different year. A playlist gets none of that: each song +simply shows its best candidate. + +Pure and Qt-widget-free apart from ``IdentifyCollector``, which runs the +lookups one at a time while a download is still going. +""" +from collections import Counter +from dataclasses import dataclass, field + +from PyQt6.QtCore import QObject, pyqtSignal + +from lintunes.art_search import _simplify_album +from lintunes.fingerprint import IdentifyCandidate, TrackIdentifier + +# What a row shows and may change, in column order. +FIELDS = ("name", "artist", "album_artist", "album", "year", "track_number", + "track_count", "disc_number") +ALBUM_FIELDS = ("album", "album_artist", "year", "track_count") + +# Share of the songs an album must cover to be "one album". +ALBUM_SHARE = 0.75 + + +@dataclass +class BatchRow: + track: object # the library Track + proposal: dict # field → proposed value (None: keep) + source: str # "AcoustID 97%", "Filename", … + candidates: list = field(default_factory=list) + + def current(self, name: str): + return getattr(self.track, name, None) + + def changes(self) -> dict: + """The proposed values that differ from the track's own. A blank + proposal never blanks a tag.""" + out = {} + for name in FIELDS: + value = self.proposal.get(name) + if value in (None, "", 0): + continue + if value != (self.current(name) or None): + out[name] = value + return out + + +@dataclass +class BatchAnalysis: + rows: list + album: str | None = None # set when the songs are one album + album_artist: str | None = None + year: int | None = None + + @property + def is_album(self) -> bool: + return self.album is not None + + +def album_key(album: str | None) -> str: + """"Last Splash (Remastered)" and "last splash" are one album.""" + return _simplify_album(album or "").casefold().strip() + + +def tags_candidate(track) -> IdentifyCandidate: + """What the song's own tags say, as a candidate.""" + candidate = IdentifyCandidate(score=0.0, source="tags") + for name in FIELDS: + value = getattr(track, name, None) + if value not in (None, "", 0): + setattr(candidate, name, value) + return candidate + + +def describe(candidate: IdentifyCandidate | None) -> str: + if candidate is None: + return "Nothing found" + if candidate.source == "acoustid": + return f"AcoustID {round(candidate.score * 100)}%" + return {"filename": "Filename", "tags": "Its own tags"}.get( + candidate.source, candidate.source) + + +def _options(track, candidates) -> list: + """Every candidate for a song, best first, its own tags last.""" + return list(candidates) + [tags_candidate(track)] + + +def find_common_album(entries) -> str | None: + """The album key every song that names an album agrees on, or None. + + ``entries`` is [(track, candidates)]. Among albums every voting song + can be on, the one its songs' *best* candidates name most wins (so a + compilation that happens to hold every song too loses to the album + the top matches say).""" + if len(entries) < 2: + return None + voters = [] + top_votes = Counter() + for track, candidates in entries: + keys = {album_key(c.album) for c in _options(track, candidates) + if album_key(c.album)} + if keys: + voters.append(keys) + for c in _options(track, candidates): + if album_key(c.album): + top_votes[album_key(c.album)] += 1 + break + if len(voters) < max(2, ALBUM_SHARE * len(entries)): + return None + common = set.intersection(*voters) + if not common: + return None + return max(sorted(common), key=lambda k: top_votes[k]) + + +def _most_common(values): + """The most frequent non-empty value; ties go to the smallest (the + earliest year, the lowest count).""" + counts = Counter(v for v in values if v not in (None, "", 0)) + if not counts: + return None + best = max(counts.values()) + return min(v for v, n in counts.items() if n == best) + + +def analyze(entries) -> BatchAnalysis: + """[(track, candidates)] → rows to review, unified into one album when + the songs are one.""" + key = find_common_album(entries) + chosen = [] + for track, candidates in entries: + pick = candidates[0] if candidates else None + if key is not None: + pick = next((c for c in _options(track, candidates) + if album_key(c.album) == key), pick) + chosen.append((track, candidates, pick)) + + if key is None: + rows = [BatchRow(track, pick.fields() if pick else {}, + describe(pick), list(candidates)) + for track, candidates, pick in chosen] + return BatchAnalysis(rows) + + picks = [pick for _t, _c, pick in chosen if pick is not None + and album_key(pick.album) == key] + album = _most_common(p.album for p in picks) + album_artist = _most_common(p.album_artist for p in picks) + year = _most_common(p.year for p in picks) + track_count = _most_common(p.track_count for p in picks) + rows = [] + for track, candidates, pick in chosen: + proposal = pick.fields() if pick is not None else {} + proposal.update({"album": album, "album_artist": album_artist, + "year": year, "track_count": track_count}) + rows.append(BatchRow(track, proposal, describe(pick), + list(candidates))) + return BatchAnalysis(rows, album=album, album_artist=album_artist, + year=year) + + +def album_art_query(analysis: BatchAnalysis) -> tuple[str, str] | None: + """(artist, album) to search cover art by, for an album batch.""" + if not analysis.is_album: + return None + artist = analysis.album_artist or _most_common( + r.proposal.get("artist") or r.current("artist") + for r in analysis.rows) + return (artist, analysis.album) if artist else None + + +class IdentifyCollector(QObject): + """Looks songs up one at a time, quietly, as they're handed in, and + says ``ready`` with [(track, candidates)] in the order they came once + ``close()`` has said no more are coming and the last lookup is back. + + ``progress`` is (looked up, handed in).""" + + progress = pyqtSignal(int, int) + ready = pyqtSignal(list) + + def __init__(self, api_key: str, parent=None): + super().__init__(parent) + self._api_key = api_key + self._tracks: list = [] + self._results: dict = {} # track id → candidates + self._queue: list = [] + self._identifier = None + self._closed = False + self._cancelled = False + self._sent = False + + def add(self, tracks): + if self._cancelled: + return + self._tracks.extend(tracks) + self._queue.extend(tracks) + if self._identifier is None: + self._next() + + def close(self): + self._closed = True + self._maybe_ready() + + def cancel(self): + self._cancelled = True + self._queue.clear() + + def total(self) -> int: + return len(self._tracks) + + def _next(self): + if not self._queue or self._cancelled: + self._identifier = None + self._maybe_ready() + return + track = self._queue.pop(0) + # Kept so the identifier and its thread's signal source outlive + # this call. + self._identifier = TrackIdentifier() + self._identifier.finished.connect( + lambda result, t=track: self._on_result(t, result)) + self._identifier.identify(track, self._api_key) + + def _on_result(self, track, result: dict): + self._results[track.track_id] = result.get("candidates") or [] + self.progress.emit(len(self._results), len(self._tracks)) + self._next() + + def _maybe_ready(self): + if (self._closed and self._identifier is None and not self._sent + and not self._cancelled): + self._sent = True + self.ready.emit([(t, self._results.get(t.track_id, [])) + for t in self._tracks]) diff --git a/lintunes/library_manager.py b/lintunes/library_manager.py index fa32bcc..2608be7 100644 --- a/lintunes/library_manager.py +++ b/lintunes/library_manager.py @@ -605,9 +605,14 @@ class LibraryManager(QObject): track already has costs nothing and a track that ends up unchanged is skipped entirely (no file write, no undo entry). """ + self.edit_many_track_fields({tid: fields for tid in track_ids}) + + def edit_many_track_fields(self, edits: dict[int, dict]): + """``edit_tracks_fields`` with different fields per track ({track id: + fields}) — a batch identify's rows — still ONE undoable command.""" changes = [] # (track_id, new_fields, old_fields) - with timed("edit_tracks_fields (%d tracks)", len(track_ids)): - for track_id in track_ids: + with timed("edit_many_track_fields (%d tracks)", len(edits)): + for track_id, fields in edits.items(): track = self.library.tracks.get(track_id) if not track: continue diff --git a/tests/test_round75.py b/tests/test_round75.py index 1492bd3..bbbb901 100644 --- a/tests/test_round75.py +++ b/tests/test_round75.py @@ -337,7 +337,10 @@ def window(qapp, tmp_path, fake, monkeypatch): manager = LibraryManager(library, tmp_path / "data") win = MainWindow(manager, Preferences(tmp_path / "data")) monkeypatch.setattr(win, "_enqueue_identify", lambda tracks: None) + monkeypatch.setattr(win, "_show_batch_review", lambda analysis: None) yield win + if win._url_batch is not None: # lookups still out: never review them + win._url_batch.cancel() win.close() diff --git a/tests/test_round76.py b/tests/test_round76.py index d639fc3..9bc054b 100644 --- a/tests/test_round76.py +++ b/tests/test_round76.py @@ -321,7 +321,12 @@ def window(qapp, tmp_path, fake, monkeypatch): identified = [] monkeypatch.setattr(win, "_enqueue_identify", identified.extend) win.identified = identified + reviews = [] + monkeypatch.setattr(win, "_show_batch_review", reviews.append) + win.reviews = reviews yield win + if win._url_batch is not None: # lookups still out: never review them + win._url_batch.cancel() win.close() @@ -360,4 +365,9 @@ def test_spotify_songs_import_tagged_and_go_to_identify(window, qapp, fake, assert sorted((t.artist, t.album, t.name) for t in lib) == [ ("The Breeders", "Last Splash", "Cannonball"), ("The Breeders", "Last Splash", "New Year")] - assert len(window.identified) == 2 + # Several songs: no dialog each, one review once both are in. + assert window.identified == [] + _pump(qapp, lambda: window.reviews) + [analysis] = window.reviews + assert analysis.is_album and analysis.album == "Last Splash" + assert len(analysis.rows) == 2 diff --git a/tests/test_round77.py b/tests/test_round77.py new file mode 100644 index 0000000..a3e0ec0 --- /dev/null +++ b/tests/test_round77.py @@ -0,0 +1,339 @@ +"""Round 77: an import of several songs is identified as one batch. + +- Nothing pops up while songs are still downloading; they're looked up + quietly and reviewed together once the last one has landed. +- Songs that are one album are reviewed as one: album-wide fields once, + unified, plus a cover from the album-art search. +- A ticked row is applied with the rest (one undo step); an unticked one + gets the ordinary Identify window afterwards. +- No window ever lands on top of another app: it waits for LinTunes to + be in front. +""" +import pytest +from PyQt6.QtCore import Qt + +from lintunes import identify_batch +from lintunes.fingerprint import IdentifyCandidate +from lintunes.gui import identify_batch_dialog as dialog_module +from lintunes.gui.identify_batch_dialog import ( + IdentifyBatchDialog, format_cell, parse_cell, +) +from lintunes.identify_batch import ( + BatchRow, IdentifyCollector, analyze, find_common_album, +) +from lintunes.library_manager import LibraryManager +from lintunes.models import Library +from lintunes.models.track import Track +from lintunes.preferences import Preferences + + +def _c(name, album, year=None, number=None, count=None, score=0.9, + artist="The Breeders", album_artist=None, source="acoustid"): + return IdentifyCandidate(score=score, source=source, name=name, + artist=artist, album=album, year=year, + track_number=number, track_count=count, + album_artist=album_artist) + + +def _t(tid, name="", album="", artist="", **kw): + return Track(track_id=tid, name=name, album=album, artist=artist, **kw) + + +# -------------------------------------------------------------------------- +# A. one album or not + + +def test_songs_on_one_album_are_one_album(): + entries = [ + (_t(1, "x"), [_c("New Year", "Last Splash", 1993, 1, 15), + _c("New Year", "90s Hits", 2004, 7, 20)]), + (_t(2, "y"), [_c("Cannonball", "Last Splash", 1993, 2, 15)]), + (_t(3, "z"), [_c("Invisible Man", "Last Splash (Remastered)", 2013, + 3, 15)]), + ] + a = analyze(entries) + assert a.is_album and a.album == "Last Splash" + # The album's own year, not the remaster's, for all three. + assert [r.proposal["year"] for r in a.rows] == [1993] * 3 + assert [r.proposal["album"] for r in a.rows] == ["Last Splash"] * 3 + assert [r.proposal["track_number"] for r in a.rows] == [1, 2, 3] + assert a.rows[0].source == "AcoustID 90%" + + +def test_the_album_the_top_matches_name_beats_a_compilation(): + entries = [ + (_t(1), [_c("A", "Real Album"), _c("A", "Best Of")]), + (_t(2), [_c("B", "Real Album"), _c("B", "Best Of")]), + ] + assert find_common_album(entries) == "real album" + + +def test_a_song_tagged_with_the_album_votes_for_it(): + """A Spotify album import carries the album in its tags even where + AcoustID only knows the song from a compilation.""" + entries = [ + (_t(1, album="Last Splash"), [_c("A", "Greatest Hits")]), + (_t(2, album="Last Splash"), [_c("B", "Last Splash")]), + ] + a = analyze(entries) + assert a.album == "Last Splash" + # The first song keeps its own tags' album, not the compilation's. + assert a.rows[0].source == "Its own tags" + + +def test_a_playlist_is_not_an_album(): + entries = [ + (_t(1), [_c("A", "One", 1990, 1, 10)]), + (_t(2), [_c("B", "Two", 2000, 4, 12)]), + ] + a = analyze(entries) + assert not a.is_album + assert [r.proposal["album"] for r in a.rows] == ["One", "Two"] + assert identify_batch.album_art_query(a) is None + + +def test_songs_nobody_knows_dont_block_the_album(): + entries = [(_t(i), [_c(f"S{i}", "LP")]) for i in range(1, 5)] + entries.append((_t(9), [])) + a = analyze(entries) + assert a.is_album and a.rows[-1].proposal["album"] == "LP" + assert a.rows[-1].source == "Nothing found" + + +def test_one_song_is_never_an_album(): + assert find_common_album([(_t(1), [_c("A", "LP")])]) is None + + +def test_changes_never_blank_a_tag(): + row = BatchRow(_t(1, "Song", "LP", "Me", year=1999), + {"name": "Song", "album": None, "year": 2001, + "artist": ""}, "x") + assert row.changes() == {"year": 2001} + + +# -------------------------------------------------------------------------- +# B. the window + + +@pytest.fixture +def no_network(monkeypatch): + searched = [] + + class _Fetcher: + def __init__(self): + from PyQt6.QtCore import QObject, pyqtSignal + + class _S(QObject): + search_finished = pyqtSignal(object) + image_finished = pyqtSignal(object) + self._s = _S() + self.search_finished = self._s.search_finished + self.image_finished = self._s.image_finished + + def search(self, artist, album): + searched.append((artist, album)) + + def fetch(self, candidate): + pass + monkeypatch.setattr(dialog_module, "AlbumArtFetcher", _Fetcher) + return searched + + +def _album_analysis(): + return analyze([ + (_t(1, "new year", artist="The Breeders"), + [_c("New Year", "Last Splash", 1993, 1, 2)]), + (_t(2, "Cannonball", artist="The Breeders", album="Last Splash", + year=1993, track_number=2, track_count=2), + [_c("Cannonball", "Last Splash", 1993, 2, 2)]), + ]) + + +def test_cells_round_trip(): + assert format_cell(("track_number", "track_count"), + {"track_number": 3, "track_count": 12}) == "3 of 12" + assert parse_cell(("track_number", "track_count"), "3/12") == { + "track_number": 3, "track_count": 12} + assert parse_cell(("track_number", "track_count"), "huh") == { + "track_number": None, "track_count": None} + assert parse_cell(("year",), "") == {"year": None} + + +def test_the_album_window_applies_ticked_rows(qapp, no_network): + analysis = _album_analysis() + dialog = IdentifyBatchDialog( + analysis, identify_batch.album_art_query(analysis)) + assert no_network == [("The Breeders", "Last Splash")] + # Only the first song changes; the second already says all of it. + assert dialog.edits() == {1: {"name": "New Year", "album": "Last Splash", + "year": 1993, "track_number": 1, + "track_count": 2}} + name_cell = dialog._table.item(0, 1) + assert name_cell.font().bold() and "new year" in name_cell.toolTip() + assert not dialog._table.item(1, 1).font().bold() + + # The album-wide field is edited once, for every row. + dialog._album_edit.setText("Last Splash (4AD)") + dialog._album_edit.textEdited.emit("Last Splash (4AD)") + edits = dialog.edits() + assert edits[1]["album"] == edits[2]["album"] == "Last Splash (4AD)" + + dialog._table.item(1, 0).setCheckState(Qt.CheckState.Unchecked) + assert list(dialog.edits()) == [1] + assert [r.track.track_id for r in dialog.unchecked_rows()] == [2] + dialog._on_apply() + assert dialog.outcome == dialog.APPLY + assert dialog.selected_image is None # no art loaded + + +def test_art_arrives_ticked_and_is_handed_over(qapp, no_network): + from lintunes.art_search import ArtCandidate + from PyQt6.QtCore import QBuffer, QByteArray, QIODevice + from PyQt6.QtGui import QImage + image = QImage(4, 4, QImage.Format.Format_RGB32) + image.fill(0xff0000) + data = QByteArray() + buf = QBuffer(data) + buf.open(QIODevice.OpenModeFlag.WriteOnly) + image.save(buf, "PNG") + + analysis = _album_analysis() + dialog = IdentifyBatchDialog(analysis, ("The Breeders", "Last Splash")) + art = ArtCandidate("The Breeders", "Last Splash", "http://x/a.jpg") + dialog._on_art_search({"candidates": [art]}) + dialog._on_art_image({"candidate": art, "image": bytes(data), + "mime": "image/png"}) + assert dialog._embed.isChecked() + dialog._on_apply() + assert dialog.selected_image == bytes(data) + assert dialog.selected_mime == "image/png" + + +def test_a_playlist_window_has_no_album_panel(qapp, no_network): + analysis = analyze([(_t(1), [_c("A", "One")]), + (_t(2), [_c("B", "Two")])]) + dialog = IdentifyBatchDialog(analysis, None) + assert dialog._album_edit is None and dialog._embed is None + assert no_network == [] + dialog._on_review_all() + assert dialog.outcome == dialog.REVIEW_ALL + + +# -------------------------------------------------------------------------- +# C. the manager: different fields per song, one undo step + + +def test_edit_many_track_fields_is_one_undo(tmp_path, mp3_file): + import shutil + library = Library() + for tid in (1, 2): + path = tmp_path / f"{tid}.mp3" + shutil.copy(mp3_file, path) + library.tracks[tid] = _t(tid, f"old{tid}", location=str(path)) + manager = LibraryManager(library, tmp_path / "data") + manager.edit_many_track_fields({1: {"name": "One"}, + 2: {"name": "Two", "year": 1993}}) + assert (library.tracks[1].name, library.tracks[2].year) == ("One", 1993) + manager.undo_stack.undo() + assert (library.tracks[1].name, library.tracks[2].name, + library.tracks[2].year) == ("old1", "old2", 0) + + +# -------------------------------------------------------------------------- +# D. the collector + + +def test_the_collector_waits_for_close_and_keeps_order(qapp, monkeypatch): + pending = [] + + class _Identifier: + def __init__(self): + from PyQt6.QtCore import QObject, pyqtSignal + + class _S(QObject): + finished = pyqtSignal(object) + self._s = _S() + self.finished = self._s.finished + + def identify(self, track, key): + pending.append((self, track)) + monkeypatch.setattr(identify_batch, "TrackIdentifier", _Identifier) + + collector = IdentifyCollector("") + ready = [] + collector.ready.connect(ready.append) + collector.add([_t(1)]) + collector.add([_t(2)]) + assert len(pending) == 1 # one lookup at a time + ident, track = pending.pop() + ident.finished.emit({"candidates": [_c("A", "LP")]}) + ident, track = pending.pop() + ident.finished.emit({"error": "nope"}) + assert ready == [] # more songs could still come + collector.close() + [entries] = ready + assert [(t.track_id, len(c)) for t, c in entries] == [(1, 1), (2, 0)] + + +# -------------------------------------------------------------------------- +# E. the window waits for LinTunes to be in front + + +@pytest.fixture +def window(qapp, tmp_path): + from lintunes.gui.main_window import MainWindow + manager = LibraryManager(Library(), tmp_path / "data") + win = MainWindow(manager, Preferences(tmp_path / "data")) + yield win + win.close() + + +def test_popups_wait_until_lintunes_is_in_front(window, qapp, monkeypatch): + from lintunes.gui import main_window as mw + state = [Qt.ApplicationState.ApplicationInactive] + monkeypatch.setattr(qapp, "applicationState", lambda: state[0]) + alerts = [] + monkeypatch.setattr(mw.QApplication, "alert", + lambda w, ms=0: alerts.append(w)) + shown = [] + window._when_app_active(lambda: shown.append(1)) + assert shown == [] and alerts == [window] + window._on_app_state_changed(Qt.ApplicationState.ApplicationActive) + qapp.processEvents() + assert shown == [1] + state[0] = Qt.ApplicationState.ApplicationActive + window._when_app_active(lambda: shown.append(2)) + assert shown == [1, 2] + + +def test_a_batch_review_applies_then_reviews_the_unticked(window, qapp, + monkeypatch): + from lintunes.gui import main_window as mw + library = window._manager.library + for tid in (1, 2, 3): + library.tracks[tid] = _t(tid, f"t{tid}") + rows = [BatchRow(library.tracks[t], {}, "x", [_c(f"S{t}", "LP")]) + for t in (1, 2, 3)] + applied, individually = [], [] + + class _Batch: + APPLY, REVIEW_ALL = 1, 2 + outcome = 1 + selected_image = None + def __init__(self, analysis, query, parent): + pass + def exec(self): + return True + def edits(self): + return {1: {"name": "S1"}} + def unchecked_rows(self): + return rows[2:] + def all_rows(self): + return rows + monkeypatch.setattr(mw, "IdentifyBatchDialog", _Batch) + monkeypatch.setattr(window._manager, "edit_many_track_fields", + applied.append) + monkeypatch.setattr(window, "_review_individually", individually.append) + window._show_batch_review(identify_batch.BatchAnalysis(rows)) + assert applied == [{1: {"name": "S1"}}] + assert individually == [rows[2:]]