Compare commits

...

2 Commits

Author SHA1 Message Date
aa2e450454 add documentation 2025-12-10 03:45:39 +01:00
4a33281ca4 improve typing 2025-12-10 03:41:11 +01:00
6 changed files with 136 additions and 26 deletions

View File

@@ -1,5 +1,6 @@
from pydantic import BaseModel
from pypresence.types import ActivityType
from typing import Optional
class DiscordRPCUpdatePayload(BaseModel):
@@ -8,6 +9,6 @@ class DiscordRPCUpdatePayload(BaseModel):
subtitle: str
image_url: str
details: str
start: int | None
end: int | None
start: Optional[int]
end: Optional[int]
activity_type: ActivityType

View File

@@ -5,7 +5,14 @@ import logging
class DiscordRPC:
"""
Client for interacting with Discord Rich Presence (RPC).
"""
def __init__(self):
"""
Initializes the Discord RPC client and connects to Discord via IPC.
"""
self.logger = logging.getLogger('DiscordRPC')
self.logger.info("Connecting to Discord RPC...")
@@ -14,6 +21,12 @@ class DiscordRPC:
self.logger.info("Connected to Discord RPC.")
def update(self, payload: DiscordRPCUpdatePayload):
"""
Updates the Discord RPC presence with the provided payload.
Args:
payload (DiscordRPCUpdatePayload): The payload containing presence information.
"""
self.logger.info("Updating Discord RPC presence...")
self.rpc.update(
activity_type=payload.activity_type,
@@ -27,6 +40,9 @@ class DiscordRPC:
self.logger.info("Discord RPC presence updated.")
def clear(self):
"""
Clears the Discord RPC presence.
"""
self.logger.info("Clearing Discord RPC presence...")
self.rpc.clear()
self.logger.info("Discord RPC presence cleared.")

View File

@@ -1,15 +1,23 @@
from settings import settings
from jellyfin_apiclient_python import JellyfinClient
from getmac import get_mac_address
from jellyfin.models import JellyfinMediaItem, JellyfinMediaType
from jellyfin.models import JellyfinMediaItem, JellyfinMediaType, JellyfinMusicMediaMetadata, JellyfinMovieMediaMetadata, JellyfinEpisodeMediaMetadata
from datetime import datetime
from typing import Optional, Tuple
import logging
import time
import os
class JellyfinApiClient:
"""
Client for interacting with the Jellyfin server API.
"""
def __init__(self):
"""
Initializes the Jellyfin API client and authenticates with the server.
"""
machine_name = os.uname().nodename
unique_id = get_mac_address(hostname="localhost")
@@ -26,6 +34,9 @@ class JellyfinApiClient:
self.logger.info("Connected to Jellyfin server.")
def authenticate(self):
"""
Authenticates with the Jellyfin server.
"""
self.logger.info("Authenticating with Jellyfin server...")
self.client.auth.connect_to_address(settings.jellyfin_server_url)
self.client.auth.login(
@@ -37,6 +48,12 @@ class JellyfinApiClient:
self.logger.info("Authenticated with Jellyfin server.")
def get_current_playback(self) -> JellyfinMediaItem | None:
"""
Fetches the current playback information from the Jellyfin server.
Returns:
JellyfinMediaItem | None: The current playback media item or None if no active playback is found.
"""
if time.time() - self.last_auth_time > settings.jellyfin_auth_timeout:
self.authenticate()
@@ -60,10 +77,26 @@ class JellyfinApiClient:
return None
def get_image_url(self, media_id: str) -> str:
"""
Constructs the image URL for a given media item.
Args:
media_id (str): The ID of the media item.
Returns:
str: The constructed image URL.
"""
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:
"""
Converts a Jellyfin media item dictionary to a JellyfinMediaItem model.
Args:
item (dict): The Jellyfin media item dictionary.
Returns:
JellyfinMediaItem: The converted JellyfinMediaItem model.
"""
media_type = item.get('Type')
if media_type == 'Audio':
@@ -73,7 +106,16 @@ class JellyfinApiClient:
elif media_type == 'Movie':
return self.to_movie_model(item)
def get_playback_info(self, media: dict) -> tuple[int, int]:
def get_playback_info(
self, media: dict) -> Tuple[Optional[int], Optional[int]]:
"""
Extracts playback start and end timestamps from a media item.
Args:
media (dict): The Jellyfin media item dictionary.
Returns:
Tuple[Optional[int], Optional[int]]: A tuple containing the start and end timestamps in seconds, or (None, None) if the media is paused.
"""
play_state = media.get('PlayState')
is_paused = play_state.get('IsPaused')
@@ -92,12 +134,24 @@ class JellyfinApiClient:
return (start, end)
def to_music_model(self, item: dict) -> JellyfinMediaItem:
"""
Converts a Jellyfin music media item dictionary to a JellyfinMediaItem model.
Args:
item (dict): The Jellyfin music media item dictionary.
Returns:
JellyfinMediaItem: The converted JellyfinMediaItem model.
"""
media_id = item.get('Id')
parent_id = item.get('ParentId')
premiere_date = item.get('PremiereDate')
premiere_year = datetime.fromisoformat(
premiere_date).year if premiere_date else None
metadata = JellyfinMusicMediaMetadata(
artist=item.get('AlbumArtist'), album=f"{
item.get('Album')} ({premiere_year})" if premiere_date else item.get('Album'))
(start, end) = self.get_playback_info(item)
return JellyfinMediaItem(
@@ -107,15 +161,24 @@ class JellyfinApiClient:
image_url=self.get_image_url(parent_id),
start=start,
end=end,
metadata={
'artist': item.get('AlbumArtist'),
'album': f"{
item.get('Album')} ({premiere_year})" if premiere_date else item.get('Album')})
metadata=metadata
)
def to_movie_model(self, item: dict) -> JellyfinMediaItem:
"""
Converts a Jellyfin movie media item dictionary to a JellyfinMediaItem model.
Args:
item (dict): The Jellyfin movie media item dictionary.
Returns:
JellyfinMediaItem: The converted JellyfinMediaItem model.
"""
media_id = item.get('Id')
premiere_date = item.get('PremiereDate')
metadata = JellyfinMovieMediaMetadata(date=datetime.fromisoformat(
premiere_date).strftime('%d/%m/%Y') if premiere_date else None)
(start, end) = self.get_playback_info(item)
return JellyfinMediaItem(
@@ -125,12 +188,18 @@ class JellyfinApiClient:
image_url=self.get_image_url(media_id),
start=start,
end=end,
metadata={
'date': datetime.fromisoformat(premiere_date).strftime('%d/%m/%Y') if premiere_date else None
}
metadata=metadata
)
def to_episode_model(self, item: dict) -> JellyfinMediaItem:
"""
Converts a Jellyfin episode media item dictionary to a JellyfinMediaItem model.
Args:
item (dict): The Jellyfin episode media item dictionary.
Returns:
JellyfinMediaItem: The converted JellyfinMediaItem model.
"""
media_id = item.get('Id')
parent = self.client.jellyfin.get_item(item.get('ParentId'))
parent_id = parent.get('ParentId')
@@ -138,7 +207,10 @@ class JellyfinApiClient:
season_number = item.get('ParentIndexNumber')
episode_number = item.get('IndexNumber')
subtitle = f"S{season_number:02}E{episode_number:02} of {series_name}"
metadata = JellyfinEpisodeMediaMetadata(
subtitle=f"S{
season_number:02}E{
episode_number:02} of {series_name}")
(start, end) = self.get_playback_info(item)
@@ -149,7 +221,5 @@ class JellyfinApiClient:
image_url=self.get_image_url(parent_id),
start=start,
end=end,
metadata={
'subtitle': subtitle,
}
metadata=metadata
)

View File

@@ -1,5 +1,6 @@
from pydantic import BaseModel
from enum import Enum
from typing import Optional, Union
class JellyfinMediaType(str, Enum):
@@ -8,11 +9,26 @@ class JellyfinMediaType(str, Enum):
EPISODE = 'Episode'
class JellyfinMusicMediaMetadata(BaseModel):
artist: Optional[str]
album: Optional[str]
class JellyfinMovieMediaMetadata(BaseModel):
date: Optional[str]
class JellyfinEpisodeMediaMetadata(BaseModel):
subtitle: str
class JellyfinMediaItem(BaseModel):
id: str
name: str
type: JellyfinMediaType
image_url: str
start: int | None
end: int | None
metadata: dict
start: Optional[int]
end: Optional[int]
metadata: Union[JellyfinMusicMediaMetadata,
JellyfinMovieMediaMetadata,
JellyfinEpisodeMediaMetadata]

View File

@@ -4,19 +4,23 @@ from pypresence.types import ActivityType
def to_rpc_payload(media_item: JellyfinMediaItem) -> DiscordRPCUpdatePayload:
"""
Converts a JellyfinMediaItem to a DiscordRPCUpdatePayload.
Args:
media_item (JellyfinMediaItem): The Jellyfin media item to convert.
Returns:
DiscordRPCUpdatePayload: The converted Discord RPC update payload.
"""
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')}",
media_item.metadata.artist}",
image_url=media_item.image_url,
details=media_item.metadata.get(
'album',
'Unknown Album'),
details=media_item.metadata.album,
activity_type=ActivityType.LISTENING,
start=media_item.start,
end=media_item.end)
@@ -24,7 +28,7 @@ def to_rpc_payload(media_item: JellyfinMediaItem) -> DiscordRPCUpdatePayload:
return DiscordRPCUpdatePayload(
id=media_item.id,
title=f"Watching {media_item.name}",
subtitle=media_item.metadata.get('date'),
subtitle=media_item.metadata.date,
image_url=media_item.image_url,
details=media_item.name,
activity_type=ActivityType.WATCHING,
@@ -35,7 +39,7 @@ def to_rpc_payload(media_item: JellyfinMediaItem) -> DiscordRPCUpdatePayload:
return DiscordRPCUpdatePayload(
id=media_item.id,
title=f"Watching {media_item.name}",
subtitle=media_item.metadata.get('subtitle'),
subtitle=media_item.metadata.subtitle,
image_url=media_item.image_url,
details=media_item.name,
activity_type=ActivityType.WATCHING,

View File

@@ -3,6 +3,9 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""
Application settings loaded from environment variables or a .env file.
"""
jellyfin_server_url: str = Field(..., env="JELLYFIN_SERVER_URL")
jellyfin_username: str = Field(..., env="JELLYFIN_USERNAME")
jellyfin_password: str = Field(..., env="JELLYFIN_PASSWORD")