Merge pull request #1475 from glensc/refactor-pl

Refactor: Playlists refactoring

plextraktsync/media.py

@@ -81,6 +81,10 @@ def episode_number(self):
     def trakt_id(self):
         return self.trakt.trakt
 
+    @cached_property
+    def plex_key(self):
+        return self.plex.key
+
     @property
     def trakt_url(self):
         path = self.trakt.slug if self.trakt.slug else self.trakt_id

plextraktsync/plex/PlexPlaylist.py

@@ -3,7 +3,11 @@
 from functools import cached_property
 from typing import TYPE_CHECKING
 
+from rich.markup import escape
+
+from plextraktsync.decorators.flatten import flatten_dict
 from plextraktsync.factory import logging
+from plextraktsync.media import Media
 
 if TYPE_CHECKING:
     from plexapi.playlist import Playlist
@@ -18,6 +22,15 @@ def __init__(self, server: PlexServer, name: str):
         self.name = name
         self.logger = logging.getLogger("PlexTraktSync.PlexPlaylist")
 
+    def __iter__(self):
+        return iter(self.items)
+
+    def __len__(self):
+        return len(self.items.keys())
+
+    def __contains__(self, m: Media):
+        return m.plex_key in self.items
+
     @cached_property
     def playlist(self) -> Playlist | None:
         try:
@@ -32,6 +45,14 @@ def playlist(self) -> Playlist | None:
             self.logger.debug(f'Unable to find Plex playlist with title "{self.name}".')
             return None
 
+    @cached_property
+    @flatten_dict
+    def items(self) -> dict[int, PlexMedia]:
+        if self.playlist is None:
+            return
+        for m in self.playlist.items():
+            yield m.ratingKey, m
+
     def update(self, items: list[PlexMedia], description=None) -> bool:
         """
         Updates playlist (creates if name missing) replacing contents with items[]
@@ -40,6 +61,7 @@ def update(self, items: list[PlexMedia], description=None) -> bool:
         if playlist is None and len(items) > 0:
             # Force reload
             del self.__dict__["playlist"]
+            del self.__dict__["items"]
             playlist = self.server.createPlaylist(self.name, items=items)
             self.logger.info(f"Created plex playlist '{self.name}' with {len(items)} items")
 
@@ -63,6 +85,15 @@ def update(self, items: list[PlexMedia], description=None) -> bool:
 
         return True
 
+    @property
+    def title_link(self):
+        if self.playlist is not None:
+            link = self.playlist._getWebURL()
+
+            return f"[link={link}][green]{escape(self.name)}[/][/]"
+
+        return f"[green]{escape(self.name)}[/]"
+
     @staticmethod
     def same_list(list_a: list[PlexMedia], list_b: list[PlexMedia]) -> bool:
         """

plextraktsync/plex/PlexPlaylistCollection.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from collections import UserDict
+from typing import TYPE_CHECKING
+
+from plextraktsync.plex.PlexPlaylist import PlexPlaylist
+
+if TYPE_CHECKING:
+    from plexapi.server import PlexServer
+
+
+class PlexPlaylistCollection(UserDict):
+    def __init__(self, server: PlexServer):
+        super().__init__()
+        self.server = server
+
+    def __missing__(self, name: str):
+        playlist = PlexPlaylist(self.server, name)
+        self[name] = playlist
+
+        return playlist

plextraktsync/sync.py

@@ -5,6 +5,7 @@
 
 from plextraktsync.decorators.measure_time import measure_time
 from plextraktsync.factory import logger
+from plextraktsync.trakt.TraktUserListCollection import TraktUserListCollection
 from plextraktsync.trakt.types import TraktMedia
 from plextraktsync.trakt_list_util import TraktListUtil
 
@@ -41,8 +42,8 @@ def trakt_wl(self):
         return TraktWatchList(self.trakt.watchlist_movies + self.trakt.watchlist_shows)
 
     def sync(self, walker: Walker, dry_run=False):
-        listutil = TraktListUtil()
-        is_partial = walker.is_partial
+        trakt_lists = TraktUserListCollection()
+        is_partial = walker.is_partial and not dry_run
 
         if is_partial and self.config.clear_collected:
             logger.warning("Running partial library sync. Clear collected will be disabled.")
@@ -51,14 +52,13 @@ def sync(self, walker: Walker, dry_run=False):
             if is_partial:
                 logger.warning("Running partial library sync. Watchlist as playlist won't update because it needs full library sync.")
             else:
