mirrors/beets
2
0
Fork 0
mirror of https://github.com/beetbox/beets.git synced 2026-07-22 02:06:50 -04:00
Code Issues Packages Projects Releases Wiki Activity

tidal: add flexattrs and tidalsync command

Browse Source 
- Add item_types ClassVar with tidal_track_id, tidal_album_id,
  tidal_artist_id, tidal_track_popularity, tidal_alb_popularity,
  tidal_updated fields
- Populate flexattrs during album/track import via AlbumInfo/TrackInfo
  kwargs
- Add beet tidalsync command to refresh popularity data post-import
- Add tidal fields to REIMPORT_FRESH_FIELDS_ITEM for reimport support
- Add tests for flexattr population and tidalsync behavior
- Update tidal plugin docs with attribute reference and tidalsync usage
This commit is contained in:
Alok Saboo
2026-06-14 13:44:16 -04:00
parent 60047dfabd
commit 98681406af
5 changed files with 360 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
beets/importer/tasks.py
View File

@@ -64,6 +64,11 @@ REIMPORT_FRESH_FIELDS_ITEM = [
"deezer_album_id",
"beatport_album_id",
"tidal_album_id",
"tidal_track_id",
"tidal_artist_id",
"tidal_track_popularity",
"tidal_alb_popularity",
"tidal_updated",
"data_url",
]
REIMPORT_FRESH_FIELDS_ALBUM = [*REIMPORT_FRESH_FIELDS_ITEM, "media"]

144
beetsplug/tidal/__init__.py
View File

