feat: new sync plugin to sync music metadata

docs/plugins/plugins.rst

@@ -19,4 +19,5 @@ These are all the plugins that are enabled by default.
    move
    musicbrainz
    remove
+   sync
    write

docs/plugins/sync.rst

@@ -0,0 +1,30 @@
+####
+Sync
+####
+Syncs music metadata from connected sources.
+
+*************
+Configuration
+*************
+The ``sync`` plugin is enabled by default.
+
+***********
+Commandline
+***********
+.. code-block:: bash
+
+    moe sync [-h] [-a | -e] [-p] query
+
+Options
+=======
+``-h, --help``
+    Display the help message.
+``-a, --album``
+    Query for matching albums instead of tracks.
+``-e, --extra``
+    Query for matching extras instead of tracks.
+
+Arguments
+=========
+``query``
+    Query your library for items to sync. See the :doc:`query docs <../query>` for more info.

moe/config.py

@@ -45,6 +45,7 @@
     "list",
     "move",
     "musicbrainz",
+    "sync",
     "remove",
     "write",
 }

moe/plugins/musicbrainz/mb_core.py

@@ -33,6 +33,7 @@
     "add_releases_to_collection",
     "get_album_by_id",
     "get_matching_album",
+    "get_track_by_id",
     "rm_releases_from_collection",
     "set_collection",
     "update_album",