-                listutil.addList(None, "Trakt Watchlist", trakt_list=self.trakt.watchlist_movies)
+                trakt_lists.add_watchlist(self.trakt.watchlist_movies)
 
         if self.config.sync_liked_lists:
             if is_partial:
                 logger.warning("Partial walk, disabling liked lists updating. Liked lists won't update because it needs full library sync.")
             else:
-                for lst in self.trakt.liked_lists:
-                    listutil.addList(lst["listid"], lst["listname"])
+                trakt_lists.load_lists(self.trakt.liked_lists)
 
         if self.config.need_library_walk:
             movie_trakt_ids = set()
@@ -67,7 +67,7 @@ def sync(self, walker: Walker, dry_run=False):
                 self.sync_ratings(movie, dry_run=dry_run)
                 self.sync_watched(movie, dry_run=dry_run)
                 if not is_partial:
-                    listutil.addPlexItemToLists(movie)
+                    trakt_lists.add_to_lists(movie)
                     if self.config.clear_collected:
                         movie_trakt_ids.add(movie.trakt_id)
 
@@ -81,7 +81,7 @@ def sync(self, walker: Walker, dry_run=False):
                 self.sync_ratings(episode, dry_run=dry_run)
                 self.sync_watched(episode, dry_run=dry_run)
                 if not is_partial:
-                    listutil.addPlexItemToLists(episode)
+                    trakt_lists.add_to_lists(episode)
                     if self.config.clear_collected:
                         episode_trakt_ids.add(episode.trakt_id)
 
@@ -99,8 +99,9 @@ def sync(self, walker: Walker, dry_run=False):
             if is_partial:
                 logger.warning("Running partial library sync. Liked lists won't update because it needs full library sync.")
             else:
-                with measure_time("Updated liked list"):
-                    self.update_playlists(listutil, dry_run=dry_run)
+                if not dry_run:
+                    with measure_time("Updated liked list"):
+                        trakt_lists.sync()
 
         if walker.config.walk_watchlist and self.sync_wl:
             with measure_time("Updated watchlist"):

plextraktsync/trakt/TraktApi.py

@@ -20,14 +20,14 @@
 from plextraktsync.trakt.PartialTraktMedia import PartialTraktMedia
 from plextraktsync.trakt.TraktLookup import TraktLookup
 from plextraktsync.trakt.TraktRatingCollection import TraktRatingCollection
-from plextraktsync.trakt.types import TraktMedia
 
 if TYPE_CHECKING:
     from trakt.movies import Movie
     from trakt.tv import TVEpisode, TVShow
 
     from plextraktsync.plex.PlexGuid import PlexGuid
     from plextraktsync.plex.PlexLibraryItem import PlexLibraryItem
+    from plextraktsync.trakt.types import TraktLikedList, TraktMedia
 
 
 class TraktApi:
@@ -58,12 +58,13 @@ def me(self):
     @rate_limit()
     @retry()
     @flatten_list
-    def liked_lists(self):
+    def liked_lists(self) -> list[TraktLikedList]:
         for item in self.me.get_liked_lists("lists", limit=1000):
