v0.15.0: ask the song what it is

Right-click → Identify Track… fingerprints the file with Chromaprint's
fpcalc and asks AcoustID who it is, then proposes the tags it found.

The year is the point. A 60s song kept getting stamped with the year its
CD reissue came out, which sorts the library wrong, so every candidate
carries the recording's *original* year — the earliest release across all
of its release groups — including the candidate that proposes a later
greatest-hits as the album. Ranking separately prefers a plain studio
Album over EP/Single over anything wearing a Compilation or Live
secondary type, so the default proposal is the real album as well.

Nothing is written until accepted: the dialog shows each current value
beside an editable proposed one, and a row starts checked only where the
proposal actually differs from what the file already has, so an untouched
field can never quietly blank a tag. Applying goes through
edit_track_fields, which is what buys the tag write, the abort-if-it-fails
rule, artist/album file relocation and one undo step without re-deriving
any of them.

A selection is a queue reviewed one track at a time; a failure is a
status-bar message that moves on rather than ending the batch. fpcalc is
detected at runtime like ffmpeg and the AcoustID key lives in preferences
(so it syncs, and stays out of git) — the round adds no new dependency.

Also fixes a pre-existing round 40 test that used the real tummult mount
point as its stand-in for an unreachable path, so it failed whenever that
drive was plugged in.

