Add include/exclude for playlist groups.

spotcrates/cli.py

@@ -6,13 +6,15 @@
 
 import argparse
 import logging
+import re
 import sys
 from typing import Dict, Any, List
 
 import pygtrie
 import tomli_w
 
-from spotcrates.common import BaseLookup, truncate_long_value, get_spotify_handle, DEFAULT_CONFIG_FILE, get_config
+from spotcrates.common import BaseLookup, truncate_long_value, get_spotify_handle, DEFAULT_CONFIG_FILE, get_config, \
+    ValueFilter
 from spotcrates.filters import FieldName
 
 import importlib.metadata
@@ -28,6 +30,8 @@
 # Turn down noisy third-party debug logs
 logging.getLogger('spotipy').setLevel(logging.INFO)
 logging.getLogger('urllib3').setLevel(logging.INFO)
+# Logs on error for 404s (which we don't want to see)
+logging.getLogger('spotipy.client').setLevel(logging.FATAL)
 
 logger = logging.getLogger(__name__)
 
@@ -44,8 +48,8 @@
 {'subscriptions':<16} Add new tracks from configured playlists to the target playlist, filtering for excluded entries.
 """
 
-# LOG_FORMAT = "%(levelname)s (%(name)s): %(message)s"
-LOG_FORMAT = "%(message)s"
+LOG_FORMAT = "%(levelname)s (%(name)s): %(message)s"
+# LOG_FORMAT = "%(message)s"
 
 
 def print_commands():
@@ -69,7 +73,8 @@ def append_recent_subscriptions(config: Dict[str, Any], args: argparse.Namespace
     sp = get_spotify_handle(config)
 
     playlists = Playlists(sp, config.get("subscriptions"))
-    playlists.append_recent_subscriptions(args.randomize, args.target)
+    playlist_set_filter = ValueFilter(includes=args.include_playlist_sets, excludes=args.exclude_playlist_sets)
+    playlists.append_recent_subscriptions(args.randomize, args.target, playlist_set_filter)
 
 
 def randomize_lists(config: Dict[str, Any], args: argparse.Namespace):
@@ -149,6 +154,13 @@ def list_playlists(config: Dict[str, Any], args: argparse.Namespace):
 }
 
 
+def playlist_sets(value: str | None) -> list[str]:
+    """Parses the playlist groups from the command line."""
+    if not value:
+        return [""]
+    return [clean.strip() for clean in re.split("[,|]", value)]
+
+
 def init_config(args: argparse.Namespace):
     config_file = args.config_file
 
@@ -188,16 +200,21 @@ def parse_cmdline(argv: List):
     parser.add_argument("-c", "--config_file",
                         help=f"The location of the config file (default: {DEFAULT_CONFIG_FILE})",
                         default=DEFAULT_CONFIG_FILE, type=Path)
-    parser.add_argument("-s", "--sort_fields", help="The fields to sort against, applied in order")
     parser.add_argument("-f", "--filters", help="Filters to apply to the list")
+    parser.add_argument("-e", "--exclude_playlist_sets", type=playlist_sets,
+                        action="append", help="The playlist sets to exclude. Takes precedence over include")
+    parser.add_argument("-i", "--include_playlist_sets", type=playlist_sets,
+                        action="append", help="The playlist sets to include. Includes all if not specified")
     parser.add_argument("-r", "--randomize", help="Randomize the target list", action='store_true')
     parser.add_argument('--version', action='version', version=__version__)
+    parser.add_argument("-s", "--sort_fields", help="The fields to sort against, applied in order")
     parser.add_argument("-t", "--target",
                         help="Specify the target name of the operation (overrides any default value)")
     parser.add_argument("command", metavar="COMMAND",
                         help=f"The command to run (one of {','.join(COMMANDS)})")
     parser.add_argument("arguments", metavar='ARGUMENTS', nargs='*',
                         help="the arguments to the command")
+    # noinspection PyTypeChecker
     parser.add_argument('-log',
                         '--loglevel',
                         default='info',

spotcrates/common.py

@@ -49,13 +49,15 @@ def get_all_items(spotify: Spotify, first_page: Dict[str, Any]):
         while next_page:
             all_items.extend(next_page["items"])
             next_page = spotify.next(next_page)
-    except Exception:
-        logging.warning("Problems paging given Spotify items list", exc_info=True)
+    except TimeoutError:
+        logging.info("Timeout error encountered while paging Spotify items")
+    except Exception as e:
+        logging.warning(f"Problems paging given Spotify items list: {e}", exc_info=True)
 
     return [item for item in all_items if item is not None]
 
 
-def truncate_long_value(full_value: str, length: int, trim_tail: bool = True) -> str:
+def truncate_long_value(full_value: str | None, length: int, trim_tail: bool = True) -> str:
     """Returns the given value truncated from the start of the value so that it is at most the given length.
 
     :param full_value: The value to trim.