-            yield {
+            tll: TraktLikedList = {
                 'listname': item['list']['name'],
                 'listid': item['list']['ids']['trakt'],
             }
+            yield tll
 
     @cached_property
     @rate_limit()

plextraktsync/trakt/TraktUserList.py

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+from functools import cached_property
+from itertools import count
+from typing import TYPE_CHECKING
+
+from plextraktsync.factory import factory, logging
+
+if TYPE_CHECKING:
+    from trakt.movies import Movie
+    from trakt.tv import TVEpisode
+
+    from plextraktsync.media import Media
+    from plextraktsync.plex.PlexLibraryItem import PlexLibraryItem
+
+
+class TraktUserList:
+    plex_items: list[tuple[int, PlexLibraryItem]]
+
+    def __init__(self,
+                 trakt_id: int = None,
+                 name: str = None,
+                 items=None,
+                 ):
+        if items is None:
+            items = []
+        self.trakt_id = trakt_id
+        self.name = name
+        self._items = items
+        self.description = None
+        self.plex_items = []
+        self.logger = logging.getLogger("PlexTraktSync.TraktUserList")
+
+    def __iter__(self):
+        return iter(self.items)
+
+    def __len__(self):
+        return len(self.items)
+
+    def __contains__(self, m: Media):
+        rank = self.items.get((m.media_type, m.trakt_id))
+
+        return rank is not None
+
+    @property
+    def items(self):
+        if not self._items:
+            self.description, self._items = self.load_items()
+        return self._items
+
+    def load_items(self):
+        from plextraktsync.trakt_list_util import LazyUserList
+
+        userlist = LazyUserList._get(self.name, self.trakt_id)
+        list_items = userlist._items
+        prelist = [
+            (elem[0], elem[1])
+            for elem in list_items
+            if elem[0] in ["movies", "episodes"]
+        ]
+        self.logger.info(f"Downloaded Trakt list '{self.name}' https://trakt.tv/lists/{self.trakt_id}")
+
+        return userlist.description, dict(zip(prelist, count(1)))
+
+    @classmethod
+    def from_trakt_list(cls, name: str, items: list[Movie, TVEpisode]):
+        items = zip([(item.media_type, item.trakt) for item in items], count(1))
+
+        return cls(name=name, items=dict(items))
+
+    @classmethod
+    def from_watchlist(cls, items: list[Movie, TVEpisode]):
+        trakt_items = dict(
+            zip([(elem.media_type, elem.trakt) for elem in items], count(1))
+        )
+        return cls(name="Trakt Watchlist", items=trakt_items)
+
+    @cached_property
+    def plex_lists(self):
+        return factory.plex_lists
+
+    @cached_property
+    def plex_list(self):
+        if not self.name:
+            raise RuntimeError("Name is required")
+
+        return self.plex_lists[self.name]
+
+    def add(self, m: Media):
+        rank = self.items.get((m.media_type, m.trakt_id))
+        if rank is None:
+            # Item is not in this trakt list
+            return
+
+        # TODO: add with rank
+        self.plex_items.append((rank, m.plex))
+
+        if m in self.plex_list:
+            # Already in the list
+            return
+
+        self.logger.info(f'Adding {m.title_link} ({m.plex_key}) to Plex list {self.title_link}', extra={"markup": True})
+
+    @property
+    def title_link(self):
+        return self.plex_list.title_link
+
+    @property
+    def plex_items_sorted(self):
+        """
+        Returns items sorted by trakt rank
+
+        https://github.com/Taxel/PlexTraktSync/pull/58
+        """
+        if len(self.plex_items) == 0:
+            return []
+
+        plex_items = [(r, p.item) for (r, p) in self.plex_items]
+        _, items = zip(*sorted(dict(reversed(plex_items)).items()))
+
+        return items

plextraktsync/trakt/TraktUserListCollection.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+from collections import UserList
+from typing import TYPE_CHECKING
+
+from plextraktsync.factory import logging
+from plextraktsync.trakt.TraktUserList import TraktUserList
+
+if TYPE_CHECKING:
+    from trakt.movies import Movie
+    from trakt.tv import TVEpisode
+
+    from plextraktsync.media import Media
+    from plextraktsync.trakt.types import TraktLikedList
+
+
+class TraktUserListCollection(UserList):
+    def __init__(self):
+        super().__init__()
+        self.logger = logging.getLogger("PlexTraktSync.TraktUserListCollection")
+
+    def add_to_lists(self, m: Media):
+        for tl in self:
+            tl.add(m)
+
+    def load_lists(self, liked_lists: list[TraktLikedList]):
+        for liked_list in liked_lists:
+            self.add_list(liked_list["listid"], liked_list["listname"])
+
+    def add_watchlist(self, items: list[Movie, TVEpisode]):
+        tl = TraktUserList.from_watchlist(items)
+        self.append(tl)
+        return tl
+
+    def add_list(self, list_id: int, list_name: str):
+        tl = TraktUserList(list_id, list_name)
+        self.append(tl)
+        return tl
+
+    def sync(self):
+        for tl in self:
+            updated = tl.plex_list.update(tl.plex_items_sorted)
+            if not updated:
+                continue
+            self.logger.info(f"Plex list {tl.title_link} ({len(tl.plex_items)} items) updated", extra={"markup": True})

plextraktsync/trakt/types.py

@@ -1,6 +1,11 @@
-from typing import Union
+from typing import TypedDict, Union
 
 from trakt.movies import Movie
 from trakt.tv import TVEpisode, TVSeason, TVShow
 
 TraktMedia = Union[Movie, TVShow, TVSeason, TVEpisode]
+
+
+class TraktLikedList(TypedDict):
+    listid: int
+    listname: str

plextraktsync/util/Factory.py

@@ -80,6 +80,13 @@ def plex_server(self):
             token=server.token,
         )
 
+    @cached_property
+    def plex_lists(self):
+        from plextraktsync.plex.PlexPlaylistCollection import \
+            PlexPlaylistCollection
+
+        return PlexPlaylistCollection(self.plex_server)
+
     @cached_property
     def has_plex_token(self):
         try: