Compare commits
2 Commits
b91e0e1002
...
aa2e450454
| Author | SHA1 | Date | |
|---|---|---|---|
|
aa2e450454
|
|||
|
4a33281ca4
|
@@ -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
|
||||
|
||||
@@ -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.")
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user