224 lines
7.7 KiB
Python
224 lines
7.7 KiB
Python
from settings import settings
|
|
from jellyfin_apiclient_python import JellyfinClient
|
|
from jellyfin.models import JellyfinMediaItem, JellyfinMediaType, JellyfinMusicMediaMetadata, JellyfinMovieMediaMetadata, JellyfinEpisodeMediaMetadata
|
|
from jellyfin.utils import JellyfinUtils
|
|
from datetime import datetime
|
|
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()
|
|
JellyfinUtils.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 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':
|
|
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 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')
|
|
|
|
if is_paused:
|
|
return (None, None)
|
|
|
|
runtime_ticks = media.get('RunTimeTicks')
|
|
total_runtime_seconds = runtime_ticks // 10_000_000
|
|
|
|
playback_position_ticks = play_state.get('PositionTicks')
|
|
playback_position_seconds = playback_position_ticks // 10_000_000
|
|
|
|
start = int(datetime.now().timestamp()) - playback_position_seconds
|
|
end = start + total_runtime_seconds
|
|
|
|
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(
|
|
id=media_id,
|
|
name=item.get('Name'),
|
|
type=JellyfinMediaType.AUDIO,
|
|
image_url=self.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) = self.get_playback_info(item)
|
|
|
|
return JellyfinMediaItem(
|
|
id=media_id,
|
|
name=item.get('Name'),
|
|
type=JellyfinMediaType.MOVIE,
|
|
image_url=self.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')
|
|
|
|
metadata = JellyfinEpisodeMediaMetadata(
|
|
subtitle=f"S{
|
|
season_number:02}E{
|
|
episode_number:02} of {series_name}")
|
|
|
|
(start, end) = self.get_playback_info(item)
|
|
|
|
return JellyfinMediaItem(
|
|
id=media_id,
|
|
name=item.get('Name'),
|
|
type=JellyfinMediaType.EPISODE,
|
|
image_url=self.get_image_url(parent_id),
|
|
start=start,
|
|
end=end,
|
|
metadata=metadata
|
|
)
|