Major code refactoring (#32)

* Major code refactoring

* Minor code improvements

* Add pylint to code quality test workflow

.github/workflows/code_quality_test.yml

@@ -21,11 +21,16 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          pip install flake8
+          pip install flake8 pylint
           if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
       - name: Lint with flake8
         run: |
           # stop the build if there are Python syntax errors or undefined names
           flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
           # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
           flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
+      - name: Lint with pylint
+        run: |
+          # stop the build if pylint gives a score under 9
+          export PWD=$(pwd)
+          python3 -m pylint plex_auto_languages --fail-under=9 --max-line-length=127 --disable=C0114,C0115,C0116,W1203,R0903,W0238 --init-hook='import sys; sys.path.append("${PWD}")'

plex_auto_languages/alerts/__init__.py

@@ -0,0 +1,5 @@
+from .base import PlexAlert             # noqa: F401
+from .activity import PlexActivity      # noqa: F401
+from .playing import PlexPlaying        # noqa: F401
+from .timeline import PlexTimeline      # noqa: F401
+from .status import PlexStatus          # noqa: F401

plex_auto_languages/alerts/activity.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+from datetime import datetime, timedelta
+from plexapi.video import Episode
+
+from plex_auto_languages.alerts.base import PlexAlert
+from plex_auto_languages.utils.logger import get_logger
+
+if TYPE_CHECKING:
+    from plex_auto_languages.plex_server import PlexServer
+
+
+logger = get_logger()
+
+
+class PlexActivity(PlexAlert):
+
+    TYPE = "activity"
+
+    TYPE_LIBRARY_REFRESH_ITEM = "library.refresh.items"
+    TYPE_LIBRARY_UPDATE_SECTION = "library.update.section"
+    TYPE_PROVIDER_SUBSCRIPTIONS_PROCESS = "provider.subscriptions.process"
+    TYPE_MEDIA_GENERATE_BIF = "media.generate.bif"
+
+    def is_type(self, activity_type: str):
+        return self.type == activity_type
+
+    @property
+    def event(self):
+        return self._message.get("event", None)
+
+    @property
+    def type(self):
+        return self._message.get("Activity", {}).get("type", None)
+
+    @property
+    def item_key(self):
+        return self._message.get("Activity", {}).get("Context", {}).get("key", None)
+
+    @property
+    def user_id(self):
+        return self._message.get("Activity", {}).get("userID", None)
+
+    def process(self, plex: PlexServer):
+        if self.event != "ended":
+            return
+        if not self.is_type(self.TYPE_LIBRARY_REFRESH_ITEM):
+            return
+
+        # Switch to the user's Plex instance
+        user_plex = plex.get_plex_instance_of_user(self.user_id)
+        if user_plex is None:
+            return
+
+        # Skip if not an Episode
+        item = user_plex.fetch_item(self.item_key)
+        if item is None or not isinstance(item, Episode):
+            return
+
+        # Skip if this item has already been seen in the last 3 seconds
+        activity_key = (self.user_id, self.item_key)
+        if activity_key in plex.cache.recent_activities and \
+                plex.cache.recent_activities[activity_key] > datetime.now() - timedelta(seconds=3):
+            return
+        plex.cache.recent_activities[activity_key] = datetime.now()
+
+        # Change tracks if needed
+        item.reload()
+        user = plex.get_user_by_id(self.user_id)
+        if user is None:
+            return
+        logger.debug(f"[Activity] User: {user.name} | Episode: {item}")
+        plex.change_default_tracks_if_needed(user.name, item)

plex_auto_languages/alerts/base.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from plex_auto_languages.plex_server import PlexServer
+
+
+class PlexAlert():
+
+    TYPE = None
+
+    def __init__(self, message: dict):
+        self._message = message
+
+    def process(self, plex: PlexServer):
+        raise NotImplementedError

plex_auto_languages/alerts/playing.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+from plexapi.video import Episode
+
+from plex_auto_languages.alerts.base import PlexAlert
+from plex_auto_languages.utils.logger import get_logger
+
+if TYPE_CHECKING:
+    from plex_auto_languages.plex_server import PlexServer
+
+
+logger = get_logger()
+
+
+class PlexPlaying(PlexAlert):
+
+    TYPE = "playing"
+
+    @property
+    def client_identifier(self):
+        return self._message.get("clientIdentifier", None)
+
+    @property
+    def item_key(self):
+        return self._message.get("key", None)
+
+    @property
+    def session_key(self):
+        return self._message.get("sessionKey", None)
+
+    @property
+    def session_state(self):
+        return self._message.get("state", None)
+
+    def process(self, plex: PlexServer):
+        # Get User id and user's Plex instance
+        if self.client_identifier not in plex.cache.user_clients:
+            plex.cache.user_clients[self.client_identifier] = plex.get_user_from_client_identifier(self.client_identifier)
+        user_id, username = plex.cache.user_clients[self.client_identifier]
+        if user_id is None:
+            return
+        user_plex = plex.get_plex_instance_of_user(user_id)
+        if user_plex is None:
+            return
+
+        # Skip if not an Episode
+        item = user_plex.fetch_item(self.item_key)
+        if item is None or not isinstance(item, Episode):
+            return
+
+        # Skip is the session state is unchanged
+        if self.session_key in plex.cache.session_states and plex.cache.session_states[self.session_key] == self.session_state:
+            return
+        logger.debug(f"[Play Session] "
+                     f"Session: {self.session_key} | State: '{self.session_state}' | User id: {user_id} | Episode: {item}")
+        plex.cache.session_states[self.session_key] = self.session_state
+
+        # Reset cache if the session is stopped
+        if self.session_state == "stopped":
+            logger.debug(f"[Play Session] End of session {self.session_key} for user {user_id}")
+            del plex.cache.session_states[self.session_key]
+            del plex.cache.user_clients[self.client_identifier]
+
+        # Skip if selected streams are unchanged
+        item.reload()
+        audio_stream, subtitle_stream = plex.get_selected_streams(item)
+        pair_id = (
+            audio_stream.id if audio_stream is not None else None,
+            subtitle_stream.id if subtitle_stream is not None else None
+        )
+        if item.key in plex.cache.default_streams and plex.cache.default_streams[item.key] == pair_id:
+            return
+        plex.cache.default_streams.setdefault(item.key, pair_id)
+
+        # Change tracks if needed
+        plex.change_default_tracks_if_needed(username, item)

plex_auto_languages/alerts/status.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+
+from plex_auto_languages.alerts.base import PlexAlert
+from plex_auto_languages.utils.logger import get_logger
+
+if TYPE_CHECKING:
+    from plex_auto_languages.plex_server import PlexServer
+
+
+logger = get_logger()
+
+
+class PlexStatus(PlexAlert):
+
+    TYPE = "status"
+
+    @property
+    def title(self):
+        return self._message.get("title", None)
+
+    def process(self, plex: PlexServer):
+        if self.title != "Library scan complete":
+            return
+        logger.info("[Status] Library scan complete")
+
+        # Get all recently added episodes
+        for section in plex.get_all_show_sections():
+            recent = section.searchEpisodes(filters={"addedAt>>": "5m"})
+            if len(recent) == 0:
+                continue
+            logger.debug(f"[Status] Found {len(recent)} newly added episode(s) in section {section}")
+            for item in recent:
+                # Check if the item has already been processed
+                if item.key in plex.cache.newly_added and plex.cache.newly_added[item.key] == item.addedAt:
+                    continue
+                plex.cache.newly_added[item.key] = item.addedAt
+
+                # Change tracks for all users
+                logger.info(f"[Status] Processing newly added episode {plex.get_episode_short_name(item)}")
+                plex.process_new_episode(item.key)

plex_auto_languages/alerts/timeline.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+from datetime import datetime, timedelta
+from plexapi.video import Episode
+
+from plex_auto_languages.alerts.base import PlexAlert
+from plex_auto_languages.utils.logger import get_logger
+
+if TYPE_CHECKING:
+    from plex_auto_languages.plex_server import PlexServer
+
+
+logger = get_logger()
+
+
+class PlexTimeline(PlexAlert):
+
+    TYPE = "timeline"
+
+    @property
+    def has_metadata_state(self):
+        return "metadataState" in self._message
+
+    @property
+    def has_media_state(self):
+        return "mediaState" in self._message
+
+    @property
+    def item_id(self):
+        return int(self._message.get("itemID", None))
+
+    @property
+    def identifier(self):
+        return self._message.get("identifier", None)
+
+    @property
+    def state(self):
+        return self._message.get("state", None)
+
+    @property
+    def entry_type(self):
+        return self._message.get("type", None)
+
+    def process(self, plex: PlexServer):
+        if self.has_metadata_state or self.has_media_state:
+            return
+        if self.identifier != "com.plexapp.plugins.library" or self.state != 5 or self.entry_type == -1:
+            return
+
+        # Skip if not an Episode
+        item = plex.fetch_item(self.item_id)
+        if item is None or not isinstance(item, Episode):
+            return
+
+        # Check if the item has been added recently
+        if item.addedAt < datetime.now() - timedelta(minutes=5) or \
+                (item.key in plex.cache.newly_added and plex.cache.newly_added[item.key] == item.addedAt):
+            return
+        plex.cache.newly_added[item.key] = item.addedAt
+
+        # Change tracks for all users
+        logger.info(f"[Timeline] Processing newly added episode {plex.get_episode_short_name(item)}")
+        plex.process_new_episode(self.item_id)

plex_auto_languages/plex_alert_handler.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+from plex_auto_languages.alerts import PlexActivity, PlexTimeline, PlexPlaying, PlexStatus
+from plex_auto_languages.utils.logger import get_logger
+
+if TYPE_CHECKING:
+    from plex_auto_languages.plex_server import PlexServer
+
+
+logger = get_logger()
+
+
+class PlexAlertHandler():
+
+    def __init__(self, plex: PlexServer, trigger_on_play: bool, trigger_on_scan: bool, trigger_on_activity: bool):
+        self._plex = plex
+        self._trigger_on_play = trigger_on_play
+        self._trigger_on_scan = trigger_on_scan
+        self._trigger_on_activity = trigger_on_activity
+
+    def __call__(self, message: dict):
+        alert_class = None
+        alert_field = None
+        if self._trigger_on_play and message["type"] == "playing":
+            alert_class = PlexPlaying
+            alert_field = "PlaySessionStateNotification"
+        elif self._trigger_on_activity and message["type"] == "activity":
+            alert_class = PlexActivity
+            alert_field = "ActivityNotification"
+        elif self._trigger_on_scan and message["type"] == "timeline":
+            alert_class = PlexTimeline
+            alert_field = "TimelineEntry"
+        elif self._trigger_on_scan and message["type"] == "status":
+            alert_class = PlexStatus
+            alert_field = "StatusNotification"
+
+        if alert_class is None or alert_field is None or alert_field not in message:
+            return
+
+        for alert_message in message[alert_field]:
+            alert = alert_class(alert_message)
+            try:
+                alert.process(self._plex)
+            except Exception:
+                logger.exception(f"Unable to process {alert_class.TYPE}")
+                logger.debug(message)