from settings import settings from jellyfin_apiclient_python import JellyfinClient from jellyfin.models import ( JellyfinMediaItem, JellyfinMediaType, JellyfinMusicMediaMetadata, JellyfinMovieMediaMetadata, JellyfinEpisodeMediaMetadata) from jellyfin.utils.config import configure_client from jellyfin.utils.playback import get_current_playback from jellyfin.utils.image import get_image_url from typing import Optional, Tuple import logging import time class JellyfinApiClient: """ Client for interacting with the Jellyfin server API. Attributes: last_auth_time (Optional[float]): Timestamp of the last authentication. logger (logging.Logger): Logger instance for logging messages. """ last_auth_time: Optional[float] = None logger: logging.Logger = logging.getLogger('JellyfinApiClient') def __init__(self): """ Initializes the Jellyfin API client and authenticates with the server. """ self.logger.info("Connecting to Jellyfin server...") self.client = JellyfinClient() configure_client(self.client) self.authenticate() 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( settings.jellyfin_server_url, settings.jellyfin_username, settings.jellyfin_password ) self.last_auth_time = time.time() self.logger.info("Authenticated with Jellyfin server.") def get_current_playback(self) -> Optional[JellyfinMediaItem]: """ Fetches the current playback information from the Jellyfin server. Returns: Optional[JellyfinMediaItem]: 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() 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: session_id = session.get('Id') current_item = self.client.jellyfin.get_now_playing(session_id) if current_item and current_item.get( 'Type') in ['Audio', 'Episode', 'Movie']: self.logger.info("Current playback information fetched.") return self.to_model(current_item) self.logger.info("No active playback found.") return None 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': 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) 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) = get_current_playback(item) return JellyfinMediaItem( id=media_id, name=item.get('Name'), type=JellyfinMediaType.AUDIO, image_url=get_image_url(parent_id), start=start, end=end, 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) = get_current_playback(item) return JellyfinMediaItem( id=media_id, name=item.get('Name'), type=JellyfinMediaType.MOVIE, image_url=get_image_url(media_id), start=start, end=end, 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') series_name = item.get('SeriesName') season_number = item.get('ParentIndexNumber') episode_number = item.get('IndexNumber') name = f"{series_name} S{season_number:02}E{episode_number:02}" metadata = JellyfinEpisodeMediaMetadata( subtitle=item.get('Name') ) (start, end) = get_current_playback(item) return JellyfinMediaItem( id=media_id, name=name, type=JellyfinMediaType.EPISODE, image_url=get_image_url(parent_id), start=start, end=end, metadata=metadata )