@@ -139,3 +141,15 @@ def get_config(config_file: Path):
         return {}
 
 
+class ValueFilter:
+    def __init__(self, includes: list[str] | None = None, excludes: list[str] | None = None):
+        self._includes = includes
+        self._excludes = excludes
+
+    def include(self, value: str) -> bool:
+        if self._excludes and value in self._excludes:
+            return False
+        return not self._includes or value in self._includes
+
+    def exclude(self, value: str) -> bool:
+        return not self.include(value)

spotcrates/playlists.py

@@ -6,9 +6,9 @@
 from typing import List, Set, Dict, Tuple, Any, Iterable
 
 from durations_nlp import Duration
-from spotipy import Spotify
+from spotipy import Spotify, SpotifyException
 
-from spotcrates.common import batched, get_all_items, ISO_8601_TIMESTAMP_FORMAT, ZERO_TIMESTAMP
+from spotcrates.common import batched, get_all_items, ISO_8601_TIMESTAMP_FORMAT, ZERO_TIMESTAMP, ValueFilter
 from spotcrates.filters import FieldName, filter_list, sort_list
 
 config_defaults = {
@@ -163,13 +163,15 @@ def randomize_playlist(self, playlist: Dict[str, Any]) -> PlaylistResult:
             self.logger.warning(f"Problems randomizing playlist '{playlist['name']}'", exc_info=True)
             return PlaylistResult.FAILURE
 
-    def append_recent_subscriptions(self, randomize: bool, target_name: str):
+    def append_recent_subscriptions(
+            self, randomize: bool, target_name: str, playlist_set_filter: ValueFilter| None = None):
         """Adds new tracks from configured lists to the target list. Newness is determined
         based on the configured maximum age, which is 3 days by default.
 
         :param randomize: Whether to randomize the new tracks before appending them.
         :param target_name: The target list to append to. The configured target list at
           'subscriptions_target' is used if no name is provided here.
+        :param playlist_set_filter: A filter to apply to the playlist sets.
         """
         target_list = None
         exclude_lists = []
@@ -178,6 +180,10 @@ def append_recent_subscriptions(self, randomize: bool, target_name: str):
         else:
             subscriptions_target = self.config.get("subscriptions_target")
         exclude_prefix = self.config.get("daily_mix_exclude_prefix")
+
+        if playlist_set_filter is None:
+            playlist_set_filter = ValueFilter()
+
         for playlist in self.get_all_playlists():
             list_name = playlist["name"]
             if list_name:
@@ -199,7 +205,8 @@ def append_recent_subscriptions(self, randomize: bool, target_name: str):
         excludes = self._get_excludes(exclude_lists, target_list)
 
         include_zero_timestamps = self.config.get("include_zero_timestamps", False)
-        playlist_ids = self._get_subscription_playlist_ids(oldest_timestamp, excludes, include_zero_timestamps)
+        playlist_ids = self._get_subscription_playlist_ids(
+            oldest_timestamp, excludes, include_zero_timestamps, playlist_set_filter)
 
         self.logger.info(f"{len(playlist_ids)} subscription tracks to add")
 
@@ -300,7 +307,8 @@ def _fetch_daily_tracks(self, dailies: List, exclude_ids: Iterable[str]):
     def _get_subscription_playlist_ids(self,
                                        oldest_timestamp: datetime,
                                        excluded_ids: Iterable[str],
-                                       include_zero_timestamps: bool) -> Set[str]:
+                                       include_zero_timestamps: bool,
+                                       playlist_set_filter: ValueFilter) -> Set[str]:
         target_playlist_ids: Set[str] = set()
         subscription_playlists = self.config.get("playlists")
         if not subscription_playlists:
