feat: add jellyfin & emby support.

.github/workflows/release-please.yaml

@@ -19,24 +19,3 @@ jobs:
         with:
           token: ${{ secrets.RELEASE_PLEASE_TOKEN }}
           release-type: python
-
-      - name: Checkout code
-        if: ${{ steps.release.outputs.release_created }}
-        uses: actions/checkout@v4
-
-      - name: Install deps
-        if: ${{ steps.release.outputs.release_created }}
-        run: sudo apt-get install -y jq curl
-
-      - name: Announce release
-        if: ${{ steps.release.outputs.release_created }}
-        env:
-          DISCORD_WEBHOOK: ${{ secrets.DISCORD_WEBHOOK }}
-          VERSION: ${{ steps.release.outputs.tag_name }}
-        run: |
-          RELEASE_NOTES=$(awk '/^## Version \[${{ env.VERSION }}\]/ {flag=1; next} /^## Version \[/ {flag=0} flag' CHANGELOG.md)
-          ANNOUNCEMENT_BODY="🚀🎉 **New Release: Version ${{ env.VERSION }}**${RELEASE_NOTES}"
-          ESCAPED_BODY=$(echo "$ANNOUNCEMENT_BODY" | jq -Rsa .)
-          curl -H "Content-Type: application/json" \
-               -d "{\"content\": $ESCAPED_BODY, \"flags\": 4}" \
-               $DISCORD_WEBHOOK

src/program/updaters/emby.py

@@ -0,0 +1,105 @@
+"""Emby Updater module"""
+from types import SimpleNamespace
+from typing import Generator
+
+from program.settings.manager import settings_manager
+from program.media.item import MediaItem
+from utils.request import get, post
+from utils.logger import logger
+
+
+class EmbyUpdater:
+    def __init__(self):
+        self.key = "emby"
+        self.initialized = False
+        self.settings = settings_manager.settings.updaters.emby
+        self.initialized = self.validate()
+        if not self.initialized:
+            return
+        logger.success("Emby Updater initialized!")
+
+    def validate(self) -> bool:
+        """Validate Emby library"""
+        if not self.settings.enabled:
+            logger.warning("Emby Updater is set to disabled.")
+            return False
+        if not self.settings.api_key:
+            logger.error("Emby API key is not set!")
+            return False
+        if not self.settings.url:
+            logger.error("Emby URL is not set!")
+            return False
+        try:
+            response = get(f"{self.settings.url}/Users?api_key={self.settings.api_key}")
+            if response.is_ok:
+                self.initialized = True
+                return True
+        except Exception as e:
+            logger.exception(f"Emby exception thrown: {e}")
+        return False
+
+    def run(self, item: MediaItem) -> Generator[MediaItem, None, None]:
+        """Update Emby library for a single item or a season with its episodes"""
+        items_to_update = []
+
+        if item.type in ["movie", "episode"]:
+            items_to_update = [item]
+        elif item.type == "show":
+            for season in item.seasons:
+                items_to_update += [e for e in season.episodes if e.symlinked and e.update_folder != "updated"]
+        elif item.type == "season":
+            items_to_update = [e for e in item.episodes if e.symlinked and e.update_folder != "updated"]
+
+        if not items_to_update:
+            logger.debug(f"No items to update for {item.log_string}")
+            return
+
+        updated = False
+        updated_episodes = []
+
+        for item_to_update in items_to_update:
+            if self.update_item(item_to_update):
+                updated_episodes.append(item_to_update)
+                updated = True
+
+        if updated:
+            if item.type in ["show", "season"]:
+                if len(updated_episodes) == len(items_to_update):
+                    logger.log("EMBY", f"Updated all episodes for {item.log_string}")
+                else:
+                    updated_episodes_log = ", ".join([str(ep.number) for ep in updated_episodes])
+                    logger.log("EMBY", f"Updated episodes {updated_episodes_log} in {item.log_string}")
+            else:
+                logger.log("EMBY", f"Updated {item.log_string}")
+
+        yield item
+
+
+    def update_item(self, item: MediaItem) -> bool:
+        """Update the Emby item"""
+        if item.symlinked and item.update_folder != "updated" and item.symlink_path:
+            try:
+                response = post(
+                    f"{self.settings.url}/Library/Media/Updated",
+                    json={"Updates": [{"Path": item.symlink_path, "UpdateType": "Created"}]},
+                    params={"api_key": self.settings.api_key},
+                )
+                if response.is_ok:
+                    return True
+            except Exception as e:
+                logger.error(f"Failed to update Emby item: {e}")
+        return False
+
+    # not needed to update, but maybe useful in the future?
+    def get_libraries(self) -> list[SimpleNamespace]:
+        """Get the libraries from Emby"""
+        try:
+            response = get(
+                f"{self.settings.url}/Library/VirtualFolders",
+                params={"api_key": self.settings.api_key},
+            )
+            if response.is_ok and response.data:
+                return response.data
+        except Exception as e:
+            logger.error(f"Failed to get Emby libraries: {e}")
+        return []