@@ -173,6 +174,18 @@ def read_custom_tags(
     track_fields["mb_track_id"] = audio_file.mb_releasetrackid
 
 
+@moe.hookimpl
+def sync_metadata(item: LibItem):
+    """Sync musibrainz metadata for associated items."""
+    if isinstance(item, Album) and hasattr(item, "mb_album_id"):
+        item.merge(get_album_by_id(item.mb_album_id), overwrite=True)
+    elif isinstance(item, Track) and hasattr(item, "mb_track_id"):
+        item.merge(
+            get_track_by_id(item.mb_track_id, item.album_obj.mb_album_id),
+            overwrite=True,
+        )
+
+
 @moe.hookimpl
 def write_custom_tags(track: Track):
     """Write musicbrainz ID fields as tags."""
@@ -327,8 +340,7 @@ def get_matching_album(album: Album) -> Album:
         album: Album used to search for the release.
 
     Returns:
-        Dictionary of release information. See the ``tests/musicbrainz/resources``
-        directory for an idea of what this contains.
+        An Album containing all musicbrainz metadata.
     """
     if album.mb_album_id:
         return get_album_by_id(album.mb_album_id)
@@ -357,10 +369,7 @@ def get_album_by_id(release_id: str) -> Album:
         release_id: Musicbrainz release ID to search.
 
     Returns:
-        Dictionary of release information. See ``tests/resources/musicbrainz`` for
-        an idea of what this contains. Note this is a different dictionary that what
-        is returned from searching by fields for a release. Notably, searching by an id
-        results in more information including track information.
+        An album containing all metadata from the given ``release_id``.
     """
     log.debug(f"Fetching release from musicbrainz. [release={release_id!r}]")
 
@@ -422,6 +431,32 @@ def _flatten_artist_credit(artist_credit: list[dict]) -> str:
     return full_artist
 
 
+def get_track_by_id(track_id: str, album_id: str) -> Track:
+    """Gets a musicbrainz track from a given track and release id.
+
+    Args:
+        track_id: Musicbrainz track ID to match.
+        album_id: Release album ID the track belongs to.
+
+    Returns:
+        A track containing all metadata associated with the given IDs.
+
+    Raises:
+        ValueError: Couldn't find track based on given ``track_id`` and ``album_id``.
+    """
+    log.debug(f"Fetching track from musicbrainz. [{track_id=!r}, {album_id=!r}]")
+
+    album = get_album_by_id(album_id)
+    for track in album.tracks:
+        if track.mb_track_id == track_id:
+            log.info(f"Fetched track from musicbrainz. [{track=!r}]")
+            return track
+
+    raise ValueError(
+        f"Given track or album id could not be found. [{track_id=!r}, {album_id=!r}]"
+    )
+
+
 def update_album(album: Album):
     """Updates an album with metadata from musicbrainz.
 

moe/plugins/sync/__init__.py

@@ -0,0 +1,18 @@
+"""Syncs library metadata with external sources."""
+
+import moe
+from moe import config
+
+from . import sync_cli, sync_core
+from .sync_core import *
+
+__all__ = []
+__all__.extend(sync_core.__all__)
+
+
+@moe.hookimpl
+def plugin_registration():
+    """Only register the cli sub-plugin if the cli is enabled."""
+    config.CONFIG.pm.register(sync_core, "sync_core")
+    if config.CONFIG.pm.has_plugin("cli"):
+        config.CONFIG.pm.register(sync_cli, "sync_cli")

moe/plugins/sync/sync_cli.py

@@ -0,0 +1,48 @@
+"""Adds the ``sync`` command to sync library metadata."""
+
+import argparse
+import logging
+
+import moe
+import moe.cli
+from moe.plugins import sync as moe_sync
+from moe.query import QueryError, query
+
+log = logging.getLogger("moe.cli.sync")
+
+__all__: list[str] = []
+
+
+@moe.hookimpl
+def add_command(cmd_parsers: argparse._SubParsersAction):
+    """Adds the ``add`` command to Moe's CLI."""
+    add_parser = cmd_parsers.add_parser(
+        "sync",
+        description="Sync music metadata.",
+        help="sync music metadata",
+        parents=[moe.cli.query_parser],
+    )
+    add_parser.set_defaults(func=_parse_args)
+
+
+def _parse_args(args: argparse.Namespace):
+    """Parses the given commandline arguments.
+
+    Args:
+        args: Commandline arguments to parse.
+
+    Raises:
+        SystemExit: Invalid query given or query returned no items.
+    """
+    try:
+        items = query(args.query, query_type=args.query_type)
+    except QueryError as err:
+        log.error(err)
+        raise SystemExit(1) from err
+
+    if not items:
+        log.error("No items found to sync.")
+        raise SystemExit(1)
+
+    for item in items:
+        moe_sync.sync_item(item)

moe/plugins/sync/sync_core.py

@@ -0,0 +1,29 @@
+"""Sync metadata from external sources."""
+
+import logging
+
+import moe
+from moe import config
+from moe.library import LibItem
+
+log = logging.getLogger("moe.sync")
+
+__all__ = ["sync_item"]
+
+
+class Hooks:
+    """Sync plugin hook specifications."""
+
+    @staticmethod
+    @moe.hookspec
+    def sync_metadata(item: LibItem):
+        """Implement specific metadata syncing for ``item``."""
+
+
+def sync_item(item: LibItem):
+    """Syncs metadata from external sources and merges changes into ``item``."""
+    log.debug(f"Syncing metadata. [{item=!r}]")
+
+    config.CONFIG.pm.hook.sync_metadata(item)
+
+    log.debug(f"Synced metadata. [{item=!r}]")

tests/plugins/musicbrainz/test_mb_core.py

@@ -261,6 +261,38 @@ def test_invalid_credentials(self, caplog, mb_config):
         assert any(record.levelname == "ERROR" for record in caplog.records)
 
 
+class TestSyncMetadata:
+    """Test the `sync_metadata` hook implementation."""
+
+    def test_sync_album(self, mb_config):
+        """Albums are synced with musicbrainz when called."""
+        old_album = album_factory(title="unsynced", mb_album_id="1")
+        new_album = album_factory(title="synced")
+
+        with patch.object(
+            moe_mb.mb_core, "get_album_by_id", return_value=new_album
+        ) as mock_id:
+            config.CONFIG.pm.hook.sync_metadata(item=old_album)
+
+        mock_id.assert_called_once_with(old_album.mb_album_id)
+        assert old_album.title == "synced"
+
+    def test_sync_track(self, mb_config):
+        """Tracks are synced with musicbrainz when called."""
+        old_track = track_factory(title="unsynced", mb_track_id="1")
+        new_track = track_factory(title="synced")
+
+        with patch.object(
+            moe_mb.mb_core, "get_track_by_id", return_value=new_track
+        ) as mock_id:
+            config.CONFIG.pm.hook.sync_metadata(item=old_track)
+
+        mock_id.assert_called_once_with(
+            old_track.mb_track_id, old_track.album_obj.mb_album_id
+        )
+        assert old_track.title == "synced"
+
+
 class TestGetMatchingAlbum:
     """Test `get_matching_album()`."""
 
@@ -378,6 +410,30 @@ def test_multi_disc_release(self, mock_mb_by_id):
         assert any(track.disc == 2 for track in mb_album.tracks)
 
 
+class TestGetTrackByID:
+    """Test `get_track_by_id`."""
+
+    def test_track_search(self, mock_mb_by_id):
+        """We can't search for tracks so we search for albums and match tracks."""
+        mb_album_id = "112dec42-65f2-3bde-8d7d-26deddde10b2"
+        mb_track_id = "219e6b01-c962-355c-8a87-5d4ab3fc13bc"
+        mock_mb_by_id.return_value = mb_rsrc.full_release.release
+
+        mb_track = moe_mb.get_track_by_id(mb_track_id, mb_album_id)
+
+        assert mb_track.title == "Dark Fantasy"
+        mock_mb_by_id.assert_called_once_with(
+            mb_album_id, includes=moe_mb.mb_core.RELEASE_INCLUDES
+        )
+
+    def test_track_not_found(self, mock_mb_by_id):
+        """Raise ValueError if track or album cannot be found."""
+        mock_mb_by_id.return_value = mb_rsrc.full_release.release
+
+        with pytest.raises(ValueError, match="track_id"):
+            moe_mb.get_track_by_id("track id", "album id")
+
+
 class TestCustomFields:
     """Test reading, writing, and setting musicbrainz custom fields."""
 

tests/plugins/sync/test_sync_cli.py

@@ -0,0 +1,76 @@
+"""Test sync plugin cli."""
+
+from unittest.mock import call, patch
+
+import pytest
+
+import moe
+import moe.cli
+from moe.query import QueryError
+from tests.conftest import track_factory
+
+
+@pytest.fixture
+def mock_sync():
+    """Mock the `sync_item()` api call."""
+    with patch(
+        "moe.plugins.sync.sync_cli.moe_sync.sync_item", autospec=True
+    ) as mock_sync:
+        yield mock_sync
+
+
+@pytest.fixture
+def mock_query():
+    """Mock a database query call.
+
+    Use ``mock_query.return_value` to set the return value of a query.
+
+    Yields:
+        Mock query
+    """
+    with patch("moe.plugins.sync.sync_cli.query", autospec=True) as mock_query:
+        yield mock_query
+
+
+@pytest.fixture
+def _tmp_sync_config(tmp_config):
+    """A temporary config for the list plugin with the cli."""
+    tmp_config('default_plugins = ["cli", "sync"]')
+
+
+@pytest.mark.usefixtures("_tmp_sync_config")
+class TestCommand:
+    """Test the `sync` command."""
+
+    def test_items(self, mock_query, mock_sync):
+        """Tracks are synced with a valid query."""
+        track1 = track_factory()
+        track2 = track_factory()
+        cli_args = ["sync", "*"]
+        mock_query.return_value = [track1, track2]
+
+        moe.cli.main(cli_args)
+
+        mock_query.assert_called_once_with("*", query_type="track")
+        mock_sync.assert_has_calls([call(track1), call(track2)])
+
+    def test_exit_code(self, mock_query, mock_sync):
+        """Return a non-zero exit code if no items are returned from the query."""
+        cli_args = ["sync", "*"]
+        mock_query.return_value = []
+
+        with pytest.raises(SystemExit) as error:
+            moe.cli.main(cli_args)
+
+        assert error.value.code != 0
+        mock_sync.assert_not_called()
+
+    def test_bad_query(self, mock_query):
+        """Raise SystemExit if given a bad query."""
+        cli_args = ["sync", "*"]
+        mock_query.side_effect = QueryError
+
+        with pytest.raises(SystemExit) as error:
+            moe.cli.main(cli_args)
+
+        assert error.value.code != 0