Compare commits

..

1 Commits

Author SHA1 Message Date
b91e0e1002 improve typing 2025-12-10 03:36:34 +01:00
5 changed files with 23 additions and 131 deletions

View File

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

View File

@@ -1,23 +1,15 @@
from settings import settings from settings import settings
from jellyfin_apiclient_python import JellyfinClient from jellyfin_apiclient_python import JellyfinClient
from getmac import get_mac_address from getmac import get_mac_address
from jellyfin.models import JellyfinMediaItem, JellyfinMediaType, JellyfinMusicMediaMetadata, JellyfinMovieMediaMetadata, JellyfinEpisodeMediaMetadata from jellyfin.models import JellyfinMediaItem, JellyfinMediaType
from datetime import datetime from datetime import datetime
from typing import Optional, Tuple
import logging import logging
import time import time
import os import os
class JellyfinApiClient: class JellyfinApiClient:
"""
Client for interacting with the Jellyfin server API.
"""
def __init__(self): def __init__(self):
"""
Initializes the Jellyfin API client and authenticates with the server.
"""
machine_name = os.uname().nodename machine_name = os.uname().nodename
unique_id = get_mac_address(hostname="localhost") unique_id = get_mac_address(hostname="localhost")
@@ -34,9 +26,6 @@ class JellyfinApiClient:
self.logger.info("Connected to Jellyfin server.") self.logger.info("Connected to Jellyfin server.")
def authenticate(self): def authenticate(self):
"""
Authenticates with the Jellyfin server.
"""
self.logger.info("Authenticating with Jellyfin server...") self.logger.info("Authenticating with Jellyfin server...")
self.client.auth.connect_to_address(settings.jellyfin_server_url) self.client.auth.connect_to_address(settings.jellyfin_server_url)
self.client.auth.login( self.client.auth.login(
@@ -48,12 +37,6 @@ class JellyfinApiClient:
self.logger.info("Authenticated with Jellyfin server.") self.logger.info("Authenticated with Jellyfin server.")
def get_current_playback(self) -> JellyfinMediaItem | None: 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: if time.time() - self.last_auth_time > settings.jellyfin_auth_timeout:
self.authenticate() self.authenticate()
@@ -77,26 +60,10 @@ class JellyfinApiClient:
return None return None
def get_image_url(self, media_id: str) -> str: 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('/') server_address = settings.jellyfin_server_url.rstrip('/')
return f"{server_address}/Items/{media_id}/Images/Primary?maxWidth=300&maxHeight=300" return f"{server_address}/Items/{media_id}/Images/Primary?maxWidth=300&maxHeight=300"
def to_model(self, item: dict) -> JellyfinMediaItem: 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') media_type = item.get('Type')
if media_type == 'Audio': if media_type == 'Audio':
@@ -106,16 +73,7 @@ class JellyfinApiClient:
elif media_type == 'Movie': elif media_type == 'Movie':
return self.to_movie_model(item) return self.to_movie_model(item)
def get_playback_info( def get_playback_info(self, media: dict) -> tuple[int, int]:
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') play_state = media.get('PlayState')
is_paused = play_state.get('IsPaused') is_paused = play_state.get('IsPaused')
@@ -134,24 +92,12 @@ class JellyfinApiClient:
return (start, end) return (start, end)
def to_music_model(self, item: dict) -> JellyfinMediaItem: 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') media_id = item.get('Id')
parent_id = item.get('ParentId') parent_id = item.get('ParentId')
premiere_date = item.get('PremiereDate') premiere_date = item.get('PremiereDate')
premiere_year = datetime.fromisoformat( premiere_year = datetime.fromisoformat(
premiere_date).year if premiere_date else None 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) (start, end) = self.get_playback_info(item)
return JellyfinMediaItem( return JellyfinMediaItem(
@@ -161,24 +107,15 @@ class JellyfinApiClient:
image_url=self.get_image_url(parent_id), image_url=self.get_image_url(parent_id),
start=start, start=start,
end=end, end=end,
metadata=metadata metadata={
) 'artist': item.get('AlbumArtist'),
'album': f"{
item.get('Album')} ({premiere_year})" if premiere_date else item.get('Album')})
def to_movie_model(self, item: dict) -> JellyfinMediaItem: 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') media_id = item.get('Id')
premiere_date = item.get('PremiereDate') 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) (start, end) = self.get_playback_info(item)
return JellyfinMediaItem( return JellyfinMediaItem(
@@ -188,18 +125,12 @@ class JellyfinApiClient:
image_url=self.get_image_url(media_id), image_url=self.get_image_url(media_id),
start=start, start=start,
end=end, end=end,
metadata=metadata metadata={
'date': datetime.fromisoformat(premiere_date).strftime('%d/%m/%Y') if premiere_date else None
}
) )
def to_episode_model(self, item: dict) -> JellyfinMediaItem: 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') media_id = item.get('Id')
parent = self.client.jellyfin.get_item(item.get('ParentId')) parent = self.client.jellyfin.get_item(item.get('ParentId'))
parent_id = parent.get('ParentId') parent_id = parent.get('ParentId')
@@ -207,10 +138,7 @@ class JellyfinApiClient:
season_number = item.get('ParentIndexNumber') season_number = item.get('ParentIndexNumber')
episode_number = item.get('IndexNumber') episode_number = item.get('IndexNumber')
metadata = JellyfinEpisodeMediaMetadata( subtitle = f"S{season_number:02}E{episode_number:02} of {series_name}"
subtitle=f"S{
season_number:02}E{
episode_number:02} of {series_name}")
(start, end) = self.get_playback_info(item) (start, end) = self.get_playback_info(item)
@@ -221,5 +149,7 @@ class JellyfinApiClient:
image_url=self.get_image_url(parent_id), image_url=self.get_image_url(parent_id),
start=start, start=start,
end=end, end=end,
metadata=metadata metadata={
'subtitle': subtitle,
}
) )

View File

@@ -1,6 +1,6 @@
from pydantic import BaseModel from pydantic import BaseModel
from enum import Enum from enum import Enum
from typing import Optional, Union from typing import Optional
class JellyfinMediaType(str, Enum): class JellyfinMediaType(str, Enum):
@@ -9,19 +9,6 @@ class JellyfinMediaType(str, Enum):
EPISODE = 'Episode' 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): class JellyfinMediaItem(BaseModel):
id: str id: str
name: str name: str
@@ -29,6 +16,4 @@ class JellyfinMediaItem(BaseModel):
image_url: str image_url: str
start: Optional[int] start: Optional[int]
end: Optional[int] end: Optional[int]
metadata: Union[JellyfinMusicMediaMetadata, metadata: dict
JellyfinMovieMediaMetadata,
JellyfinEpisodeMediaMetadata]

