discogs: refactor _coalesce_tracks around more specific track types (#6818)

- `beetsplug/discogs/types.py` now models Discogs tracklist entries as
distinct shapes: `AudioTrack`, `IndexTrack`, and `HeadingTrack`. This
shifts the plugin from "guessing by `position`" to handling each entry
by its declared `type_`.

- `beetsplug/discogs/__init__.py` refactors `_coalesce_tracks()` into
smaller helpers for flat subtracks and index tracks. The normalization
flow is now type-driven, which makes the rules for "one physical track"
vs "many physical tracks" more explicit.

- `beetsplug/discogs/states.py` now advances track state only for real
audio entries (`type_ == "track"`), so structural entries no longer
affect track counting through indirect `position` checks.

- Tests now use dedicated Discogs track factories and add coverage for
logical index tracks. This gives the refactor a clearer test model and
protects the new coalescing behavior.

- High-level impact: Discogs imports should now treat headings, index
containers, and subtracks more consistently, especially for releases
where one displayed work contains multiple logical parts but maps to a
single playable track.

### This change is based on the following research (thanks, Codex!):

# Discogs track coalescing

`DiscogsPlugin._coalesce_tracks()` handles three distinct notions all
called
"subtracks." This is the main source of its complexity.

Discogs distinguishes:

- `track`: an audio entry.
- `index`: a titled container whose `sub_tracks` are parts or movements.
- `heading`: descriptive text applying to following tracks.

That distinction is documented in the current
[Discogs tracklisting
guidelines](https://support.discogs.com/hc/en-us/articles/360005055373-Database-Guidelines-12-Tracklisting).
Positions such as `3.1`, `3.2` or `3a`, `3b` separately indicate
multiple
musical pieces inside one physical track.

## Observed input shapes

| Case | API shape | Current result | Example |
|---|---|---|---|
| Normal track | `type_="track"`, nonempty position | Passed through |
Most releases |
| Flat virtual subtracks | Multiple top-level `track` entries with
`1.1`, `1.2` or `22a`, `22b` | Merged into a copy of the first track |
[This Is Not](https://www.discogs.com/release/7117597), [Eclectic Beatz
8](https://www.discogs.com/release/2885582) |
| Nested logical subtracks | `index` with nested children sharing one
physical position | Parent index becomes the physical track; children
discarded | [Defqon.1 Weekend
Festival](https://www.discogs.com/release/7168134) |
| Nested physical subtracks | `index` with children numbered as
independent tracks | Index discarded; children promoted | [Classical
Hollywood III](https://www.discogs.com/release/3647530) |
| Heading | `heading`, empty position, no `sub_tracks` | Passed through
as structural metadata | Defqon.1 CD headings |
| Ambiguous/nonstandard index | `index` with positions the regex cannot
classify | Usually promoted incorrectly as separate tracks | [King
Crimson - Lizard](https://www.discogs.com/release/1156598) |

### Flat virtual subtracks

`This Is Not` returns nine top-level entries:

```python
{"type_": "track", "position": "1.1", "title": "Intro", "duration": "1:49"}
{"type_": "track", "position": "1.2", "title": "Untitled", "duration": "6:31"}
# ...
{"type_": "track", "position": "1.9", "title": "2 Bad ...", "duration": "10:18"}
```

There is no index container. `_coalesce_tracks()` detects the suffixes
and
`_add_merged_subtracks()` produces:

```python
{
    "type_": "track",
    "position": "1.1",
    "title": "Intro / Untitled / ... / 2 Bad ...",
    "duration": "1:49",
    "artists": ["Unknown Artist"],
}
```

It copies all metadata from the first entry and only combines titles.
This
explains the incorrect duration and artist.

`Eclectic Beatz 8` has the same flat shape:

```python
{"type_": "track", "position": "22a", "title": "Amplifier", "artists": ["Fedde Le Grand"]}
{"type_": "track", "position": "22b", "title": "Autograph ...", "artists": ["Hatiras", "MC Flipside"]}
```

Those become one track, retaining only the first artist.

### Nested logical subtracks

Defqon.1 contains:

```python
{
    "type_": "index",
    "position": "",
    "title": "Eat This / God! What The Hell!",
    "duration": "3:06",
    "sub_tracks": [
        {"type_": "track", "position": "2-20A", "title": "Eat This"},
        {"type_": "track", "position": "2-20B", "title": "God! What The Hell!"},
    ],
}
```

Because `2-20A` has a parsed subindex, the helper changes the parent to:

```python
{
    "type_": "index",
    "position": "2-20",
    "title": "Eat This / God! What The Hell!",
    "duration": "3:06",
}
```

The children disappear. Downstream code treats this as audio solely
because
its position is now nonempty. It never checks that `type_` remains
`"index"`.

### Nested physical subtracks

`Classical Hollywood III` contains index containers such as:

```python
{
    "type_": "index",
    "position": "",
    "title": "Auld Lang Syne Variations For Piano Quartet",
    "sub_tracks": [
        {"type_": "track", "position": "1", "title": "Eine Kleine Nichtmusik ..."},
        {"type_": "track", "position": "2", "title": "L.V.B. - Adagio"},
        {"type_": "track", "position": "3", "title": "Chaconne ..."},
        {"type_": "track", "position": "4", "title": "Homage ..."},
    ],
}
```

Because position `1` has no subindex:

- the index container is removed;
- its children become top-level physical tracks;
- parent artists are copied to children missing artists;
- when `index_tracks` is enabled, the parent title prefixes every child
title.

This behavior was added for
[issue #2318](https://github.com/beetbox/beets/issues/2318) by
[PR #2355](https://github.com/beetbox/beets/pull/2355).

### Headings

Defqon.1 also contains:

```python
{
    "type_": "heading",
    "position": "",
    "title": "CD2  Mixed By Partyraiser",
    "duration": "",
}
```

This is not coalesced. It passes through, but `TracklistState`
subsequently
treats every positionless object as structural metadata without checking
`type_`.

The code therefore cannot distinguish:

```python
position == "" and type_ == "heading"
position == "" and type_ == "index"
```

Worse, `_add_merged_subtracks()` identifies a parent using only:

```python
tracklist and not tracklist[-1]["position"]
```

A heading can therefore theoretically be mistaken for an index parent.

## History

- [Issue #1543](https://github.com/beetbox/beets/issues/1543) reported
flat
  positions such as `22a` and `22b` representing one CD track.
- [PR #2222](https://github.com/beetbox/beets/pull/2222) added both flat
and
nested subtrack support. Its author explicitly noted that `type_` might
  identify index tracks robustly, but avoided the larger refactor.
- [Issue #2318](https://github.com/beetbox/beets/issues/2318)
demonstrated that
  nested `sub_tracks` can instead be independent physical tracks.
- PR #2355 added the current heuristic:
  - parsed child subindex means logical fragments of one physical track;
  - no parsed child subindex means independent physical tracks.
- That PR documented the heuristic as imperfect. `Lizard`, with `Ba`,
`Bb`,
  `Bc`, and related positions, was already a known failure.

## Test data does not match the API shapes

The current test helper always creates:

```python
"type_": "track"
```

Consequently, test inputs such as:

```python
_track("TRACK GROUP TITLE", sub_tracks=[...])
```

are not realistic. Real API data uses:

```python
"type_": "index"
```

Likewise, test medium titles are manufactured as positionless `track`
objects
rather than `heading` or `index` objects.

## Recommended direction

Before refactoring production code, introduce accurate input types and
factories:

```python
TrackEntry = AudioTrack | IndexTrack | HeadingTrack
```

With approximate shapes:

```python
class AudioTrack(TypedDict):
    type_: Literal["track"]
    position: str
    title: str
    duration: str
    artists: NotRequired[list[Artist]]


class IndexTrack(TypedDict):
    type_: Literal["index"]
    position: Literal[""]
    title: str
    duration: str
    sub_tracks: list[AudioTrack]
    artists: NotRequired[list[Artist]]


class HeadingTrack(TypedDict):
    type_: Literal["heading"]
    position: Literal[""]
    title: str
    duration: str
```

Then test using `_track()`, `_index()`, and `_heading()` factories.

The eventual normalization can dispatch structurally:

1. `track` without subindex: pass through.
2. Flat subindexed `track` group: apply the flat-fragment policy.
3. `index`: normalize its children as logical or physical.
4. `heading`: preserve as structural metadata.
5. Never infer an index merely from `position == ""`.

This removes the most dangerous ambiguity while preserving the
positional
heuristic where Discogs genuinely requires it.
This commit is contained in:
Šarūnas Nejus
2026-07-16 12:02:00 +01:00
committed by GitHub
5 changed files with 221 additions and 128 deletions

View File

@@ -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,

View File

@@ -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

View File

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

View File

@@ -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 = ""

View File

@@ -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)
@@ -429,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):