Refactor: Use pluggy framework to extract Sync to plugins

Pipfile

@@ -21,6 +21,7 @@ oauthlib = "==3.2.2"
 pfzy = {version="==0.3.4", markers="python_version >= '3.7' and python_version < '4.0'"}
 platformdirs = {version="==4.2.0", python_version=">='3.7'"}
 plexapi = "==4.15.11"
+pluggy = "==1.4.0"
 prompt-toolkit = "==3.0.43"
 pygments = "==2.17.2"
 python-dotenv = "==1.0.1"

plextraktsync/plugin/__init__.py

@@ -0,0 +1 @@
+from .plugin import hookimpl, hookspec  # noqa: F401

plextraktsync/plugin/plugin.py

@@ -0,0 +1,4 @@
+import pluggy
+
+hookspec = pluggy.HookspecMarker("PlexTraktSync")
+hookimpl = pluggy.HookimplMarker("PlexTraktSync")

plextraktsync/sync/AddCollectionPlugin.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from plextraktsync.factory import logging
+from plextraktsync.plugin import hookimpl
+
+if TYPE_CHECKING:
+    from plextraktsync.config.SyncConfig import SyncConfig
+    from plextraktsync.media.Media import Media
+    from plextraktsync.sync.Sync import Sync
+
+
+class AddCollectionPlugin:
+    logger = logging.getLogger(__name__)
+
+    @staticmethod
+    def enabled(config: SyncConfig):
+        return config.plex_to_trakt["collection"]
+
+    @classmethod
+    def factory(cls, sync: Sync):
+        return cls()
+
+    @hookimpl
+    def walk_movie(self, movie: Media, dry_run: bool):
+        self.sync_collection(movie, dry_run=dry_run)
+
+    @hookimpl
+    def walk_episode(self, episode: Media, dry_run: bool):
+        self.sync_collection(episode, dry_run=dry_run)
+
+    def sync_collection(self, m: Media, dry_run: bool):
+        if m.is_collected:
+            return
+
+        self.logger.info(f"Adding to Trakt collection: {m.title_link}", extra={"markup": True})
+
+        if not dry_run:
+            m.add_to_collection()

plextraktsync/sync/ClearCollectedPlugin.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Iterable
+
+from plextraktsync.factory import logging
+from plextraktsync.media.Media import Media
+from plextraktsync.plugin import hookimpl
+
+if TYPE_CHECKING:
+    from plextraktsync.config.SyncConfig import SyncConfig
+    from plextraktsync.sync.Sync import Sync
+    from plextraktsync.trakt.TraktApi import TraktApi
+    from plextraktsync.trakt.types import TraktMedia
+
+
+class ClearCollectedPlugin:
+    logger = logging.getLogger(__name__)
+
+    def __init__(self, trakt: TraktApi):
+        self.trakt = trakt
+        self.episode_trakt_ids = set()
+        self.movie_trakt_ids = set()
+        self.is_partial = None
+
+    @staticmethod
+    def enabled(config: SyncConfig):
+        return config.clear_collected
+
+    @classmethod
+    def factory(cls, sync: Sync):
+        return cls(sync.trakt)
+
+    @hookimpl
+    def init(self, is_partial: bool):
+        self.is_partial = is_partial
+        if is_partial:
+            self.logger.warning("Running partial library sync. Clear collected will be disabled.")
+
+    @hookimpl
+    def fini(self, dry_run: bool):
+        if self.is_partial:
+            return
+
+        self.clear_collected(self.trakt.movie_collection, self.movie_trakt_ids, dry_run=dry_run)
+        self.clear_collected(self.trakt.episodes_collection, self.episode_trakt_ids, dry_run=dry_run)
+
+    @hookimpl
+    def walk_movie(self, movie: Media):
+        if self.is_partial:
+            return
+
+        self.movie_trakt_ids.add(movie.trakt_id)
+
+    def clear_collected(self, existing_items: Iterable[TraktMedia], keep_ids: set[int], dry_run):
+        from plextraktsync.trakt.trakt_set import trakt_set
+
+        existing_ids = trakt_set(existing_items)
+        delete_ids = existing_ids - keep_ids
+        delete_items = (tm for tm in existing_items if tm.trakt in delete_ids)
+
+        n = len(delete_ids)
+        for i, tm in enumerate(delete_items, start=1):
+            self.logger.info(f"Remove from Trakt collection ({i}/{n}): {tm}")
+            if not dry_run:
+                self.trakt.remove_from_collection(tm)

plextraktsync/sync/LikedListsPlugin.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from plextraktsync.factory import logging
+from plextraktsync.plugin import hookimpl
+
+if TYPE_CHECKING:
+    from plextraktsync.config.SyncConfig import SyncConfig
+    from plextraktsync.sync.Sync import Sync
+    from plextraktsync.trakt.TraktApi import TraktApi
+    from plextraktsync.trakt.TraktUserListCollection import \
+        TraktUserListCollection
+
+
+class LikedListsPlugin:
+    logger = logging.getLogger(__name__)
+
+    def __init__(self, trakt: TraktApi):
+        self.trakt = trakt
+
+    @staticmethod
+    def enabled(config: SyncConfig):
+        return config.sync_liked_lists
+
+    @classmethod
+    def factory(cls, sync: Sync):
+        return cls(sync.trakt)
+
+    @hookimpl
+    def init(self, trakt_lists: TraktUserListCollection, is_partial: bool):
+        if is_partial:
+            self.logger.warning("Partial walk, disabling liked lists updating. "
+                                "Liked lists won't update because it needs full library sync.")
+        else:
+            trakt_lists.load_lists(self.trakt.liked_lists)