update rpc from jellyfin

.env.example

@@ -1,5 +0,0 @@
-JELLYFIN_ADDRESS=
-JELLYFIN_USER=
-JELLYFIN_PASSWORD=
-
-DISCORD_APP_ID=

discord.py

@@ -1,9 +0,0 @@
-from pypresence import Presence
-import os
-
-def get_rpc():
-    app_id = os.getenv('DISCORD_APP_ID')
-    RPC = Presence(app_id)
-    RPC.connect()
-    return RPC
-

discord/models.py

@@ -0,0 +1,10 @@
+from pydantic import BaseModel
+from pypresence.types import ActivityType
+
+class DiscordRPCUpdatePayload(BaseModel):
+    id: str
+    title: str
+    subtitle: str
+    image_url: str
+    details: str
+    activity_type: ActivityType

discord/rpc.py

@@ -0,0 +1,34 @@
+from pypresence import Presence
+from settings import settings
+from discord.models import DiscordRPCUpdatePayload
+import logging
+
+class DiscordRPC:
+    def __init__(self):
+        self.logger = logging.getLogger('DiscordRPC')
+
+        self.logger.info("Connecting to Discord RPC...")
+        self.last_update_id = None
+        self.rpc = Presence(settings.discord_app_id)
+        self.rpc.connect()
+        self.logger.info("Connected to Discord RPC.")
+
+    def update(self, payload: DiscordRPCUpdatePayload):
+        if self.last_update_id == payload.id:
+            self.logger.debug("No update needed for Discord RPC presence.")
+            return
+
+        self.logger.info("Updating Discord RPC presence...")
+        self.rpc.update(
+            activity_type=payload.activity_type,
+            details=payload.title,
+            state=payload.subtitle,
+            large_image=payload.image_url,
+            large_text=payload.details
+        )
+        self.logger.info("Discord RPC presence updated.")
+
+    def clear(self):
+        self.logger.info("Clearing Discord RPC presence...")
+        self.rpc.clear()
+        self.logger.info("Discord RPC presence cleared.")

jellyfin.py

@@ -1,45 +0,0 @@
-from jellyfin_apiclient_python import JellyfinClient
-from getmac import get_mac_address
-import os
-
-def get_client():
-    server_address = os.getenv('JELLYFIN_ADDRESS')
-    username = os.getenv('JELLYFIN_USER')
-    password = os.getenv('JELLYFIN_PASSWORD')
-    host_mac = get_mac_address(hostname="localhost")
-
-    client = JellyfinClient()
-    client.config.app('jellydisc', '0.0.1', os.uname().nodename, host_mac)
-    client.config.data['auth.ssl'] = True
-
-    client.auth.connect_to_address(server_address)
-    client.auth.login(server_address, username, password)
-
-    return client 
-
-def get_active_media(client):
-    server_address = os.getenv('JELLYFIN_ADDRESS')
-    sessions = client.jellyfin.get_sessions()
-
-    for session in sessions:
-        # Check if there is a NowPlayingItem
-        media = session.get('NowPlayingItem')
-        if not media:
-            continue
-
-        # Skip non-audio media
-        media_type = media.get('Type')
-        if media_type != 'Audio':
-            continue
-
-        media_id = media.get('Id')
-        image = f"{server_address}/Items/{media_id}/Images/Primary?maxWidth=300&maxHeight=300"
-
-        return {
-            'id': media_id,
-            'artist': media.get('AlbumArtist', 'Unknown Artist'),
-            'title': media.get('Name', 'Unknown Title'),
-            'image': image,
-        }
-
-    return None

jellyfin/api_client.py