Claude-Session: https://claude.ai/code/session_01NsiFHdyVg1UhBSxDJfTNRm
This commit is contained in:
2026-09-02 15:40:52 -04:00
parent 5831175584
commit 312748d05e
12 changed files with 946 additions and 2 deletions
+1 -1
View File
@@ -1,3 +1,3 @@
"""LinTunes — iTunes-style music library manager and player for Linux."""
__version__ = "0.14.0"
__version__ = "0.15.0"
+248
View File
@@ -0,0 +1,248 @@
"""Track identification via Chromaprint (fpcalc) + the AcoustID web API.
Pure parsing/ranking helpers are separated from the subprocess and network
calls so they can be tested offline with canned AcoustID JSON;
``TrackIdentifier`` runs the whole fingerprint+lookup on a daemon thread (the
art_search pattern) and reports back over a Qt signal, which is delivered
queued on the GUI thread.
fpcalc is the Chromaprint *CLI binary* (Fedora: chromaprint-tools, Debian:
libchromaprint-tools) — detected at runtime, never a pip dependency. The
AcoustID application key is user-supplied via Preferences (the Last.fm
precedent); registration is free at https://acoustid.org/new-application.
"""
import json
import shutil
import subprocess
import threading
from dataclasses import asdict, dataclass
from PyQt6.QtCore import QObject, pyqtSignal
ACOUSTID_LOOKUP_URL = "https://api.acoustid.org/v2/lookup"
LOOKUP_META = "recordings releasegroups releases tracks compress"
SCORE_THRESHOLD = 0.5 # AcoustID scores below this are noise
MAX_CANDIDATES = 8
TIMEOUT_S = 15
FPCALC_TIMEOUT_S = 60
@dataclass
class IdentifyCandidate:
"""One proposed identification. Field names match Track/tagging so
``fields()`` can go straight into ``LibraryManager.edit_track_fields``."""
score: float
name: str | None = None
artist: str | None = None
album_artist: str | None = None
album: str | None = None
year: int | None = None
track_number: int | None = None
track_count: int | None = None
disc_number: int | None = None
def fields(self) -> dict:
"""Proposed edits keyed by LinTunes field names, Nones omitted."""
return {k: v for k, v in asdict(self).items()
if k != "score" and v is not None}
# ---- pure parsing/ranking helpers (offline-testable) ----
def _join_artists(artists: list | None) -> str | None:
"""Concatenate MusicBrainz artist credits, honoring ``joinphrase``."""
if not artists:
return None
parts = []
for i, artist in enumerate(artists):
name = artist.get("name", "")
if not name:
continue
if parts:
parts.append(artists[i - 1].get("joinphrase") or "; ")
parts.append(name)
return "".join(parts) or None
def _release_year(release: dict) -> int | None:
year = (release.get("date") or {}).get("year")
return year if isinstance(year, int) and year > 0 else None
def _oldest_year(releasegroup: dict) -> int | None:
"""Earliest dated release in the group — the original pressing, not the
CD reissue."""
years = [y for r in releasegroup.get("releases", [])
if (y := _release_year(r)) is not None]
return min(years) if years else None
def _releasegroup_sort_key(releasegroup: dict) -> tuple:
"""Rank a release group for the default proposal: the earliest proper
studio Album first, then EPs/Singles, then Compilations/Live/etc."""
rg_type = releasegroup.get("type") or ""
secondary = releasegroup.get("secondarytypes") or []
if not secondary:
rank = {"Album": 0, "EP": 1, "Single": 2}.get(rg_type, 4)
else:
rank = 3 if rg_type == "Album" else 4
return (rank, _oldest_year(releasegroup) or 9999)
def _pick_release(releasegroup: dict) -> dict | None:
"""Earliest dated release that carries mediums (track positions); falls
back to the earliest dated one, then the first."""
releases = releasegroup.get("releases", [])
if not releases:
return None
dated = sorted(releases, key=lambda r: _release_year(r) or 9999)
for release in dated:
if release.get("mediums"):
return release
return dated[0]
def _track_position(release: dict | None) -> tuple[int | None, int | None,
int | None]:
"""(track_number, track_count, disc_number) from the matching medium.
With ``meta=tracks`` AcoustID returns only the medium/track entries that
match the looked-up recording, so the first entries are the match.
"""
if not release:
return None, None, None
mediums = release.get("mediums") or []
if not mediums:
return None, None, None
medium = mediums[0]
tracks = medium.get("tracks") or []
number = tracks[0].get("position") if tracks else None
count = medium.get("track_count")
disc = medium.get("position")
return (number if isinstance(number, int) else None,
count if isinstance(count, int) else None,
disc if isinstance(disc, int) else None)
def parse_lookup(payload: dict, threshold: float = SCORE_THRESHOLD,
limit: int = MAX_CANDIDATES) -> list[IdentifyCandidate]:
"""Turn an AcoustID lookup response into ranked candidates.
One candidate per release group of each matched recording, so the
alternates offered in the dialog genuinely differ (different album/year).
Every candidate of a recording carries the recording's *original* year —
the minimum across all its release groups' releases — even when the
proposed album is a later compilation: years sort by when the song first
came out, not by which pressing this file happens to be from.
"""
candidates = []
seen = set()
results = sorted(payload.get("results", []),
key=lambda r: r.get("score", 0), reverse=True)
for result in results:
score = result.get("score", 0)
if score < threshold:
continue
for recording in result.get("recordings", []):
title = recording.get("title")
artist = _join_artists(recording.get("artists"))
groups = sorted(recording.get("releasegroups") or [],
key=_releasegroup_sort_key)
original_years = [y for rg in groups
if (y := _oldest_year(rg)) is not None]
original_year = min(original_years) if original_years else None
if not groups:
if title:
candidates.append(IdentifyCandidate(
score=score, name=title, artist=artist))
continue
for releasegroup in groups:
album = releasegroup.get("title")
key = (title, artist, album)
if key in seen:
continue
seen.add(key)
number, count, disc = _track_position(
_pick_release(releasegroup))
candidates.append(IdentifyCandidate(
score=score,
name=title,
artist=artist,
album_artist=_join_artists(releasegroup.get("artists")),
album=album,
year=original_year,
track_number=number,
track_count=count,
disc_number=disc,
))
return candidates[:limit]
# ---- subprocess + network ----
def fpcalc_available() -> bool:
"""Whether the fpcalc *binary* is on PATH (the ffmpeg_available pattern —
Chromaprint's CLI tool, not a Python package)."""
return shutil.which("fpcalc") is not None
def fingerprint_file(location: str) -> tuple[int, str]:
"""Run fpcalc on a file; returns (duration seconds, fingerprint).
Raises OSError on failure, carrying fpcalc's last stderr line."""
try:
proc = subprocess.run(
["fpcalc", "-json", location],
stdin=subprocess.DEVNULL, capture_output=True,
timeout=FPCALC_TIMEOUT_S)
except (OSError, ValueError, subprocess.TimeoutExpired) as e:
raise OSError(f"couldn't run fpcalc: {e}") from e
name = location.rsplit("/", 1)[-1]
if proc.returncode != 0:
detail = proc.stderr.decode("utf-8", "replace").strip().splitlines()
raise OSError(f"fpcalc failed on {name}"
+ (f": {detail[-1]}" if detail else ""))
try:
payload = json.loads(proc.stdout.decode("utf-8", "replace"))
return int(payload["duration"]), payload["fingerprint"]
except (ValueError, KeyError, TypeError) as e:
raise OSError(f"couldn't read fpcalc output for {name}") from e
def lookup_fingerprint(api_key: str, duration: int,
fingerprint: str) -> list[IdentifyCandidate]:
import requests
response = requests.post(
ACOUSTID_LOOKUP_URL,
data={"client": api_key, "duration": duration,
"fingerprint": fingerprint, "meta": LOOKUP_META,
"format": "json"},
timeout=TIMEOUT_S,
)
response.raise_for_status()
payload = response.json()
if payload.get("status") != "ok":
message = (payload.get("error") or {}).get("message", "lookup failed")
raise RuntimeError(message)
return parse_lookup(payload)
class TrackIdentifier(QObject):
"""One fingerprint+lookup running off the GUI thread.
``finished`` carries {"track_id": int, "candidates": [IdentifyCandidate]}
or {"track_id": int, "error": str}.
"""
finished = pyqtSignal(object)
def identify(self, track_id: int, location: str, api_key: str):
def work():
try:
duration, fp = fingerprint_file(location)
candidates = lookup_fingerprint(api_key, duration, fp)
self.finished.emit(
{"track_id": track_id, "candidates": candidates})
except Exception as e:
self.finished.emit({"track_id": track_id, "error": str(e)})
threading.Thread(target=work, daemon=True).start()
+143
View File
@@ -0,0 +1,143 @@
"""Confirmation dialog for a track identification (Identify Track…).
Shows the proposed tag changes for one track — current value vs an editable
proposed value per field, with a checkbox to opt out of any row — and a
dropdown of alternate candidates (other albums/releases the recording appears
on). Passive like AlbumArtDialog: candidates arrive pre-parsed, nothing is
written until the user accepts, and the caller applies ``result_fields()``
through LibraryManager afterwards.
"""
from PyQt6.QtWidgets import (
QCheckBox, QComboBox, QDialog, QDialogButtonBox, QGridLayout, QLabel,
QLineEdit, QPushButton, QSpinBox, QVBoxLayout,
)
# (field, label, is_number, spinbox max)
_FIELDS = [
("name", "Name", False, 0),
("artist", "Artist", False, 0),
("album_artist", "Album Artist", False, 0),
("album", "Album", False, 0),
("year", "Year", True, 9999),
("track_number", "Track #", True, 999),
("track_count", "Track Count", True, 999),
("disc_number", "Disc #", True, 99),
]
class IdentifyDialog(QDialog):
def __init__(self, track, candidates, remaining: int = 0, parent=None):
super().__init__(parent)
self.setWindowTitle("Identify Track")
self._track = track
self._candidates = candidates
# True when the user pressed "Stop Identifying" on a queued batch.
self.cancel_all = False
layout = QVBoxLayout(self)
best = candidates[0]
header = QLabel(f"Identified “{track.name or '(untitled)'}” — "
f"match confidence {round(best.score * 100)}%")
header.setWordWrap(True)
layout.addWidget(header)
# All alternates at a glance — a candidate summarizes to one line, so
# a dropdown beats AlbumArtDialog's blind Next-clicking.
self._combo = QComboBox()
for c in candidates:
year = f" ({c.year})" if c.year else ""
self._combo.addItem(" — ".join(
p for p in (c.name, c.artist, c.album) if p) + year)
layout.addWidget(self._combo)
grid = QGridLayout()
grid.setColumnStretch(1, 1)
grid.setColumnStretch(3, 1)
self._checks: dict[str, QCheckBox] = {}
self._current_labels: dict[str, QLabel] = {}
self._editors: dict[str, QLineEdit | QSpinBox] = {}
for row, (field, label, is_number, maximum) in enumerate(_FIELDS):
check = QCheckBox(label)
current = QLabel()
current.setEnabled(False) # the greyed-hint idiom
arrow = QLabel("→")
if is_number:
editor = QSpinBox()
editor.setRange(0, maximum)
editor.setSpecialValueText(" ")
else:
editor = QLineEdit()
check.toggled.connect(editor.setEnabled)
self._checks[field] = check
self._current_labels[field] = current
self._editors[field] = editor
grid.addWidget(check, row, 0)
grid.addWidget(current, row, 1)
grid.addWidget(arrow, row, 2)
grid.addWidget(editor, row, 3)
layout.addLayout(grid)
buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Ok
| QDialogButtonBox.StandardButton.Cancel)
buttons.button(QDialogButtonBox.StandardButton.Ok).setText(
"Apply Tags")
if remaining > 0:
noun = "track" if remaining == 1 else "tracks"
stop_btn = QPushButton(f"Stop Identifying ({remaining} {noun} left)")
buttons.addButton(stop_btn,
QDialogButtonBox.ButtonRole.ActionRole)
stop_btn.clicked.connect(self._stop_all)
buttons.accepted.connect(self.accept)
buttons.rejected.connect(self.reject)
layout.addWidget(buttons)
self._combo.currentIndexChanged.connect(self._fill_from_candidate)
self._fill_from_candidate(0)
# ---- candidate switching ----
def _current_value(self, field):
return getattr(self._track, field, None)
def _fill_from_candidate(self, index: int):
"""Re-fill the grid from a candidate (hand edits are discarded on a
switch, like AlbumArtDialog cycling). A row is checked and editable
iff the candidate proposes a value that differs from the current one;
rows with nothing proposed are unchecked and disabled."""
candidate = self._candidates[index]
for field, _label, is_number, _max in _FIELDS:
current = self._current_value(field)
proposed = getattr(candidate, field)
label = self._current_labels[field]
editor = self._editors[field]
check = self._checks[field]
label.setText(str(current) if current else "—")
if is_number:
editor.setValue(proposed or 0)
else:
editor.setText(proposed or "")
has_proposal = proposed is not None
differs = has_proposal and proposed != (current or None)
check.setEnabled(has_proposal)
check.setChecked(differs)
editor.setEnabled(differs)
def _stop_all(self):
self.cancel_all = True
self.reject()
# ---- result ----
def result_fields(self) -> dict:
"""Checked rows only, keyed by LinTunes field names — ready for
``LibraryManager.edit_track_fields``. Blank editors are dropped (a
checked row the user then emptied proposes nothing)."""
fields = {}
for field, _label, is_number, _max in _FIELDS:
if not self._checks[field].isChecked():
continue
editor = self._editors[field]
value = editor.value() if is_number else editor.text().strip()
if value:
fields[field] = value
return fields
+90
View File
@@ -14,12 +14,14 @@ from lintunes.art_search import AlbumArtFetcher
from lintunes.cast.controller import CastController
from lintunes.eventlog import log_control
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.cast_dialog import ChromecastDialog
from lintunes.gui.export_dialog import ExportKindDialog, WebMixDialog
from lintunes.gui.identify_dialog import IdentifyDialog
from lintunes.gui.sidebar import SidebarPanel
from lintunes.gui.library_view import LibraryView
from lintunes.gui.playlist_view import PlaylistView
@@ -159,6 +161,11 @@ class MainWindow(QMainWindow):
# 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
# 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.
@@ -182,6 +189,7 @@ class MainWindow(QMainWindow):
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(
@@ -1077,6 +1085,88 @@ class MainWindow(QMainWindow):
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._identify_queue = tracks
self._identify_total = len(tracks)
self._identify_next(api_key)
def _identify_next(self, api_key: str):
if not self._identify_queue:
return
track = self._identify_queue.pop(0)
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, k=api_key: self._on_identified(t, k, result))
self._identifier.identify(track.track_id, track.location, api_key)
def _on_identified(self, track, api_key: str, 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"No match found for “{track.name}”", 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.statusBar().clearMessage()
return
self._identify_next(api_key)
# ---- relocating a moved/renamed file ----
def _existing_ancestor(self, path: str) -> str:
+26
View File
@@ -55,6 +55,7 @@ class PreferencesDialog(QDialog):
layout.addWidget(self._build_music_group())
layout.addWidget(self._build_colors_group())
layout.addWidget(self._build_lastfm_group())
layout.addWidget(self._build_acoustid_group())
layout.addStretch()
lastfm.login_finished.connect(self._on_login_finished)
@@ -286,3 +287,28 @@ class PreferencesDialog(QDialog):
if logged_in and not self._status_label.text():
self._status_label.setText(
f"Logged in as {self._prefs.lastfm.get('username', '')}")
# ---- Track identification (AcoustID) ----
def _build_acoustid_group(self) -> QGroupBox:
group = QGroupBox("Track Identification")
form = QFormLayout(group)
hint = QLabel(
'Right-click a track → Identify Track… Needs a free application '
'key: <a href="https://acoustid.org/new-application">'
'acoustid.org/new-application</a>')
hint.setOpenExternalLinks(True)
hint.setWordWrap(True)
form.addRow(hint)
self._acoustid_key = QLineEdit(
self._prefs.acoustid.get("api_key", ""))
self._acoustid_key.editingFinished.connect(self._save_acoustid_key)
form.addRow("AcoustID key:", self._acoustid_key)
return group
def _save_acoustid_key(self):
key = self._acoustid_key.text().strip()
if key != self._prefs.acoustid.get("api_key", ""):
self._prefs.update_acoustid(api_key=key)
+6
View File
@@ -506,6 +506,7 @@ class TrackTableView(QTableView):
show_in_playlist_requested = pyqtSignal(int, str) # track_id, playlist pid
rating_edited = pyqtSignal(int, int) # track_id, new rating 0-100
download_art_requested = pyqtSignal(list) # selected track ids
identify_requested = pyqtSignal(list) # selected track ids
remove_from_library_requested = pyqtSignal(list) # track ids, file kept
delete_from_library_requested = pyqtSignal(list) # track ids, file trashed
@@ -765,11 +766,14 @@ class TrackTableView(QTableView):
locations = self.selected_locations()
reveal_action = copy_path_action = download_art_action = None
identify_action = None
if locations:
menu.addSeparator()
reveal_action = menu.addAction("Reveal in File Browser")
copy_path_action = menu.addAction("Copy File Path")
download_art_action = menu.addAction("Download Album Art…")
# Fingerprinting reads the audio, so this needs a file on disk.
identify_action = menu.addAction("Identify Track…")
# "Show in Playlist" — only for a single track, listing the regular
# playlists it belongs to. Hovering reveals the submenu; choosing an
@@ -813,6 +817,8 @@ class TrackTableView(QTableView):
QApplication.clipboard().setText("\n".join(locations))
elif download_art_action is not None and chosen is download_art_action:
self.download_art_requested.emit(self.selected_track_ids())
elif identify_action is not None and chosen is identify_action:
self.identify_requested.emit(self.selected_track_ids())
elif show_in_menu is not None and chosen in show_in_menu.actions():
self.show_in_playlist_requested.emit(selected[0], chosen.data())
elif remove_action is not None and chosen is remove_action:
+14
View File
@@ -28,6 +28,11 @@ DEFAULTS = {
"session_key": "",
"scrobble_enabled": False,
},
# Track identification (Identify Track…). The application key is the
# user's own, registered free at acoustid.org/new-application.
"acoustid": {
"api_key": "",
},
}
@@ -75,6 +80,15 @@ class Preferences(QObject):
def lastfm(self) -> dict:
return self._data["lastfm"]
def update_acoustid(self, **fields):
self._data["acoustid"].update(fields)
self.save()
self.changed.emit()
@property
def acoustid(self) -> dict:
return self._data["acoustid"]
def save(self):
self._path.parent.mkdir(parents=True, exist_ok=True)
tmp = self._path.with_suffix(".json.tmp")