From 8867848612ce822ae880aa06345992a76dac09bf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=A0ar=C5=ABnas=20Nejus?= Date: Sun, 5 Jul 2026 12:22:44 +0100 Subject: [PATCH 1/2] discogs: Add descriptions of each possible Tracks --- beetsplug/discogs/types.py | 49 ++++++++++++-- test/plugins/factories/discogs.py | 23 +++++++ test/plugins/test_discogs.py | 109 ++++++++++++++++++------------ 3 files changed, 131 insertions(+), 50 deletions(-) create mode 100644 test/plugins/factories/discogs.py diff --git a/beetsplug/discogs/types.py b/beetsplug/discogs/types.py index b3e69c6aa..949444721 100644 --- a/beetsplug/discogs/types.py +++ b/beetsplug/discogs/types.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Literal from typing_extensions import NotRequired, TypedDict @@ -24,14 +24,53 @@ class Artist(TypedDict): resource_url: str -class Track(TypedDict): +class AudioTrack(TypedDict): + """Represent an audio item or logical segment in a Discogs tracklist. + + Discogs uses this shape both for independently addressable media tracks and + for pieces rolled into one physical track. Position syntax and surrounding + index structure determine which interpretation applies. + """ + + type_: Literal["track"] position: str - type_: str title: str duration: str - artists: list[Artist] + artists: NotRequired[list[Artist]] extraartists: NotRequired[list[Artist]] - sub_tracks: NotRequired[list[Track]] + + +class IndexTrack(TypedDict): + """Represent a titled musical work containing related subtracks. + + The contained entries may be movements stored as separate physical tracks + or logical divisions of one track. Their position syntax determines whether + they remain separate or are coalesced under the work's metadata. + """ + + type_: Literal["index"] + position: Literal[""] + title: str + duration: str + sub_tracks: list[AudioTrack] + artists: NotRequired[list[Artist]] + extraartists: NotRequired[list[Artist]] + + +class HeadingTrack(TypedDict): + """Represent descriptive tracklist structure rather than an audio item. + + A heading labels the following block but is not the title of a musical work. + It must remain structural instead of being treated as an index container. + """ + + type_: Literal["heading"] + position: Literal[""] + title: str + duration: str + + +Track = AudioTrack | IndexTrack | HeadingTrack class ArtistInfo(TypedDict): diff --git a/test/plugins/factories/discogs.py b/test/plugins/factories/discogs.py new file mode 100644 index 000000000..581e4d886 --- /dev/null +++ b/test/plugins/factories/discogs.py @@ -0,0 +1,23 @@ +import factory + + +class AudioTrackFactory(factory.DictFactory): + type_ = "track" + position = "1" + title = "Track" + duration = "" + + +class IndexTrackFactory(factory.DictFactory): + type_ = "index" + title = "Index Track" + position = "" + duration = "" + sub_tracks = factory.List([factory.SubFactory(AudioTrackFactory)]) + + +class HeadingTrackFactory(factory.DictFactory): + type_ = "heading" + title = "Heading Track" + position = "" + duration = "" diff --git a/test/plugins/test_discogs.py b/test/plugins/test_discogs.py index aa7533885..eb66fe378 100644 --- a/test/plugins/test_discogs.py +++ b/test/plugins/test_discogs.py @@ -1,6 +1,8 @@ """Tests for discogs plugin.""" -from typing import Any +from __future__ import annotations + +from typing import TYPE_CHECKING, Any import pytest from discogs_client import Client, Release @@ -10,6 +12,12 @@ from beets.library import Item from beets.test.helper import TestHelper from beetsplug.discogs import ArtistState, DiscogsPlugin +from .factories import discogs as factories + +if TYPE_CHECKING: + from beetsplug.discogs import types + + _p = pytest.param @@ -25,13 +33,23 @@ def _artist(name: str, **kwargs): } | kwargs -def _track(title: str, position: str = "", duration: str = "", **kwargs): - return { - "title": title, - "position": position, - "duration": duration, - "type_": "track", - } | kwargs +def audio_track( + title: str, position: str = "", duration: str = "", **kwargs +) -> types.AudioTrack: + kwargs.update(title=title, position=position, duration=duration) + return factories.AudioTrackFactory.build(**kwargs) + + +def index_track( + title: str, sub_tracks: list[types.AudioTrack], **kwargs +) -> types.IndexTrack: + kwargs.update(title=title, sub_tracks=sub_tracks) + return factories.IndexTrackFactory.build(**kwargs) + + +def heading_track(title: str, **kwargs) -> types.HeadingTrack: + kwargs.update(title=title) + return factories.HeadingTrackFactory.build(**kwargs) def get_release(data: dict[str, Any]) -> Release: @@ -88,7 +106,7 @@ class DiscogsTestMixin: def _make_release_from_positions(self, positions): """Return discogs_client.Release with tracks at the given positions.""" tracks = [ - _track(f"TITLE{i}", position) + audio_track(f"TITLE{i}", position) for (i, position) in enumerate(positions, start=1) ] return self._make_release(tracks) @@ -97,8 +115,8 @@ class DiscogsTestMixin: class TestDGAlbumInfo(DiscogsTestMixin, TestHelper): def test_parse_media_for_tracks(self): tracks = [ - _track("TITLE ONE", "1", "01:01"), - _track("TITLE TWO", "2", "02:02"), + audio_track("TITLE ONE", "1", "01:01"), + audio_track("TITLE TWO", "2", "02:02"), ] release = self._make_release(tracks=tracks) @@ -184,7 +202,7 @@ class TestDGAlbumInfo(DiscogsTestMixin, TestHelper): data = { "id": 123, "uri": "https://www.discogs.com/release/123456-something", - "tracklist": [_track("A", "1", "01:01")], + "tracklist": [audio_track("A", "1", "01:01")], "artists": [_artist("ARTIST NAME", id=321)], "title": "TITLE", } @@ -300,27 +318,26 @@ class TestTracklist(DiscogsTestMixin): _p( {"index_tracks": False}, [ - _track("MEDIUM TITLE"), - _track("TRACK GROUP TITLE"), - _track("TITLE ONE", "1.1"), - _track("TITLE TWO", "1.2"), + heading_track("MEDIUM TITLE"), + audio_track("TITLE ONE", "1.1"), + audio_track("TITLE TWO", "1.2"), ], 1, - [("TRACK GROUP TITLE", "MEDIUM TITLE")], + [("TITLE ONE / TITLE TWO", "MEDIUM TITLE")], id="flat-logical-subtracks", ), _p( {"index_tracks": False}, [ - _track("TITLE ONE", "1"), - _track( + audio_track("TITLE ONE", "1"), + index_track( "TRACK GROUP TITLE", - sub_tracks=[ - _track("SUBTITLE ONE", "2.1", "01:01"), - _track("SUBTITLE TWO", "2.2", "02:02"), + [ + audio_track("SUBTITLE ONE", "2.1", "01:01"), + audio_track("SUBTITLE TWO", "2.2", "02:02"), ], ), - _track("TITLE THREE", "3"), + audio_track("TITLE THREE", "3"), ], 1, [ @@ -333,15 +350,15 @@ class TestTracklist(DiscogsTestMixin): _p( {"index_tracks": False}, [ - _track("TITLE ONE", "1"), - _track( + audio_track("TITLE ONE", "1"), + index_track( "TRACK GROUP TITLE", - sub_tracks=[ - _track("SUBTITLE ONE", "2", "01:01"), - _track("SUBTITLE TWO", "3", "02:02"), + [ + audio_track("SUBTITLE ONE", "2", "01:01"), + audio_track("SUBTITLE TWO", "3", "02:02"), ], ), - _track("TITLE FOUR", "4"), + audio_track("TITLE FOUR", "4"), ], 1, [ @@ -355,15 +372,15 @@ class TestTracklist(DiscogsTestMixin): _p( {"index_tracks": True}, [ - _track("TITLE ONE", "1"), - _track( + audio_track("TITLE ONE", "1"), + index_track( "TRACK GROUP TITLE", - sub_tracks=[ - _track("SUBTITLE ONE", "2", "01:01"), - _track("SUBTITLE TWO", "3", "02:02"), + [ + audio_track("SUBTITLE ONE", "2", "01:01"), + audio_track("SUBTITLE TWO", "3", "02:02"), ], ), - _track("TITLE FOUR", "4"), + audio_track("TITLE FOUR", "4"), ], 1, [ @@ -377,11 +394,11 @@ class TestTracklist(DiscogsTestMixin): _p( {"index_tracks": False}, [ - _track("MEDIUM TITLE CD1"), - _track("TITLE ONE", "1-1"), - _track("TITLE TWO", "1-2"), - _track("MEDIUM TITLE CD2"), - _track("TITLE THREE", "2-1"), + heading_track("MEDIUM TITLE CD1"), + audio_track("TITLE ONE", "1-1"), + audio_track("TITLE TWO", "1-2"), + heading_track("MEDIUM TITLE CD2"), + audio_track("TITLE THREE", "2-1"), ], 2, [ @@ -414,13 +431,15 @@ class TestTracklist(DiscogsTestMixin): track_artist = "TRACK ARTIST" group_artist = "GROUP ARTIST" tracks = [ - _track( + index_track( "TRACK GROUP TITLE", - artists=[_artist(group_artist)], - sub_tracks=[ - _track("TRACK ONE", "2", artists=[_artist(track_artist)]), - _track("SUBTITLE TWO", "3"), + [ + audio_track( + "TRACK ONE", "2", artists=[_artist(track_artist)] + ), + audio_track("SUBTITLE TWO", "3"), ], + artists=[_artist(group_artist)], ) ] release = self._make_release(tracks) 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 2/2] 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):