@@ -0,0 +1,98 @@
+from settings import settings
+from jellyfin_apiclient_python import JellyfinClient
+from getmac import get_mac_address
+from jellyfin.models import JellyfinMediaItem, JellyfinMediaType
+import logging
+import os
+
+class JellyfinApiClient:
+    def __init__(self):
+        machine_name = os.uname().nodename
+        unique_id = get_mac_address(hostname="localhost")
+
+        self.logger = logging.getLogger('JellyfinApiClient')
+
+        self.logger.info("Connecting to Jellyfin server...")
+        self.client = JellyfinClient()
+        self.client.config.app('jellydisc', '0.0.1', machine_name, unique_id)
+        self.client.config.data['auth.ssl'] = True
+
+        self.client.auth.connect_to_address(settings.jellyfin_server_url)
+        self.client.auth.login(
+            settings.jellyfin_server_url,
+            settings.jellyfin_username,
+            settings.jellyfin_password
+        )
+
+        self.logger.info("Connected to Jellyfin server.")
+
+    def get_current_playback(self) -> JellyfinMediaItem | None:
+        self.logger.info("Fetching current playback information...")
+        sessions = self.client.jellyfin.get_sessions()
+
+        if not sessions:
+            self.logger.info("No active playback found.")
+            return None
+
+        for session in sessions:
+            if session.get('NowPlayingItem'):
+                self.logger.info("Current playback information fetched.")
+                return self.to_model(session['NowPlayingItem'])
+
+        self.logger.info("No active playback found.")
+        return None
+
+    def get_image_url(self, media_id: str) -> str:
+        server_address = settings.jellyfin_server_url.rstrip('/')
+        return f"{server_address}/Items/{media_id}/Images/Primary?maxWidth=300&maxHeight=300"
+
+    def to_model(self, item: dict) -> JellyfinMediaItem:
+        media_type = item.get('Type')
+
+        if media_type == 'Audio':
+            return self.to_music_model(item)
+        elif media_type == 'Episode':
+            return self.to_episode_model(item)
+        elif media_type == 'Movie':
+            return self.to_movie_model(item)
+
+        raise ValueError(f"Unsupported media type: {media_type}")
+
+    def to_music_model(self, item: dict) -> JellyfinMediaItem:
+        media_id = item.get('Id')
+
+        return JellyfinMediaItem(
+            id=media_id,
+            name=item.get('Name'),
+            type=JellyfinMediaType.AUDIO,
+            image_url=self.get_image_url(media_id),
+            metadata={
+                'artist': item.get('AlbumArtist'),
+            }
+        )
+
+    def to_movie_model(self, item: dict) -> JellyfinMediaItem:
+        media_id = item.get('Id')
+
+        return JellyfinMediaItem(
+            id=media_id,
+            name=item.get('Name'),
+            type=JellyfinMediaType.MOVIE,
+            image_url=self.get_image_url(media_id),
+            metadata={}
+        )
+
+    def to_episode_model(self, item: dict) -> JellyfinMediaItem:
+        media_id = item.get('Id')
+
+        return JellyfinMediaItem(
+            id=media_id,
+            name=item.get('Name'),
+            type=JellyfinMediaType.EPISODE,
+            image_url=self.get_image_url(media_id),
+            metadata={
+                'series': item.get('SeriesName'),
+                'season': item.get('ParentIndexNumber'),
+                'episode': item.get('IndexNumber'),
+            }
+        )

jellyfin/models.py

@@ -0,0 +1,14 @@
+from pydantic import BaseModel
+from enum import Enum
+
+class JellyfinMediaType(str, Enum):
+    AUDIO = 'Audio'
+    MOVIE = 'Movie'
+    EPISODE = 'Episode'
+
+class JellyfinMediaItem(BaseModel):
+    id: str
+    name: str
+    type: JellyfinMediaType
+    image_url: str
+    metadata: dict

jellyfin/utils.py