@@ -3,13 +3,15 @@ from __future__ import annotations
import itertools
import os
import re
import time
from functools import cached_property
from typing import TYPE_CHECKING, overload
from typing import TYPE_CHECKING, ClassVar, overload
import confuse
from beets import ui
from beets.autotag import AlbumInfo, TrackInfo
from beets.dbcore import types
from beets.exceptions import UserError
from beets.logging import getLogger
from beets.metadata_plugins import MetadataSourcePlugin
@@ -36,6 +38,15 @@ log = getLogger("beets.tidal")
class TidalPlugin(MetadataSourcePlugin):
item_types: ClassVar[dict[str, types.Type]] = {
"tidal_track_id": types.INTEGER,
"tidal_album_id": types.INTEGER,
"tidal_artist_id": types.INTEGER,
"tidal_track_popularity": types.INTEGER,
"tidal_alb_popularity": types.INTEGER,
"tidal_updated": types.DATE,
}
def __init__(self) -> None:
super().__init__()
@@ -66,28 +77,6 @@ class TidalPlugin(MetadataSourcePlugin):
" using `beet tidal --auth` or disable tidal plugin"
)
def commands(self) -> list[ui.Subcommand]:
tidal_cmd = ui.Subcommand(
"tidal", help="Tidal metadata plugin commands"
)
tidal_cmd.parser.add_option(
"-a",
"--auth",
action="store_true",
help="Authenticate and login to Tidal",
default=False,
)
def func(lib: Library, opts: optparse.Values, args: list[str]):
if opts.auth:
self.api.ui_authenticate_flow()
else:
tidal_cmd.print_help()
tidal_cmd.func = func
return [tidal_cmd]
def album_for_id(self, album_id: str) -> AlbumInfo | None:
if not (tidal_id := self._extract_id(album_id)):
return None
@@ -345,7 +334,6 @@ class TidalPlugin(MetadataSourcePlugin):
track_by_id: dict[str, TidalTrack],
artist_by_id: dict[str, TidalArtist],
) -> AlbumInfo:
track_infos: list[TrackInfo] = []
for i, track_rel in enumerate(
album["relationships"]["items"]["data"], start=1
@@ -377,6 +365,13 @@ class TidalPlugin(MetadataSourcePlugin):
year=date_parts[0] if date_parts else None,
month=date_parts[1] if date_parts else None,
day=date_parts[2] if date_parts else None,
# Flexattrs
tidal_album_id=int(album["id"]),
tidal_artist_id=int(artist_ids[0]) if artist_ids else None,
tidal_alb_popularity=self._popularity(
album["attributes"]["popularity"]
),
tidal_updated=time.time(),
)
def _get_track_info(
@@ -399,6 +394,13 @@ class TidalPlugin(MetadataSourcePlugin):
artists=artist_names,
duration=self._duration_to_seconds(track["attributes"]["duration"]),
label=self._parse_label(track["attributes"]),
# Flexattrs
tidal_track_id=int(track["id"]),
tidal_artist_id=int(artist_ids[0]) if artist_ids else None,
tidal_track_popularity=self._popularity(
track["attributes"]["popularity"]
),
tidal_updated=time.time(),
)
@staticmethod
@@ -474,6 +476,100 @@ class TidalPlugin(MetadataSourcePlugin):
return int(parts[0]), int(parts[1]), int(parts[2])
return None
@staticmethod
def _popularity(val: object) -> int | None:
if val is None:
return None
if isinstance(val, float):
return int(val * 100)
return int(val)
def tidalsync(
self, items: Iterable[Item], write: bool, force: bool = False
) -> None:
"""Sync Tidal popularity data for library items."""
items = list(items)
self._log.info("Syncing popularity for {0} tracks", len(items))
for item in items:
if not force and item.get("tidal_track_popularity") is not None:
continue
tidal_track_id = item.get("tidal_track_id")
if tidal_track_id is None:
continue
results = list(
self.search_tracks_by_ids(tidal_ids=[str(tidal_track_id)])
)
if not results or results[0] is None:
self._log.debug("Track {0} not found in Tidal", tidal_track_id)
continue
track_info = results[0]
if track_info.tidal_track_popularity is not None:
item["tidal_track_popularity"] = (
track_info.tidal_track_popularity
)
item["tidal_updated"] = time.time()
item.store()
if write:
item.try_write()
self._log.debug(
"Updated popularity for track {0}: {1}",
tidal_track_id,
track_info.tidal_track_popularity,
)
def commands(self) -> list[ui.Subcommand]:
tidal_cmd = ui.Subcommand(
"tidal", help="Tidal metadata plugin commands"
)
tidal_cmd.parser.add_option(
"-a",
"--auth",
action="store_true",
help="Authenticate and login to Tidal",
default=False,
)
def func(lib: Library, opts: optparse.Values, args: list[str]):
if opts.auth:
self.api.ui_authenticate_flow()
else:
tidal_cmd.print_help()
tidal_cmd.func = func
tidalsync_cmd = ui.Subcommand(
"tidalsync", help="sync Tidal popularity data for library items"
)
tidalsync_cmd.parser.add_option(
"-f",
"--force",
action="store_true",
dest="force",
default=False,
help="re-fetch popularity even if already present",
)
tidalsync_cmd.parser.add_option(
"-w",
"--write",
action="store_true",
dest="write",
default=False,
help="write updated tags to media files",
)
def sync_func(lib: Library, opts: optparse.Values, args: list[str]):
items = lib.items(args)
self.tidalsync(items, write=opts.write, force=opts.force)
tidalsync_cmd.func = sync_func
return [tidal_cmd, tidalsync_cmd]
ISO_8601_RE = re.compile(
r"^P"

5
docs/changelog.rst
View File

@@ -27,6 +27,11 @@ New features
:conf:`plugins.musicbrainz:aliases_as_credits` to make
aliases-as-artist-credit optional.
- :doc:`plugins/badfiles`: Added settings for auto error and warning actions.
- :doc:`plugins/tidal`: New ``item_types`` flexible attributes
``tidal_track_id``, ``tidal_album_id``, ``tidal_artist_id``,
``tidal_track_popularity``, ``tidal_alb_popularity``, and ``tidal_updated``
are now populated during import. Added new ``beet tidalsync`` command to sync
popularity data for imported items.
Bug fixes
~~~~~~~~~

72
docs/plugins/tidal.rst
View File

@@ -102,4 +102,76 @@ Default
The path to the file where the Tidal authentication token is stored.
Flexible Attributes
-------------------
The plugin stores the following flexible attributes on your items during import.
You can use them in queries and path formats.
.. list-table::
:header-rows: 1
- - Attribute
- Type
- Description
- - ``tidal_track_id``
- INTEGER
- Tidal track ID
- - ``tidal_album_id``
- INTEGER
- Tidal album ID
- - ``tidal_artist_id``
- INTEGER
- Tidal artist ID
- - ``tidal_track_popularity``
- INTEGER
- Track popularity score (0-100)
- - ``tidal_alb_popularity``
- INTEGER
- Album popularity score (0-100)
- - ``tidal_updated``
- DATE
- Unix timestamp of last popularity sync
For example, to list your most popular tracks:
::
beet ls tidal_track_popularity:80..
To find tracks that have not been synced yet:
::
beet ls tidal_track_popularity:0
tidalsync Command
-----------------
After importing tracks, you can refresh their popularity data by running the
``tidalsync`` subcommand:
.. code-block:: bash
beet tidalsync
By default, ``tidalsync`` skips items that already have popularity data. Use
``--force`` (``-f``) to re-fetch all items:
::
beet tidalsync -f
Use ``--write`` (``-w``) to also write the updated metadata to the media files:
::
beet tidalsync -w
You can also filter which items to sync by passing a query:
::
beet tidalsync artist:"My Artist"
.. include:: ./shared_metadata_source_config.rst

195
test/plugins/test_tidal.py
View File

@@ -3,6 +3,7 @@
from __future__ import annotations
import logging
import time
from typing import TYPE_CHECKING
from unittest.mock import Mock
@@ -123,25 +124,30 @@ class TestAlbumParsing(TidalPluginTest):
"""High-level tests for album parsing."""
def test_parse_album(self):
track = _make_track("t1", "My Song", "PT3M30S", "ISRC001", ["a1"])
track = _make_track("101", "My Song", "PT3M30S", "ISRC001", ["1001"])
album, track_lookup, artist_lookup = _make_album(
"al1", "My Album", [track], ["a1"]
"1", "My Album", [track], ["1001"]
)
info = self.tidal._get_album_info(album, track_lookup, artist_lookup)
assert info.album == "My Album"
assert info.album_id == "al1"
assert info.album_id == "1"
assert len(info.tracks) == 1
assert info.tracks[0].title == "My Song"
assert info.tidal_album_id == 1
assert info.tidal_artist_id == 1001
assert info.tidal_alb_popularity == 50
assert info.tracks[0].tidal_track_id == 101
assert info.tracks[0].tidal_track_popularity == 50
def test_parse_album_with_multiple_tracks(self):
tracks = [
_make_track("t1", "Track One", "PT3M", "ISRC1", ["a1"]),
_make_track("t2", "Track Two", "PT4M", "ISRC2", ["a1"]),
_make_track("101", "Track One", "PT3M", "ISRC1", ["1001"]),
_make_track("102", "Track Two", "PT4M", "ISRC2", ["1001"]),
]
album, track_lookup, artist_lookup = _make_album(
"al2", "Album Two", tracks, ["a1"]
"2", "Album Two", tracks, ["1001"]
)
info = self.tidal._get_album_info(album, track_lookup, artist_lookup)
@@ -149,12 +155,13 @@ class TestAlbumParsing(TidalPluginTest):
assert len(info.tracks) == 2
assert info.tracks[0].index == 1
assert info.tracks[1].index == 2
assert info.tidal_album_id == 2
def test_parse_album_with_version(self):
"""Album title should have version appended."""
track = _make_track("t1", "My Song", "PT3M", "ISRC001", ["a1"])
track = _make_track("101", "My Song", "PT3M", "ISRC001", ["1001"])
album, track_lookup, artist_lookup = _make_album(
"al3", "My Album", [track], ["a1"], version="Deluxe Edition"
"3", "My Album", [track], ["1001"], version="Deluxe Edition"
)
info = self.tidal._get_album_info(album, track_lookup, artist_lookup)
@@ -166,27 +173,31 @@ class TestTrackParsing(TidalPluginTest):
"""High-level tests for track parsing."""
def test_parse_track(self):
track = _make_track("t1", "My Track", "PT4M", "ISRC456", ["a1"])
artist_lookup = {"a1": _make_artist("a1", "My Artist")}
track = _make_track("101", "My Track", "PT4M", "ISRC456", ["1001"])
artist_lookup = {"1001": _make_artist("1001", "My Artist")}
info = self.tidal._get_track_info(track, artist_lookup)
assert info.title == "My Track"
assert info.track_id == "t1"
assert info.track_id == "101"
assert info.duration == 240 # PT4M = 240 seconds
assert info.isrc == "ISRC456"
assert info.artist == "My Artist"
assert info.tidal_track_id == 101
assert info.tidal_artist_id == 1001
assert info.tidal_track_popularity == 50
def test_parse_track_with_version(self):
"""Track title should have version appended."""
track = _make_track(
"t2", "My Song", "PT3M", "ISRC002", ["a1"], version="Remastered"
"102", "My Song", "PT3M", "ISRC002", ["1001"], version="Remastered"
)
artist_lookup = {"a1": _make_artist("a1", "My Artist")}
artist_lookup = {"1001": _make_artist("1001", "My Artist")}
info = self.tidal._get_track_info(track, artist_lookup)
assert info.title == "My Song (Remastered)"
assert info.tidal_track_id == 102
class TestTrackForID(TidalPluginTest):
@@ -194,8 +205,10 @@ class TestTrackForID(TidalPluginTest):
def test_track_for_id(self):
"""Test fetching track by ID via API."""
track = _make_track("490839595", "API Track", "PT3M", "ISRC001", ["a1"])
artist = _make_artist("a1", "API Artist")
track = _make_track(
"490839595", "API Track", "PT3M", "ISRC001", ["1001"]
)
artist = _make_artist("1001", "API Artist")
self.tidal.api.get_tracks = Mock(
return_value={"data": [track], "included": [artist]}
@@ -225,12 +238,12 @@ class TestTracksForIDs(TidalPluginTest):
def test_tracks_for_ids(self):
"""Test fetching multiple tracks by IDs via API."""
track1 = _make_track(
"490839595", "API Track 1", "PT3M", "ISRC001", ["a1"]
"490839595", "API Track 1", "PT3M", "ISRC001", ["1001"]
)
track2 = _make_track(
"490839596", "API Track 2", "PT4M", "ISRC002", ["a1"]
"490839596", "API Track 2", "PT4M", "ISRC002", ["1001"]
)
artist = _make_artist("a1", "API Artist")
artist = _make_artist("1001", "API Artist")
self.tidal.api.get_tracks = Mock(
return_value={"data": [track1, track2], "included": [artist]}
@@ -253,8 +266,10 @@ class TestTracksForIDs(TidalPluginTest):
def test_tracks_for_ids_with_missing(self):
"""Test tracks_for_ids yields None for IDs not found."""
track = _make_track("490839595", "API Track", "PT3M", "ISRC001", ["a1"])
artist = _make_artist("a1", "API Artist")
track = _make_track(
"490839595", "API Track", "PT3M", "ISRC001", ["1001"]
)
artist = _make_artist("1001", "API Artist")
self.tidal.api.get_tracks = Mock(
return_value={"data": [track], "included": [artist]}
@@ -277,9 +292,11 @@ class TestAlbumForID(TidalPluginTest):
def test_album_for_id(self):
"""Test fetching album by ID via API."""
track = _make_track("t1", "Album Track", "PT3M30S", "ISRC001", ["a1"])
track = _make_track(
"101", "Album Track", "PT3M30S", "ISRC001", ["1001"]
)
album, track_lookup, artist_lookup = _make_album(
"226495055", "API Album", [track], ["a1"]
"226495055", "API Album", [track], ["1001"]
)
self.tidal.api.get_albums = Mock(
return_value={
@@ -311,13 +328,17 @@ class TestAlbumsForIDs(TidalPluginTest):
def test_albums_for_ids(self):
"""Test fetching multiple albums by IDs via API."""
track1 = _make_track("t1", "Album Track 1", "PT3M", "ISRC001", ["a1"])
track2 = _make_track("t2", "Album Track 2", "PT4M", "ISRC002", ["a1"])
track1 = _make_track(
"101", "Album Track 1", "PT3M", "ISRC001", ["1001"]
)
track2 = _make_track(
"102", "Album Track 2", "PT4M", "ISRC002", ["1001"]
)
album1, track_lookup1, artist_lookup1 = _make_album(
"226495055", "API Album 1", [track1], ["a1"]
"226495055", "API Album 1", [track1], ["1001"]
)
album2, track_lookup2, artist_lookup2 = _make_album(
"226495056", "API Album 2", [track2], ["a1"]
"226495056", "API Album 2", [track2], ["1001"]
)
# Combine lookups to simulate API response
@@ -351,9 +372,9 @@ class TestAlbumsForIDs(TidalPluginTest):
def test_albums_for_ids_with_missing(self):
"""Test albums_for_ids yields None for IDs not found."""
track = _make_track("t1", "Album Track", "PT3M", "ISRC001", ["a1"])
track = _make_track("101", "Album Track", "PT3M", "ISRC001", ["1001"])
album, track_lookup, artist_lookup = _make_album(
"226495055", "API Album", [track], ["a1"]
"226495055", "API Album", [track], ["1001"]
)
self.tidal.api.get_albums = Mock(
@@ -381,9 +402,9 @@ class TestCandidates(TidalPluginTest):
def test_candidates_with_barcode(self):
"""Test that candidates uses barcode lookup first."""
track = _make_track("t1", "Album Track", "PT3M", "ISRC001", ["a1"])
track = _make_track("101", "Album Track", "PT3M", "ISRC001", ["1001"])
album, track_lookup, artist_lookup = _make_album(
"al1", "Barcode Album", [track], ["a1"]
"1", "Barcode Album", [track], ["1001"]
)
self.tidal.api.get_albums = Mock(
@@ -412,16 +433,16 @@ class TestCandidates(TidalPluginTest):
return_value={
"data": {
"relationships": {
"albums": {"data": [{"id": "al1", "type": "albums"}]}
"albums": {"data": [{"id": "1", "type": "albums"}]}
}
}
}
)
# Mock album lookup by ID
track = _make_track("t1", "Album Track", "PT3M", "ISRC001", ["a1"])
track = _make_track("101", "Album Track", "PT3M", "ISRC001", ["1001"])
album, track_lookup, artist_lookup = _make_album(
"al1", "Query Album", [track], ["a1"]
"1", "Query Album", [track], ["1001"]
)
self.tidal.api.get_albums = Mock(
return_value={
@@ -446,9 +467,9 @@ class TestItemCandidates(TidalPluginTest):
def test_item_candidates_with_isrc(self):
"""Test that item_candidates uses ISRC lookup first."""
track = _make_track(
"490839595", "ISRC Track", "PT3M", "ISRC001", ["a1"]
"490839595", "ISRC Track", "PT3M", "ISRC001", ["1001"]
)
artist = _make_artist("a1", "ISRC Artist")
artist = _make_artist("1001", "ISRC Artist")
self.tidal.api.get_tracks = Mock(
return_value={"data": [track], "included": [artist]}
@@ -477,9 +498,9 @@ class TestItemCandidates(TidalPluginTest):
},
"included": [
_make_track(
"490839595", "Query Track", "PT3M", "ISRC002", ["a1"]
"490839595", "Query Track", "PT3M", "ISRC002", ["1001"]
),
_make_artist("a1", "Query Artist"),
_make_artist("1001", "Query Artist"),
],
}
)
@@ -550,3 +571,103 @@ class TestStaticHelpers:
with caplog.at_level(logging.WARNING):
TidalPlugin._duration_to_seconds("invalid")
assert "Invalid ISO 8601 duration: invalid" in caplog.text
def test_popularity_with_float(self):
assert TidalPlugin._popularity(0.5) == 50
assert TidalPlugin._popularity(1.0) == 100
assert TidalPlugin._popularity(0.0) == 0
def test_popularity_with_int(self):
assert TidalPlugin._popularity(50) == 50
assert TidalPlugin._popularity(100) == 100
assert TidalPlugin._popularity(0) == 0
def test_popularity_with_none(self):
assert TidalPlugin._popularity(None) is None
class TestTidalsync(TidalPluginTest):
"""Tests for the tidalsync command."""
def test_sync_updates_popularity(self):
"""Test that tidalsync fetches and stores popularity."""
item = self.lib.add_item(
Item(
tidal_track_id=490839595,
title="Test Track",
artist="Test Artist",
path="/dev/null/test.flac",
)
)
track = _make_track(
"490839595", "Test Track", "PT3M", "ISRC001", ["1001"]
)
artist = _make_artist("1001", "Test Artist")
self.tidal.api.get_tracks = Mock(
return_value={"data": [track], "included": [artist]}
)
self.tidal.tidalsync([item], write=False)
assert item["tidal_track_popularity"] == 50
assert item["tidal_updated"] is not None
assert isinstance(item["tidal_updated"], (int, float))
def test_sync_skips_existing(self):
"""Test that tidalsync skips items with existing popularity."""
item = self.lib.add_item(
Item(
tidal_track_id=490839595,
tidal_track_popularity=42,
tidal_updated=time.time(),
title="Test Track",
artist="Test Artist",
path="/dev/null/test.flac",
)
)
self.tidal.tidalsync([item], write=False)
assert item["tidal_track_popularity"] == 42
def test_sync_force_updates_existing(self):
"""Test that tidalsync with --force re-fetches."""
item = self.lib.add_item(
Item(
tidal_track_id=490839595,
tidal_track_popularity=42,
tidal_updated=time.time(),
title="Test Track",
artist="Test Artist",
path="/dev/null/test.flac",
)
)
track = _make_track(
"490839595", "Test Track", "PT3M", "ISRC001", ["1001"]
)
artist = _make_artist("1001", "Test Artist")
self.tidal.api.get_tracks = Mock(
return_value={"data": [track], "included": [artist]}
)
self.tidal.tidalsync([item], write=False, force=True)
assert item["tidal_track_popularity"] == 50
def test_sync_skips_items_without_track_id(self):
"""Test that tidalsync skips items without tidal_track_id."""
item = self.lib.add_item(
Item(
title="No ID Track",
artist="No ID Artist",
path="/dev/null/test.flac",
)
)
self.tidal.tidalsync([item], write=False)
assert item.get("tidal_track_popularity") is None