from __future__ import annotations import asyncio import json import logging from typing import Any, Dict, Optional import aiohttp from src.config import JellyfinConfig from src.player.models import MediaMetadata, PlaybackState, PlayerStatus from src.player.mpv_controller import MpvController logger = logging.getLogger(__name__) TICKS_PER_SECOND = 10_000_000 class JellyfinClient: """Handles communication with the Jellyfin server, session registration, and remote control commands.""" def __init__(self, config: JellyfinConfig, player: MpvController): self.config = config self.player = player self.session: Optional[aiohttp.ClientSession] = None self.ws: Optional[aiohttp.ClientWebSocketResponse] = None self.auth_token: Optional[str] = config.token self.user_id: Optional[str] = None self._running = False self._ws_task: Optional[asyncio.Task] = None self._report_task: Optional[asyncio.Task] = None @property def auth_header(self) -> str: return ( f'MediaBrowser Client="{self.config.client_name}", ' f'Device="{self.config.device_name}", ' f'DeviceId="{self.config.device_id}", ' f'Version="1.0.0"' ) def _get_headers(self) -> Dict[str, str]: headers = { "X-Emby-Authorization": self.auth_header, "Content-Type": "application/json", } if self.auth_token: headers["X-Emby-Token"] = self.auth_token return headers async def start(self) -> None: """Start the Jellyfin client connection loop.""" if not self.config.server_url or not self.config.username: logger.info("Jellyfin credentials not configured; skipping Jellyfin server connection") return self._running = True self.session = aiohttp.ClientSession(headers=self._get_headers()) self._ws_task = asyncio.create_task(self._connection_manager()) self._report_task = asyncio.create_task(self._progress_reporter()) async def _connection_manager(self) -> None: """Keeps the connection and WebSocket session alive with auto-reconnect.""" while self._running: try: if not self.auth_token: await self._authenticate() if self.auth_token: await self._report_capabilities() await self._listen_websocket() except asyncio.CancelledError: break except Exception as e: logger.error("Jellyfin connection error: %s. Reconnecting in 10s...", e) await asyncio.sleep(10) async def _authenticate(self) -> None: """Authenticate user and obtain access token.""" url = f"{self.config.server_url}/Users/AuthenticateByName" payload = { "Username": self.config.username, "Pw": self.config.password or "", } logger.info("Authenticating to Jellyfin server: %s as %s", self.config.server_url, self.config.username) async with self.session.post(url, json=payload) as resp: if resp.status != 200: err_text = await resp.text() raise RuntimeError(f"Authentication failed ({resp.status}): {err_text}") data = await resp.json() self.auth_token = data.get("AccessToken") self.user_id = data.get("User", {}).get("Id") self.session.headers["X-Emby-Token"] = self.auth_token logger.info("Authenticated successfully with Jellyfin (UserId: %s)", self.user_id) async def _report_capabilities(self) -> None: """Notify Jellyfin of client capabilities for remote control.""" url = f"{self.config.server_url}/Sessions/Capabilities/Full" capabilities = { "PlayableMediaTypes": ["Video", "Audio"], "SupportedCommands": [ "Play", "Playstate", "PlayNext", "SetVolume", "Mute", "Unmute", "ToggleMute", "SetAudioStreamIndex", "SetSubtitleStreamIndex", "Seek", "DisplayMessage", ], "SupportsMediaControl": true if True else True, "DeviceProfile": { "Name": "HTPC MPV Player", "DirectPlayProfiles": [ {"Type": "Video"}, {"Type": "Audio"}, ], }, } async with self.session.post(url, json=capabilities) as resp: if resp.status in (200, 204): logger.info("Registered Jellyfin remote control capabilities successfully") else: logger.warning("Failed to register capabilities: HTTP %d", resp.status) async def _listen_websocket(self) -> None: """Maintain WebSocket connection for remote commands from Yatse / Jellyfin Web.""" ws_url = ( f"{self.config.server_url.replace('http://', 'ws://').replace('https://', 'wss://')}" f"/socket?api_key={self.auth_token}&deviceId={self.config.device_id}" ) logger.info("Connecting to Jellyfin WebSocket: %s", ws_url) async with self.session.ws_connect(ws_url) as ws: self.ws = ws logger.info("Jellyfin WebSocket connected; waiting for remote control commands") async for msg in ws: if msg.type == aiohttp.WSMsgType.TEXT: try: payload = json.loads(msg.data) await self._handle_remote_message(payload) except Exception as e: logger.error("Error processing Jellyfin WebSocket message: %s", e) elif msg.type in (aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR): break async def _handle_remote_message(self, msg: Dict[str, Any]) -> None: msg_type = msg.get("MessageType") data = msg.get("Data", {}) logger.debug("Received Jellyfin message: %s", msg_type) if msg_type == "Play": await self._handle_play_command(data) elif msg_type == "Playstate": await self._handle_playstate_command(data) elif msg_type == "GeneralCommand": await self._handle_general_command(data) elif msg_type == "KeepAlive": pass async def _handle_play_command(self, data: Dict[str, Any]) -> None: item_ids = data.get("ItemIds", []) if not item_ids: return item_id = item_ids[0] start_ticks = data.get("StartPositionTicks") or 0 start_seconds = start_ticks / TICKS_PER_SECOND logger.info("Remote play command for Jellyfin item: %s", item_id) # Retrieve item metadata from Jellyfin item_url = f"{self.config.server_url}/Users/{self.user_id}/Items/{item_id}" async with self.session.get(item_url) as resp: if resp.status != 200: logger.error("Failed to fetch item info for %s: %s", item_id, resp.status) return item_data = await resp.json() title = item_data.get("Name", "Unknown") series_name = item_data.get("SeriesName") run_time_ticks = item_data.get("RunTimeTicks") or 0 duration = run_time_ticks / TICKS_PER_SECOND stream_url = ( f"{self.config.server_url}/Videos/{item_id}/stream" f"?static=true&api_key={self.auth_token}" ) metadata = MediaMetadata( url=stream_url, title=title, artist=series_name, album=item_data.get("SeasonName"), duration=duration, thumbnail=f"{self.config.server_url}/Items/{item_id}/Images/Primary", source_type="jellyfin", jellyfin_item_id=item_id, jellyfin_media_source_id=data.get("MediaSourceId"), ) await self.player.play_url(stream_url, metadata=metadata, start_seconds=start_seconds) await self._report_playing(start_ticks) async def _handle_playstate_command(self, data: Dict[str, Any]) -> None: cmd = data.get("Command") logger.info("Jellyfin Playstate command: %s", cmd) if cmd == "PlayPause": await self.player.play_pause() elif cmd == "Pause": await self.player.pause() elif cmd == "Unpause": await self.player.resume() elif cmd == "Stop": await self.player.stop() elif cmd == "Seek": seek_ticks = data.get("SeekPositionTicks") or 0 await self.player.seek(seek_ticks / TICKS_PER_SECOND, absolute=True) async def _handle_general_command(self, data: Dict[str, Any]) -> None: name = data.get("Name") args = data.get("Arguments", {}) logger.info("Jellyfin GeneralCommand: %s (%s)", name, args) if name == "SetVolume": vol = int(args.get("Volume", 100)) await self.player.set_volume(vol) elif name == "Mute": await self.player.set_mute(True) elif name == "Unmute": await self.player.set_mute(False) elif name == "ToggleMute": await self.player.toggle_mute() elif name == "SetAudioStreamIndex": idx = int(args.get("Index", 1)) await self.player.set_audio_track(idx) elif name == "SetSubtitleStreamIndex": idx = int(args.get("Index", 0)) await self.player.set_subtitle_track(idx) elif name == "DisplayMessage": text = args.get("Text", "") await self.player.show_osd_message(text) async def _report_playing(self, position_ticks: int = 0) -> None: if not self.auth_token or not self.session: return meta = self.player.status.metadata if not meta or not meta.jellyfin_item_id: return payload = { "ItemId": meta.jellyfin_item_id, "MediaSourceId": meta.jellyfin_media_source_id or meta.jellyfin_item_id, "PositionTicks": position_ticks, "IsPaused": self.player.status.state == PlaybackState.PAUSED, "CanSeek": True, } url = f"{self.config.server_url}/Sessions/Playing" try: await self.session.post(url, json=payload) except Exception as e: logger.debug("Error reporting playing to Jellyfin: %s", e) async def _report_progress(self) -> None: if not self.auth_token or not self.session: return meta = self.player.status.metadata if not meta or not meta.jellyfin_item_id: return position_ticks = int(self.player.status.current_time * TICKS_PER_SECOND) payload = { "ItemId": meta.jellyfin_item_id, "MediaSourceId": meta.jellyfin_media_source_id or meta.jellyfin_item_id, "PositionTicks": position_ticks, "IsPaused": self.player.status.state == PlaybackState.PAUSED, "CanSeek": True, } url = f"{self.config.server_url}/Sessions/Playing/Progress" try: await self.session.post(url, json=payload) except Exception as e: logger.debug("Error reporting progress to Jellyfin: %s", e) async def _report_stopped(self) -> None: if not self.auth_token or not self.session: return meta = self.player.status.metadata if not meta or not meta.jellyfin_item_id: return position_ticks = int(self.player.status.current_time * TICKS_PER_SECOND) payload = { "ItemId": meta.jellyfin_item_id, "MediaSourceId": meta.jellyfin_media_source_id or meta.jellyfin_item_id, "PositionTicks": position_ticks, } url = f"{self.config.server_url}/Sessions/Playing/Stopped" try: await self.session.post(url, json=payload) except Exception as e: logger.debug("Error reporting stopped to Jellyfin: %s", e) async def _progress_reporter(self) -> None: """Periodically update Jellyfin server with current playback status.""" last_reported_state = self.player.status.state while self._running: try: curr_state = self.player.status.state if curr_state in (PlaybackState.PLAYING, PlaybackState.PAUSED): await self._report_progress() elif last_reported_state in (PlaybackState.PLAYING, PlaybackState.PAUSED) and curr_state == PlaybackState.STOPPED: await self._report_stopped() last_reported_state = curr_state except asyncio.CancelledError: break except Exception as e: logger.debug("Reporter error: %s", e) await asyncio.sleep(4) async def shutdown(self) -> None: self._running = False if self._ws_task: self._ws_task.cancel() if self._report_task: self._report_task.cancel() if self.ws: await self.ws.close() if self.session: await self.session.close()