Repository: beetbox/beets (python)

Issue:
listenbrainz: the plugin appears to be broken

<!--
Describe your problem, feature request, or discussion topic here.

Please fill out this and the "Setup" section below and remember to include
enough detail so that other people can reproduce the problem.
-->

### Problem

Running this command:

```sh
$ beet lbimport
```

Led to this problem:

```
Traceback (most recent call last):
  File "/home/konstantink/.local/bin/beet", line 10, in <module>
    sys.exit(main())
             ~~~~^^
  File "/home/konstantink/.local/share/uv/tools/beets/lib/python3.13/site-packages/beets/ui/__init__.py", line 1013, in main
    _raw_main(args)
    ~~~~~~~~~^^^^^^
  File "/home/konstantink/.local/share/uv/tools/beets/lib/python3.13/site-packages/beets/ui/__init__.py", line 992, in _raw_main
    subcommand.func(lib, suboptions, subargs)
    ~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/konstantink/.local/share/uv/tools/beets/lib/python3.13/site-packages/beetsplug/listenbrainz.py", line 34, in func
    self._lbupdate(lib, self._log)
    ~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^
  File "/home/konstantink/.local/share/uv/tools/beets/lib/python3.13/site-packages/beetsplug/listenbrainz.py", line 47, in _lbupdate
    found, unknown = process_tracks(lib, tracks, log)
                     ~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^
  File "/home/konstantink/.local/share/uv/tools/beets/lib/python3.13/site-packages/beetsplug/lastimport.py", line 269, in process_tracks
    new_count = int(tracks[num]["playcount"])
                    ~~~~~~~~~~~^^^^^^^^^^^^^
KeyError: 'playcount'
```

This specific crash seems to have been caused by 6eff9712574d6a736f4840041bfb72d557008c41. Using .get silences the error, but listenbrainz probably shouldn't call a function from lastimport that's now hardcoded to write to lastfm_play_count. Also, when I run lbimport, it only seems to import 25 listens, even though I have way more than that. 

### Setup

* OS: Fedora Linux 43 (Workstation Edition)
* Python version: 3.13.12
* beets version: 2.8.0
* Turning off plugins made problem go away (yes/no): no

<!--
You can turn off plugins temporarily by passing --plugins= on the command line:

$ beet --plugins= version
-->



Patch:
diff --git a/beetsplug/_utils/playcount.py b/beetsplug/_utils/playcount.py
new file mode 100644
index 0000000000..4595027c67
--- /dev/null
+++ b/beetsplug/_utils/playcount.py
@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, TypedDict
+
+from typing_extensions import NotRequired
+
+from beets.dbcore.query import AndQuery, MatchQuery, OrQuery, SubstringQuery
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+    from beets.dbcore.query import Query
+    from beets.library import Item, Library
+    from beets.logging import BeetsLogger
+
+
+class Track(TypedDict):
+    mbid: str | None
+    name: str
+    artist: str
+    playcount: int
+    album: NotRequired[str | None]
+
+
+def get_items(lib: Library, track: Track, log: BeetsLogger) -> Sequence[Item]:
+    mbid, artist, title = track["mbid"], track["artist"], track["name"]
+    album = track.get("album") or ""
+
+    log.debug("query: {} - {} ({})", artist, title, album)
+
+    title_query = OrQuery(
+        [
+            SubstringQuery("title", title),
+            # try a right single quotation mark instead of an apostrophe
+            SubstringQuery("title", title.replace("'", "\u2019")),
+        ]
+    )
+    or_queries: list[Query] = [
+        AndQuery([SubstringQuery("artist", artist), title_query])
+    ]
+    # First try to query by musicbrainz's trackid
+    if mbid:
+        or_queries.append(MatchQuery("mb_trackid", mbid))
+    if album:
+        or_queries.append(
+            AndQuery([SubstringQuery("album", album), title_query])
+        )
+
+    return list(lib.items(OrQuery(or_queries)))
+
+
+def process_track(
+    lib: Library, track: Track, log: BeetsLogger, source: str
+) -> bool:
+    items = get_items(lib, track, log)
+    if not items:
+        return False
+
+    new_count = track["playcount"]
+    field = f"{source}_play_count"
+    for song in items:
+        count = int(song.get(field, 0))
+        log.debug(
+            "match: {0.artist} - {0.title} ({0.album}) updating:"
+            " {1} {2} => {3}",
+            song,
+            field,
+            count,
+            new_count,
+        )
+        song[field] = new_count
+        song.store()
+
+    return True
+
+
+def update_play_counts(
+    lib: Library, tracks: Sequence[Track], log: BeetsLogger, source: str
+) -> tuple[int, int]:
+    total = len(tracks)
+    total_found = 0
+    total_fails = 0
+    log.info("Received {} tracks in this page, processing...", total)
+
+    for track in tracks:
+        if process_track(lib, track, log, source):
+            total_found += 1
+        else:
+            total_fails += 1
+
+    if total_fails > 0:
+        log.info(
+            "Acquired {}/{} play-counts ({} unknown)",
+            total_found,
+            total,
+            total_fails,
+        )
+
+    return total_found, total_fails
diff --git a/beetsplug/lastimport.py b/beetsplug/lastimport.py
index 1b0d339f96..15a94c312a 100644
--- a/beetsplug/lastimport.py
+++ b/beetsplug/lastimport.py
@@ -12,13 +12,21 @@
 # The above copyright notice and this permission notice shall be
 # included in all copies or substantial portions of the Software.
 
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
 
 import pylast
 from pylast import TopItem, _extract, _number
 
