discogs: Refactor _coalesce_tracks to make use of various tracks type_

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,

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

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):