@@ -309,6 +317,9 @@ def _get_subscription_playlist_ids(self,
 
         for playlist_set, playlist_ids in subscription_playlists.items():
             self.logger.debug(f"Processing subscription set '{playlist_set}'")
+            if playlist_set_filter.exclude(playlist_set):
+                self.logger.debug(f"Skipping set '{playlist_set}'")
+                continue
             set_playlist_ids = set()
             for track in self._get_playlist_id_tracks(*playlist_ids):
                 iso_added = track.get("added_at")
@@ -393,8 +404,19 @@ def _get_playlist_id_tracks(self, *playlist_ids: str) -> List[dict]:
 
         return tracks
 
-    def _filter_for_tracks(self, playlist_id):
-        all_tracks = get_all_items(self.spotify, self.spotify.playlist_items(playlist_id))
+    def _filter_for_tracks(self, playlist_id: str) -> List[dict]:
+        try:
+            all_tracks = get_all_items(self.spotify, self.spotify.playlist_items(playlist_id))
+        except SpotifyException as e:
+            if e.http_status == 404:
+                self.logger.info(f"Playlist '{playlist_id}' not found")
+                return []
+            else:
+                self.logger.warning(f"Spotify problems getting tracks for playlist '{playlist_id}': {e}")
+                return []
+        except Exception as e:
+            self.logger.warning(f"Problems getting tracks for playlist '{playlist_id}': {e}")
+            return []
         filtered_tracks = []
         for cur_track in all_tracks:
             if cur_track.get('track') and cur_track['track'].get('id'):

tests/test_cli.py

@@ -1,6 +1,6 @@
 import unittest
 
-from spotcrates.cli import parse_cmdline
+from spotcrates.cli import parse_cmdline, playlist_sets
 
 
 class ArgparseTestCase(unittest.TestCase):
@@ -48,3 +48,21 @@ def test_random_after_3args(self):
         self.assertEqual('test-command', args.command)
         self.assertTrue(args.randomize)
         self.assertSequenceEqual(['arg1', 'arg2', 'arg3'], args.arguments)
+
+
+# playlist_sets
+
+def test_playlist_sets():
+    assert playlist_sets("a,b,c") == ["a", "b", "c"]
+
+def test_playlist_sets_pipes():
+    assert playlist_sets("a|b|c") == ["a", "b", "c"]
+
+def test_playlist_sets_pipes_and_commas():
+    assert playlist_sets("a|b,c") == ["a", "b", "c"]
+
+def test_playlist_sets_blank():
+    assert playlist_sets("") == [""]
+
+def test_playlist_sets_none():
+    assert playlist_sets(None) == [""]

tests/test_common.py

@@ -1,6 +1,6 @@
 import unittest
 
-from spotcrates.common import truncate_long_value
+from spotcrates.common import truncate_long_value, ValueFilter
 
 
 # truncate_long_value
@@ -20,3 +20,35 @@ def test_empty(self):
 
     def test_null(self):
         self.assertEqual(None, truncate_long_value(None, 5, trim_tail=False))
+
+## ValueFilter ##
+
+# include
+def test_empty_value_filter():
+    assert ValueFilter().include("anything") == True
+
+def test_explicit_include():
+    assert ValueFilter(includes=["anything"]).include("anything") == True
+    assert ValueFilter(includes=["anything"]).include("nothing") == False
+
+def test_explicit_exclude():
+    assert ValueFilter(excludes=["anything"]).include("anything") == False
+    assert ValueFilter(excludes=["anything"]).include("nothing") == True
+
+def test_explicit_include_exclude():
+    assert ValueFilter(includes=["anything"], excludes=["nothing"]).include("anything") == True
+    assert ValueFilter(includes=["anything"], excludes=["nothing"]).include("nothing") == False
+
+def test_explicit_include_exclude_precedence():
+    assert ValueFilter(includes=["anything"], excludes=["anything"]).include("anything") == False
+
+# exclude
+def test_empty_value_filter_exclude():
+    assert ValueFilter().exclude("anything") == False
+
+def test_explicit_include_exclude_exclude():
+    assert ValueFilter(includes=["anything"], excludes=["anything"]).exclude("anything") == True
+    assert ValueFilter(includes=["anything"], excludes=["anything"]).exclude("nothing") == True
+
+def test_explicit_include_exclude_exclude_precedence():
+    assert ValueFilter(includes=["anything"], excludes=["anything"]).exclude("anything") == True