-from beets import config, dbcore, plugins, ui
+from beets import config, plugins, ui
 from beets.dbcore import types
 
+from ._utils.playcount import update_play_counts
+
+if TYPE_CHECKING:
+    from ._utils.playcount import Track
+
 API_URL = "https://ws.audioscrobbler.com/2.0/"
 
 
@@ -142,7 +150,7 @@ def import_lastfm(lib, log):
                 raise ui.UserError("Last.fm reported no data.")
 
             if tracks:
-                found, unknown = process_tracks(lib, tracks, log)
+                found, unknown = update_play_counts(lib, tracks, log, "lastfm")
                 found_total += found
                 unknown_total += unknown
                 break
@@ -170,123 +178,18 @@ def import_lastfm(lib, log):
     log.info("{} play-counts imported", found_total)
 
 
-def fetch_tracks(user, page, limit):
-    """JSON format:
-    [
-        {
-            "mbid": "...",
-            "artist": "...",
-            "title": "...",
-            "playcount": "..."
-        }
-    ]
-    """
-    network = pylast.LastFMNetwork(api_key=config["lastfm"]["api_key"])
+def fetch_tracks(user, page, limit) -> tuple[list[Track], int]:
+    network = pylast.LastFMNetwork(api_key=config["lastfm"]["api_key"].get(str))
     user_obj = CustomUser(user, network)
     results, total_pages = user_obj.get_top_tracks_by_page(
         limit=limit, page=page
     )
     return [
         {
-            "mbid": track.item.mbid if track.item.mbid else "",
-            "artist": {"name": track.item.artist.name},
-            "name": track.item.title,
-            "playcount": track.weight,
+            "mbid": track.item.mbid or "",
+            "artist": track.item.artist.name.strip(),
+            "name": track.item.title.strip(),
+            "playcount": int(track.weight),
         }
         for track in results
     ], total_pages
