From 439a7fa75d3e12623846de52319eaed208fbe44d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=A0ar=C5=ABnas=20Nejus?= Date: Sun, 5 Jul 2026 12:34:42 +0100 Subject: [PATCH] discogs: Refactor _coalesce_tracks to make use of various tracks type_ --- beetsplug/discogs/__init__.py | 152 +++++++++++++++++----------------- beetsplug/discogs/states.py | 2 +- test/plugins/test_discogs.py | 14 ++++ 3 files changed, 90 insertions(+), 78 deletions(-) diff --git a/beetsplug/discogs/__init__.py b/beetsplug/discogs/__init__.py index 4a4f4a85c..e21929c58 100644 --- a/beetsplug/discogs/__init__.py +++ b/beetsplug/discogs/__init__.py @@ -12,6 +12,7 @@ import socket import time import traceback from functools import cache, cached_property +from itertools import groupby from string import ascii_lowercase from typing import TYPE_CHECKING @@ -35,7 +36,7 @@ if TYPE_CHECKING: from beets.library import Item from beets.metadata_plugins import QueryType, SearchParams - from .types import ReleaseFormat, Track + from .types import AudioTrack, IndexTrack, ReleaseFormat, Track USER_AGENT = f"beets/{beets.__version__} +https://beets.io/" API_KEY = "rAzVUQYRaoFjeBjyWuWZ" @@ -532,91 +533,88 @@ class DiscogsPlugin(SearchApiMetadataSourcePlugin[IDResponse]): return t.tracks + def _subtrack_position(self, track: Track) -> str | None: + """Identify the physical position containing a flat audio subtrack.""" + if track["type_"] == "track": + medium, medium_index, subindex = self.get_track_index( + track["position"] + ) + if subindex: + return f"{medium or ''}{medium_index or ''}" + return None + def _coalesce_tracks(self, raw_tracklist: list[Track]) -> list[Track]: - """Pre-process a tracklist, merging subtracks into a single track. The - title for the merged track is the one from the previous index track, - if present; otherwise it is a combination of the subtracks titles. - """ - # Pre-process the tracklist, trying to identify subtracks. - - subtracks: list[Track] = [] + """Normalize each Discogs tracklist entry by its declared kind.""" tracklist: list[Track] = [] - prev_subindex = "" - for track in raw_tracklist: - # Regular subtrack (track with subindex). - if track["position"]: - _, _, subindex = self.get_track_index(track["position"]) - if subindex: - if subindex.rjust(len(raw_tracklist)) > prev_subindex: - # Subtrack still part of the current main track. - subtracks.append(track) - else: - # Subtrack part of a new group (..., 1.3, *2.1*, ...). - self._add_merged_subtracks(tracklist, subtracks) - subtracks = [track] - prev_subindex = subindex.rjust(len(raw_tracklist)) - continue - - # Index track with nested sub_tracks. - if not track["position"] and "sub_tracks" in track: - # Append the index track, assuming it contains the track title. - tracklist.append(track) - self._add_merged_subtracks(tracklist, track["sub_tracks"]) + for position, tracks in groupby( + raw_tracklist, key=self._subtrack_position + ): + if position is not None: + subtracks = [ + track for track in tracks if track["type_"] == "track" + ] + tracklist.append(self._merge_subtracks(subtracks)) continue - # Regular track or index track without nested sub_tracks. - if subtracks: - self._add_merged_subtracks(tracklist, subtracks) - subtracks = [] - prev_subindex = "" - tracklist.append(track) - - # Merge and add the remaining subtracks, if any. - if subtracks: - self._add_merged_subtracks(tracklist, subtracks) + for track in tracks: + if track["type_"] in {"track", "heading"}: + tracklist.append(track) + elif track["type_"] == "index": + tracklist.extend(self._coalesce_index_track(track)) return tracklist - def _add_merged_subtracks( - self, tracklist: list[Track], subtracks: list[Track] - ) -> None: - """Modify `tracklist` in place, merging a list of `subtracks` into - a single track into `tracklist`.""" - # Calculate position based on first subtrack, without subindex. - idx, medium_idx, sub_idx = self.get_track_index( + @staticmethod + def _merge_subtracks(subtracks: list[AudioTrack]) -> AudioTrack: + """Combine flat Discogs subtracks representing one physical track.""" + merged_track = subtracks[0].copy() + merged_track["title"] = " / ".join( + subtrack["title"] for subtrack in subtracks + ) + return merged_track + + def _coalesce_index_track( + self, index_track: IndexTrack + ) -> list[AudioTrack]: + """Convert an index container into its physical audio tracks.""" + subtracks = index_track["sub_tracks"] + if not subtracks: + raise ValueError("Discogs index track does not contain subtracks") + + medium, medium_index, subindex = self.get_track_index( subtracks[0]["position"] ) - position = f"{idx or ''}{medium_idx or ''}" - if tracklist and not tracklist[-1]["position"]: - # Assume the previous index track contains the track title. - if sub_idx: - # "Convert" the track title to a real track, discarding the - # subtracks assuming they are logical divisions of a - # physical track (12.2.9 Subtracks). - tracklist[-1]["position"] = position - else: - # Promote the subtracks to real tracks, discarding the - # index track, assuming the subtracks are physical tracks. - index_track = tracklist.pop() - # Fix artists when they are specified on the index track. - if index_track.get("artists"): - for subtrack in subtracks: - if not subtrack.get("artists"): - subtrack["artists"] = index_track["artists"] - # Concatenate index with track title when index_tracks - # option is set - if self.config["index_tracks"]: - for subtrack in subtracks: - subtrack["title"] = ( - f"{index_track['title']}: {subtrack['title']}" - ) - tracklist.extend(subtracks) - else: - # Merge the subtracks, pick a title, and append the new track. - track = subtracks[0].copy() - track["title"] = " / ".join([t["title"] for t in subtracks]) - tracklist.append(track) + # Subindexed children are logical pieces contained in one physical + # track. Keep the index track's metadata and discard the child entries. + if subindex: + merged_track: AudioTrack = { + "type_": "track", + "position": f"{medium or ''}{medium_index or ''}", + "title": index_track["title"], + "duration": index_track["duration"], + } + if artists := index_track.get("artists"): + merged_track["artists"] = artists + if extraartists := index_track.get("extraartists"): + merged_track["extraartists"] = extraartists + return [merged_track] + + # Children without subindices are separate physical tracks grouped + # under a musical work. Discard the index while retaining its context. + artists = index_track.get("artists") + title_prefix = ( + f"{index_track['title']}: " if self.config["index_tracks"] else "" + ) + tracks: list[AudioTrack] = [] + for subtrack in subtracks: + track = subtrack.copy() + if artists and not track.get("artists"): + track["artists"] = artists + if title_prefix: + track["title"] = f"{title_prefix}{track['title']}" + tracks.append(track) + return tracks def strip_disambiguation(self, text: str) -> str: """Removes discogs specific disambiguations from a string. @@ -628,7 +626,7 @@ class DiscogsPlugin(SearchApiMetadataSourcePlugin[IDResponse]): def get_track_info( self, - track: Track, + track: AudioTrack, index: int, divisions: list[str], albumartistinfo: ArtistState, diff --git a/beetsplug/discogs/states.py b/beetsplug/discogs/states.py index 474c46077..feecb6b46 100644 --- a/beetsplug/discogs/states.py +++ b/beetsplug/discogs/states.py @@ -236,7 +236,7 @@ class TracklistState: ) -> TracklistState: state = cls() for track in clean_tracklist: - if track["position"]: + if track["type_"] == "track": state.index += 1 if state.next_divisions: state.divisions += state.next_divisions diff --git a/test/plugins/test_discogs.py b/test/plugins/test_discogs.py index eb66fe378..e0b40772f 100644 --- a/test/plugins/test_discogs.py +++ b/test/plugins/test_discogs.py @@ -448,6 +448,20 @@ class TestTracklist(DiscogsTestMixin): assert album.mediums == 1 assert {t.artist for t in album.tracks} == {track_artist, group_artist} + def test_coalesce_logical_index_as_audio_track(self, plugin): + index = index_track( + "TRACK GROUP TITLE", + [ + audio_track("SUBTITLE ONE", "2.1"), + audio_track("SUBTITLE TWO", "2.2"), + ], + duration="03:03", + ) + + assert plugin._coalesce_tracks([index]) == [ + audio_track("TRACK GROUP TITLE", "2", "03:03") + ] + class TestDGSearchQuery(TestHelper): def test_default_search_filters_without_extra_tags(self):