src/program/updaters/jellyfin.py

@@ -0,0 +1,106 @@
+"""Jellyfin Updater module"""
+from types import SimpleNamespace
+from typing import Generator
+
+from program.settings.manager import settings_manager
+from program.media.item import MediaItem
+from utils.request import get, post
+from utils.logger import logger
+
+
+class JellyfinUpdater:
+    def __init__(self):
+        self.key = "jellyfin"
+        self.initialized = False
+        self.settings = settings_manager.settings.updaters.jellyfin
+        self.initialized = self.validate()
+        if not self.initialized:
+            return
+        logger.success("Jellyfin Updater initialized!")
+
+    def validate(self) -> bool:
+        """Validate Jellyfin library"""
+        if not self.settings.enabled:
+            logger.warning("Jellyfin Updater is set to disabled.")
+            return False
+        if not self.settings.api_key:
+            logger.error("Jellyfin API key is not set!")
+            return False
+        if not self.settings.url:
+            logger.error("Jellyfin URL is not set!")
+            return False
+
+        try:
+            response = get(f"{self.settings.url}/Users?api_key={self.settings.api_key}")
+            if response.is_ok:
+                self.initialized = True
+                return True
+        except Exception as e:
+            logger.exception(f"Jellyfin exception thrown: {e}")
+        return False
+
+    def run(self, item: MediaItem) -> Generator[MediaItem, None, None]:
+        """Update Jellyfin library for a single item or a season with its episodes"""
+        items_to_update = []
+
+        if item.type in ["movie", "episode"]:
+            items_to_update = [item]
+        elif item.type == "show":
+            for season in item.seasons:
+                items_to_update += [e for e in season.episodes if e.symlinked and e.update_folder != "updated"]
+        elif item.type == "season":
+            items_to_update = [e for e in item.episodes if e.symlinked and e.update_folder != "updated"]
+
+        if not items_to_update:
+            logger.debug(f"No items to update for {item.log_string}")
+            return
+
+        updated = False
+        updated_episodes = []
+
+        for item_to_update in items_to_update:
+            if self.update_item(item_to_update):
+                updated_episodes.append(item_to_update)
+                updated = True
+
+        if updated:
+            if item.type in ["show", "season"]:
+                if len(updated_episodes) == len(items_to_update):
+                    logger.log("JELLYFIN", f"Updated all episodes for {item.log_string}")
+                else:
+                    updated_episodes_log = ", ".join([str(ep.number) for ep in updated_episodes])
+                    logger.log("JELLYFIN", f"Updated episodes {updated_episodes_log} in {item.log_string}")
+            else:
+                logger.log("JELLYFIN", f"Updated {item.log_string}")
+
+        yield item
+
+
+    def update_item(self, item: MediaItem) -> bool:
+        """Update the Jellyfin item"""
+        if item.symlinked and item.update_folder != "updated" and item.symlink_path:
+            try:
+                response = post(
+                    f"{self.settings.url}/Library/Media/Updated",
+                    json={"Updates": [{"Path": item.symlink_path, "UpdateType": "Created"}]},
+                    params={"api_key": self.settings.api_key},
+                )
+                if response.is_ok:
+                    return True
+            except Exception as e:
+                logger.error(f"Failed to update Jellyfin item: {e}")
+        return False
+
+    # not needed to update, but maybe useful in the future?
+    def get_libraries(self) -> list[SimpleNamespace]:
+        """Get the libraries from Jellyfin"""
+        try:
+            response = get(
+                f"{self.settings.url}/Library/VirtualFolders",
+                params={"api_key": self.settings.api_key},
+            )
+            if response.is_ok and response.data:
+                return response.data
+        except Exception as e:
+            logger.error(f"Failed to get Jellyfin libraries: {e}")
+        return []