-
-
-def process_tracks(lib, tracks, log):
-    total = len(tracks)
-    total_found = 0
-    total_fails = 0
-    log.info("Received {} tracks in this page, processing...", total)
-
-    for num in range(0, total):
-        song = None
-        trackid = tracks[num]["mbid"].strip() if tracks[num]["mbid"] else None
-        artist = (
-            tracks[num]["artist"].get("name", "").strip()
-            if tracks[num]["artist"].get("name", "")
-            else None
-        )
-        title = tracks[num]["name"].strip() if tracks[num]["name"] else None
-        album = ""
-        if "album" in tracks[num]:
-            album = (
-                tracks[num]["album"].get("name", "").strip()
-                if tracks[num]["album"]
-                else None
-            )
-
-        log.debug("query: {} - {} ({})", artist, title, album)
-
-        # First try to query by musicbrainz's trackid
-        if trackid:
-            song = lib.items(
-                dbcore.query.MatchQuery("mb_trackid", trackid)
-            ).get()
-
-        # If not, try just album/title
-        if song is None:
-            log.debug(
-                "no album match, trying by album/title: {} - {}", album, title
-            )
-            query = dbcore.AndQuery(
-                [
-                    dbcore.query.SubstringQuery("album", album),
-                    dbcore.query.SubstringQuery("title", title),
-                ]
-            )
-            song = lib.items(query).get()
-
-        # If not, try just artist/title
-        if song is None:
-            log.debug("no album match, trying by artist/title")
-            query = dbcore.AndQuery(
-                [
-                    dbcore.query.SubstringQuery("artist", artist),
-                    dbcore.query.SubstringQuery("title", title),
-                ]
-            )
-            song = lib.items(query).get()
-
-        # Last resort, try just replacing to utf-8 quote
-        if song is None:
-            title = title.replace("'", "\u2019")
-            log.debug("no title match, trying utf-8 single quote")
-            query = dbcore.AndQuery(
-                [
-                    dbcore.query.SubstringQuery("artist", artist),
-                    dbcore.query.SubstringQuery("title", title),
-                ]
-            )
-            song = lib.items(query).get()
-
-        if song is not None:
-            count = int(song.get("lastfm_play_count", 0))
-            new_count = int(tracks[num]["playcount"])
-            log.debug(
-                "match: {0.artist} - {0.title} ({0.album}) updating:"
-                " lastfm_play_count {1} => {2}",
-                song,
-                count,
-                new_count,
-            )
-            song["lastfm_play_count"] = new_count
-            song.store()
-            total_found += 1
-        else:
-            total_fails += 1
-            log.info("  - No match: {} - {} ({})", artist, title, album)
-
-    if total_fails > 0:
-        log.info(
-            "Acquired {}/{} play-counts ({} unknown)",
-            total_found,
-            total,
-            total_fails,
-        )
-
-    return total_found, total_fails
diff --git a/beetsplug/listenbrainz.py b/beetsplug/listenbrainz.py
index 6730ce7160..39f51c0d8e 100644
--- a/beetsplug/listenbrainz.py
+++ b/beetsplug/listenbrainz.py
@@ -1,14 +1,21 @@
 """Adds Listenbrainz support to Beets."""
 
+from __future__ import annotations
+
 import datetime
+from typing import TYPE_CHECKING, ClassVar
 
 import requests
 
 from beets import config, ui
+from beets.dbcore import types
 from beets.plugins import BeetsPlugin
-from beetsplug.lastimport import process_tracks
 
 from ._utils.musicbrainz import MusicBrainzAPIMixin
+from ._utils.playcount import update_play_counts
+
+if TYPE_CHECKING:
+    from ._utils.playcount import Track
 
 
 class ListenBrainzPlugin(MusicBrainzAPIMixin, BeetsPlugin):
@@ -16,6 +23,10 @@ class ListenBrainzPlugin(MusicBrainzAPIMixin, BeetsPlugin):
 
     ROOT = "http://api.listenbrainz.org/1/"
 
+    item_types: ClassVar[dict[str, types.Type]] = {
+        "listenbrainz_play_count": types.INTEGER
+    }
+
     def __init__(self):
         """Initialize the plugin."""
         super().__init__()
@@ -44,7 +55,9 @@ def _lbupdate(self, lib, log):
         tracks = self.get_tracks_from_listens(ls)
         log.info("Found {} listens", len(ls))
         if tracks:
-            found, unknown = process_tracks(lib, tracks, log)
+            found, unknown = update_play_counts(
+                lib, tracks, log, "listenbrainz"
+            )
             found_total += found
             unknown_total += unknown
         log.info("... done!")
@@ -103,9 +116,9 @@ def get_listens(self, min_ts=None, max_ts=None, count=None):
         else:
             return None
 
-    def get_tracks_from_listens(self, listens):
+    def get_tracks_from_listens(self, listens) -> list[Track]:
         """Returns a list of tracks from a list of listens."""
-        tracks = []
+        tracks: list[Track] = []
         for track in listens:
             if track["track_metadata"].get("release_name") is None:
                 continue
