101 lines
3.4 KiB
Python
101 lines
3.4 KiB
Python
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.image import get_image_url
|
|
from jellyfin.utils.models import episode, movie, audio
|
|
from typing import Optional
|
|
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_media(self) -> Optional[JellyfinMediaItem]:
|
|
"""
|
|
Fetches the current media 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 audio.to_media_item(item)
|
|
elif media_type == 'Episode':
|
|
return episode.to_media_item(item, self.client)
|
|
elif media_type == 'Movie':
|
|
return movie.to_media_item(item)
|