View File

@@ -4,23 +4,19 @@ from pypresence.types import ActivityType
def to_rpc_payload(media_item: JellyfinMediaItem) -> DiscordRPCUpdatePayload: 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: if media_item.type == JellyfinMediaType.AUDIO:
return DiscordRPCUpdatePayload( return DiscordRPCUpdatePayload(
id=media_item.id, id=media_item.id,
title=f"Listening to { title=f"Listening to {
media_item.name}", media_item.name}",
subtitle=f"by { subtitle=f"by {
media_item.metadata.artist}", media_item.metadata.get(
'artist',
'Unknown Artist')}",
image_url=media_item.image_url, image_url=media_item.image_url,
details=media_item.metadata.album, details=media_item.metadata.get(
'album',
'Unknown Album'),
activity_type=ActivityType.LISTENING, activity_type=ActivityType.LISTENING,
start=media_item.start, start=media_item.start,
end=media_item.end) end=media_item.end)
@@ -28,7 +24,7 @@ def to_rpc_payload(media_item: JellyfinMediaItem) -> DiscordRPCUpdatePayload:
return DiscordRPCUpdatePayload( return DiscordRPCUpdatePayload(
id=media_item.id, id=media_item.id,
title=f"Watching {media_item.name}", title=f"Watching {media_item.name}",
subtitle=media_item.metadata.date, subtitle=media_item.metadata.get('date'),
image_url=media_item.image_url, image_url=media_item.image_url,
details=media_item.name, details=media_item.name,
activity_type=ActivityType.WATCHING, activity_type=ActivityType.WATCHING,
@@ -39,7 +35,7 @@ def to_rpc_payload(media_item: JellyfinMediaItem) -> DiscordRPCUpdatePayload:
return DiscordRPCUpdatePayload( return DiscordRPCUpdatePayload(
id=media_item.id, id=media_item.id,
title=f"Watching {media_item.name}", title=f"Watching {media_item.name}",
subtitle=media_item.metadata.subtitle, subtitle=media_item.metadata.get('subtitle'),
image_url=media_item.image_url, image_url=media_item.image_url,
details=media_item.name, details=media_item.name,
activity_type=ActivityType.WATCHING, activity_type=ActivityType.WATCHING,

View File

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