@@ -116,16 +129,17 @@ def get_tracks_from_listens(self, listens):
                 mbid = self.get_mb_recording_id(track)
             tracks.append(
                 {
-                    "album": {
-                        "name": track["track_metadata"].get("release_name")
-                    },
-                    "name": track["track_metadata"].get("track_name"),
-                    "artist": {
-                        "name": track["track_metadata"].get("artist_name")
-                    },
+                    "album": (
+                        track["track_metadata"].get("release_name") or ""
+                    ).strip(),
+                    "name": (
+                        track["track_metadata"].get("track_name") or ""
+                    ).strip(),
+                    "artist": (
+                        track["track_metadata"].get("artist_name") or ""
+                    ).strip(),
                     "mbid": mbid,
-                    "release_mbid": mbid_mapping.get("release_mbid"),
-                    "listened_at": track.get("listened_at"),
+                    "playcount": 1,
                 }
             )
         return tracks
diff --git a/docs/changelog.rst b/docs/changelog.rst
index c8d5c0edd0..2536d8b837 100644
--- a/docs/changelog.rst
+++ b/docs/changelog.rst
@@ -21,6 +21,11 @@ Bug fixes
 
 - Correctly handle semicolon-delimited genre values from externally-tagged
   files. :bug:`6450`
+- :doc:`plugins/listenbrainz`: Fix ``lbimport`` crashing when ListenBrainz
+  tracks are processed through Last.fm-specific play-count import logic.
+  Play-count imports now use source-specific fields so
+  :doc:`plugins/listenbrainz`, :doc:`plugins/lastimport`, and
+  :doc:`plugins/mpdstats` do not clash. :bug:`6469`
 
 ..
     For plugin developers
diff --git a/docs/plugins/listenbrainz.rst b/docs/plugins/listenbrainz.rst
index ceff0e8002..cb45504e6f 100644
--- a/docs/plugins/listenbrainz.rst
+++ b/docs/plugins/listenbrainz.rst
@@ -32,6 +32,17 @@ Usage
 Once the plugin is enabled, you can import the listening history using the
 ``lbimport`` command in beets.
 
+Matched tracks are populated with the ``listenbrainz_play_count`` field, which
+you can use in queries and templates. For example:
+
+::
+
+    $ beet ls -f '$title: $listenbrainz_play_count' listenbrainz_play_count:1..
+
+This field is separate from ``lastfm_play_count`` used by :doc:`lastimport` and
+``play_count`` used by :doc:`mpdstats`, so importing play counts from one plugin
+does not overwrite the others.
+
 .. _config.yaml: ../reference/config.rst
 
 .. _here: https://listenbrainz.readthedocs.io/en/latest/users/api/index.html#get-the-user-token