@@ -0,0 +1,38 @@
+from discord.models import DiscordRPCUpdatePayload
+from jellyfin.models import JellyfinMediaItem, JellyfinMediaType
+from pypresence.types import ActivityType
+
+def to_rpc_payload(media_item: JellyfinMediaItem) -> DiscordRPCUpdatePayload:
+    if media_item.type == JellyfinMediaType.AUDIO:
+        return DiscordRPCUpdatePayload(
+            id=media_item.id,
+            title=f"Listening to {media_item.name}",
+            subtitle=f"by {media_item.metadata.get('artist', 'Unknown Artist')}",
+            image_url=media_item.image_url,
+            details=media_item.name,
+            activity_type=ActivityType.LISTENING
+        )
+    elif media_item.type == JellyfinMediaType.MOVIE:
+        return DiscordRPCUpdatePayload(
+            id=media_item.id,
+            title=f"Watching {media_item.name}",
+            subtitle="🍿",
+            image_url=media_item.image_url,
+            details=media_item.name,
+            activity_type=ActivityType.WATCHING
+        )
+    elif media_item.type == JellyfinMediaType.EPISODE:
+        series_name = media_item.metadata.get('series', 'Unknown Series')
+        season = media_item.metadata.get('season', '?')
+        episode = media_item.metadata.get('episode', '?')
+
+        subtitle = f"S{season:02}E{episode:02} of {series_name}"
+
+        return DiscordRPCUpdatePayload(
+            id=media_item.id,
+            title=f"Watching {media_item.name}",
+            subtitle=subtitle,
+            image_url=media_item.image_url,
+            details=media_item.name,
+            activity_type=ActivityType.WATCHING
+        )

main.py

@@ -1,46 +1,26 @@
-from dotenv import load_dotenv
+from discord.rpc import DiscordRPC
-from jellyfin import get_client, get_active_media
+from jellyfin.api_client import JellyfinApiClient
-from discord import get_rpc
+from jellyfin.utils import to_rpc_payload
-from pypresence.types import ActivityType
+import logging
 import time
 
-load_dotenv()
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
 
-rpc = get_rpc()
+discordRPC = DiscordRPC()
-client = get_client()
+jellyfinApiClient = JellyfinApiClient()
 
-def generate_media_id(media):
+def main():
-    return f"{media['artist']}-{media['title']}"
+    while True:
-
+        media_item = jellyfinApiClient.get_current_playback()
-def main_loop(last_media_id):
+        if not media_item:
-    media = get_active_media(client)
+            discordRPC.clear()
-
+            time.sleep(15)
-    if media is None:
+            continue
-        print("No active media found.")
-        rpc.clear()
-        return None
-
-    media_id = generate_media_id(media)
-
-    if media_id != last_media_id:
-        print(f"Updating Discord RPC: Listening to {media['title']} by {media['artist']}")
-
-        rpc.update(
-            activity_type=ActivityType.LISTENING,
-            state=f"by {media['artist']}",
-            details=f"Listening to {media['title']}",
-            large_image=media['image'],
-            large_text=media['title'],
-        )
-        return media_id
-    else:
-        print("No change in media. Skipping update.")
-        return last_media_id
 
+        discordRPC.update(to_rpc_payload(media_item))
+        time.sleep(15)
 
 if __name__ == "__main__":
-    last_media_id = None
+    main()
-    print("Starting Jellyfin Discord Rich Presence...")
-    while True:
-        last_media_id = main_loop(last_media_id)
-        time.sleep(15)

requirements.txt

@@ -1,4 +1,5 @@
 jellyfin-apiclient-python==1.11.0
-python-dotenv==1.2.1
 getmac==0.9.5
 pypresence==4.6.1
+pydantic==2.12.5
+pydantic-settings==2.12.0

settings.py

@@ -0,0 +1,13 @@
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+class Settings(BaseSettings):
+    jellyfin_server_url: str = Field(..., env="JELLYFIN_SERVER_URL")
+    jellyfin_username: str = Field(..., env="JELLYFIN_USERNAME")
+    jellyfin_password: str = Field(..., env="JELLYFIN_PASSWORD")
+
+    discord_app_id: str = Field(..., env="DISCORD_APP_ID")
+
+    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
+
+settings = Settings()