Test patch:
diff --git a/test/plugins/utils/test_playcount.py b/test/plugins/utils/test_playcount.py
new file mode 100644
index 0000000000..c434bfa4cf
--- /dev/null
+++ b/test/plugins/utils/test_playcount.py
@@ -0,0 +1,226 @@
+import pytest
+
+from beets import logging
+from beets.library import Item
+from beets.test.helper import TestHelper, capture_log
+from beetsplug._utils.playcount import (
+    get_items,
+    process_track,
+    update_play_counts,
+)
+
+LOGGER_NAME = "beets.test_playcount"
+
+
+@pytest.fixture
+def helper(request):
+    helper = TestHelper()
+    helper.setup_beets()
+
+    request.instance.lib = helper.lib
+    request.instance.log = logging.getLogger(LOGGER_NAME)
+    request.instance.log.set_global_level(logging.DEBUG)
+
+    yield helper
+
+    helper.teardown_beets()
+
+
+@pytest.mark.usefixtures("helper")
+class TestPlayCount:
+    def track(self, **overrides):
+        return {
+            "mbid": "",
+            "artist": "Artist",
+            "name": "Song",
+            "playcount": 1,
+            **overrides,
+        }
+
+    def get_playcount(self, item_id, source="lastfm"):
+        field = f"{source}_play_count"
+        return int(self.lib.get_item(item_id).get(field, 0))
+
+    def add_item(
+        self,
+        *,
+        title="Song",
+        artist="Artist",
+        album="Album",
+        mb_trackid="",
+        play_count=None,
+        source="lastfm",
+    ):
+        item = Item(
+            title=title,
+            artist=artist,
+            album=album,
+            mb_trackid=mb_trackid,
+        )
+        self.lib.add(item)
+
+        if play_count is not None:
+            item[f"{source}_play_count"] = play_count
+            item.store()
+
+        return item
+
+    @pytest.mark.parametrize(
+        "item_kwargs, track_kwargs",
+        [
+            pytest.param(
+                {"title": "Song", "artist": "Artist"},
+                {"playcount": 7},
+                id="artist-and-title",
+            ),
+            pytest.param(
+                {
+                    "title": "Different Song",
+                    "artist": "Different Artist",
+                    "mb_trackid": "track-id",
+                },
+                {"mbid": "track-id", "playcount": 5},
+                id="musicbrainz-track-id",
+            ),
+            pytest.param(
+                {
+                    "title": "Song",
+                    "artist": "Different Artist",
+                    "album": "Album",
+                },
+                {"album": "Album", "playcount": 3},
+                id="album-and-title",
+            ),
+            pytest.param(
+                {"title": "Don\u2019t Stop", "artist": "Artist"},
+                {"name": "Don't Stop", "playcount": 11},
+                id="apostrophe-normalized",
+            ),
+        ],
+    )
+    def test_get_items_matches_supported_query_paths(
+        self, item_kwargs, track_kwargs
+    ):
+        item = self.add_item(**item_kwargs)
+        matched_ids = [
+            matched.id
+            for matched in get_items(
+                self.lib, self.track(**track_kwargs), self.log
+            )
+        ]
+
+        assert matched_ids == [item.id]
+
+    def test_process_track_updates_every_matching_song(self):
+        first = self.add_item(
+            title="Song",
+            artist="Artist",
+            album="First Album",
+            play_count=1,
+        )
+        second = self.add_item(
+            title="Song",
+            artist="Artist",
+            album="Second Album",
+            play_count=9,
+        )
+
+        assert (
+            process_track(
+                self.lib,
+                self.track(playcount=0),
+                self.log,
+                "lastfm",
+            )
+            is True
+        )
+
+        assert self.get_playcount(first.id) == 0
+        assert self.get_playcount(second.id) == 0
+
+    def test_process_track_returns_false_when_nothing_matches(self):
+        assert (
+            process_track(
+                self.lib,
+                self.track(
+                    artist="Missing Artist",
+                    name="Missing Song",
+                    album="Missing Album",
+                    playcount=4,
+                ),
+                self.log,
+                "lastfm",
+            )
+            is False
+        )
+
+    def test_process_track_updates_requested_source_field(self):
+        new_count = 6
+        item = self.add_item(play_count=1, source="lastfm")
+
+        assert (
+            process_track(
+                self.lib,
+                self.track(playcount=new_count),
+                self.log,
+                "listenbrainz",
+            )
+            is True
+        )
+
+        assert self.get_playcount(item.id, "lastfm") == 1
+        assert self.get_playcount(item.id, "listenbrainz") == new_count
+
+    @pytest.mark.parametrize(
+        "tracks, expected_counts, expected_summary, expected_playcount",
+        [
+            pytest.param([], (0, 0), None, None, id="empty-page"),
+            pytest.param(
+                [
+                    {
+                        "artist": "Artist",
+                        "name": "Known Song",
+                        "playcount": 8,
+                    },
+                    {
+                        "artist": "Missing Artist",
+                        "name": "Missing Song",
+                        "playcount": 2,
+                    },
+                ],
+                (1, 1),
+                "Acquired 1/2 play-counts (1 unknown)",
+                8,
+                id="mixed-results",
+            ),
+        ],
+    )
+    def test_update_play_counts_counts_and_logs_summary(
+        self, tracks, expected_counts, expected_summary, expected_playcount
+    ):
+        item = self.add_item(
+            title="Known Song",
+            artist="Artist",
+            play_count=1,
+        )
+
+        with capture_log(LOGGER_NAME) as logs:
+            assert (
+                update_play_counts(
+                    self.lib,
+                    [self.track(**track) for track in tracks],
+                    self.log,
+                    "lastfm",
+                )
+                == expected_counts
+            )
+
+        assert any(
+            f"Received {len(tracks)} tracks in this page, processing..." in log
+            for log in logs
+        )
+        if expected_summary is None:
+            assert not any("Acquired" in log for log in logs)
+        else:
+            assert expected_summary in logs
+            assert self.get_playcount(item.id) == expected_playcount