diff --git a/core/jellyfin_client.py b/core/jellyfin_client.py index ee564f6c..2439e1cf 100644 --- a/core/jellyfin_client.py +++ b/core/jellyfin_client.py @@ -1,33 +1,19 @@ import requests import time from typing import List, Optional, Dict, Any -from dataclasses import dataclass from datetime import datetime import json from utils.logging_config import get_logger from config.settings import config_manager +# Shared dataclasses live in the neutral media_server package — every +# server client used to define a near-identical XTrackInfo / +# XPlaylistInfo. Lifted to one canonical type so consumers (matching +# engine, sync service) get a single import. +from core.media_server.types import TrackInfo, PlaylistInfo + logger = get_logger("jellyfin_client") -@dataclass -class JellyfinTrackInfo: - id: str - title: str - artist: str - album: str - duration: int - track_number: Optional[int] = None - year: Optional[int] = None - rating: Optional[float] = None - -@dataclass -class JellyfinPlaylistInfo: - id: str - title: str - description: Optional[str] - duration: int - leaf_count: int - tracks: List[JellyfinTrackInfo] class JellyfinArtist: """Wrapper class to mimic Plex artist object interface""" @@ -1204,7 +1190,7 @@ class JellyfinClient(MediaServerClient): return stats - def get_all_playlists(self) -> List[JellyfinPlaylistInfo]: + def get_all_playlists(self) -> List[PlaylistInfo]: """Get all playlists from Jellyfin server""" if not self.ensure_connection(): return [] @@ -1221,7 +1207,7 @@ class JellyfinClient(MediaServerClient): playlists = [] for item in response.get('Items', []): - playlist_info = JellyfinPlaylistInfo( + playlist_info = PlaylistInfo( id=item.get('Id', ''), title=item.get('Name', 'Unknown Playlist'), description=item.get('Overview'), @@ -1238,7 +1224,7 @@ class JellyfinClient(MediaServerClient): logger.error(f"Error getting playlists from Jellyfin: {e}") return [] - def get_playlist_by_name(self, name: str) -> Optional[JellyfinPlaylistInfo]: + def get_playlist_by_name(self, name: str) -> Optional[PlaylistInfo]: """Get a specific playlist by name""" playlists = self.get_all_playlists() for playlist in playlists: diff --git a/core/matching_engine.py b/core/matching_engine.py index da1fb9bc..fc76fb18 100644 --- a/core/matching_engine.py +++ b/core/matching_engine.py @@ -7,7 +7,7 @@ from utils.logging_config import get_logger from config.settings import config_manager from core.spotify_client import Track as SpotifyTrack -from core.plex_client import PlexTrackInfo +from core.media_server.types import TrackInfo from core.soulseek_client import TrackResult, AlbumResult @@ -16,7 +16,7 @@ logger = get_logger("matching_engine") @dataclass class MatchResult: spotify_track: SpotifyTrack - plex_track: Optional[PlexTrackInfo] + plex_track: Optional[TrackInfo] confidence: float match_type: str @@ -302,7 +302,7 @@ class MusicMatchingEngine: confidence = (title_score * 0.60) + (artist_score * 0.30) + (duration_score * 0.10) return confidence, "standard_match" - def calculate_match_confidence(self, spotify_track: SpotifyTrack, plex_track: PlexTrackInfo) -> Tuple[float, str]: + def calculate_match_confidence(self, spotify_track: SpotifyTrack, plex_track: TrackInfo) -> Tuple[float, str]: """Calculates a confidence score using a prioritized model, starting with a strict 'core' title check.""" return self.score_track_match( source_title=spotify_track.name, @@ -313,7 +313,7 @@ class MusicMatchingEngine: candidate_duration_ms=plex_track.duration if plex_track.duration else 0 ) - def find_best_match(self, spotify_track: SpotifyTrack, plex_tracks: List[PlexTrackInfo]) -> MatchResult: + def find_best_match(self, spotify_track: SpotifyTrack, plex_tracks: List[TrackInfo]) -> MatchResult: """Finds the best Plex track match from a list of candidates.""" best_match = None best_confidence = 0.0 diff --git a/core/media_server/types.py b/core/media_server/types.py new file mode 100644 index 00000000..7d48e1c5 --- /dev/null +++ b/core/media_server/types.py @@ -0,0 +1,130 @@ +"""Shared dataclasses for the media-server contract. + +Plex / Jellyfin / Navidrome / SoulSync clients all surfaced near- +identical ``XTrackInfo`` and ``XPlaylistInfo`` shapes (id, title, +artist, album, duration, track_number, year, optional rating) +because every consumer needed the same shape downstream. The +per-server class names were a copy-paste artifact, not a real +contract difference. + +This module owns the canonical types. Per-server constructors live +on the unified class as classmethods (``TrackInfo.from_plex_track``, +``TrackInfo.from_jellyfin_dict``, ``TrackInfo.from_navidrome_dict``) +— same pattern Cin used for ``Album.from_spotify_dict`` etc. on the +metadata engine. Heavy server SDK types (``PlexTrack``, dict +shapes) imported under TYPE_CHECKING so this module stays +import-light. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from typing import TYPE_CHECKING, Any, Dict, List, Optional + +if TYPE_CHECKING: + # plexapi types — only loaded when type-checking; runtime + # paths through from_plex_track accept whatever PlexClient + # passes through. + from plexapi.audio import Track as PlexTrack + from plexapi.playlist import Playlist as PlexPlaylist + + +@dataclass +class TrackInfo: + """Canonical track-shape returned by every media server client. + + All four servers (Plex, Jellyfin, Navidrome, SoulSync standalone) + used to define their own near-identical ``XTrackInfo`` dataclass. + Lifted to one canonical type here so consumers (matching engine, + sync service, library scanners) get a single import. + """ + + id: str + title: str + artist: str + album: str + duration: int + track_number: Optional[int] = None + year: Optional[int] = None + rating: Optional[float] = None + + # ------------------------------------------------------------------ + # Per-server constructors — mirror Cin's metadata Album.from_X_dict + # pattern. Each one knows ONE server's wire shape. + # ------------------------------------------------------------------ + + @classmethod + def from_plex_track(cls, track: 'PlexTrack') -> 'TrackInfo': + """Build a TrackInfo from a plexapi PlexTrack. + + Defensive: tracks may be missing artist or album metadata in + Plex (especially fan-uploaded content); fall back to + "Unknown Artist" / "Unknown Album" instead of raising. + """ + # Imported lazily so this module stays import-light. plexapi + # is heavy and pulls in network + ssl deps just to define + # exception types. + from plexapi.exceptions import NotFound + + try: + artist_title = track.artist().title if track.artist() else "Unknown Artist" + except (NotFound, AttributeError): + artist_title = "Unknown Artist" + + try: + album_title = track.album().title if track.album() else "Unknown Album" + except (NotFound, AttributeError): + album_title = "Unknown Album" + + return cls( + id=str(track.ratingKey), + title=track.title, + artist=artist_title, + album=album_title, + duration=track.duration, + track_number=track.trackNumber, + year=track.year, + rating=track.userRating, + ) + + +@dataclass +class PlaylistInfo: + """Canonical playlist-shape returned by every media server client. + + Same lift rationale as ``TrackInfo`` — every server defined the + same five-field dataclass + a list of tracks. + """ + + id: str + title: str + description: Optional[str] + duration: int + leaf_count: int + tracks: List[TrackInfo] = field(default_factory=list) + + # ------------------------------------------------------------------ + # Per-server constructors + # ------------------------------------------------------------------ + + @classmethod + def from_plex_playlist(cls, playlist: 'PlexPlaylist') -> 'PlaylistInfo': + """Build a PlaylistInfo from a plexapi Playlist. Skips items + that aren't audio tracks (Plex playlists can mix media types + in theory, though music libraries shouldn't).""" + from plexapi.audio import Track as PlexTrack + + tracks: List[TrackInfo] = [] + for item in playlist.items(): + if isinstance(item, PlexTrack): + tracks.append(TrackInfo.from_plex_track(item)) + + return cls( + id=str(playlist.ratingKey), + title=playlist.title, + description=playlist.summary, + duration=playlist.duration, + leaf_count=playlist.leafCount, + tracks=tracks, + ) diff --git a/core/navidrome_client.py b/core/navidrome_client.py index 0e396fe4..5dd29136 100644 --- a/core/navidrome_client.py +++ b/core/navidrome_client.py @@ -2,33 +2,19 @@ import requests import hashlib import secrets from typing import List, Optional, Dict, Any -from dataclasses import dataclass from datetime import datetime import json from utils.logging_config import get_logger from config.settings import config_manager +# Shared dataclasses live in the neutral media_server package — every +# server client used to define a near-identical XTrackInfo / +# XPlaylistInfo. Lifted to one canonical type so consumers (matching +# engine, sync service) get a single import. +from core.media_server.types import TrackInfo, PlaylistInfo + logger = get_logger("navidrome_client") -@dataclass -class NavidromeTrackInfo: - id: str - title: str - artist: str - album: str - duration: int - track_number: Optional[int] = None - year: Optional[int] = None - rating: Optional[float] = None - -@dataclass -class NavidromePlaylistInfo: - id: str - title: str - description: Optional[str] - duration: int - leaf_count: int - tracks: List[NavidromeTrackInfo] class NavidromeArtist: """Wrapper class to mimic Plex artist object interface""" @@ -827,7 +813,7 @@ class NavidromeClient(MediaServerClient): return {} return {} - def get_all_playlists(self) -> List[NavidromePlaylistInfo]: + def get_all_playlists(self) -> List[PlaylistInfo]: """Get all playlists from Navidrome server""" if not self.ensure_connection(): return [] @@ -841,7 +827,7 @@ class NavidromeClient(MediaServerClient): playlists_data = response.get('playlists', {}).get('playlist', []) for playlist_data in playlists_data: - playlist_info = NavidromePlaylistInfo( + playlist_info = PlaylistInfo( id=playlist_data.get('id', ''), title=playlist_data.get('name', 'Unknown Playlist'), description=playlist_data.get('comment'), @@ -858,7 +844,7 @@ class NavidromeClient(MediaServerClient): logger.error(f"Error getting playlists from Navidrome: {e}") return [] - def get_playlist_by_name(self, name: str) -> Optional[NavidromePlaylistInfo]: + def get_playlist_by_name(self, name: str) -> Optional[PlaylistInfo]: """Get a specific playlist by name""" playlists = self.get_all_playlists() for playlist in playlists: @@ -979,7 +965,7 @@ class NavidromeClient(MediaServerClient): logger.error(f"Error getting tracks for playlist {playlist_id}: {e}") return [] - def get_playlists_by_name(self, name: str) -> List[NavidromePlaylistInfo]: + def get_playlists_by_name(self, name: str) -> List[PlaylistInfo]: """Get all playlists matching a specific name (case-insensitive)""" matches = [] playlists = self.get_all_playlists() @@ -1143,7 +1129,7 @@ class NavidromeClient(MediaServerClient): self._track_cache.clear() logger.info("Navidrome client cache cleared") - def search_tracks(self, title: str, artist: str, limit: int = 15) -> List[NavidromeTrackInfo]: + def search_tracks(self, title: str, artist: str, limit: int = 15) -> List[TrackInfo]: """Search for tracks using Navidrome search API""" if not self.ensure_connection(): logger.warning("Navidrome not connected. Cannot perform search.") @@ -1169,7 +1155,7 @@ class NavidromeClient(MediaServerClient): search_result = response.get('searchResult3', {}) for track_data in search_result.get('song', []): - track_info = NavidromeTrackInfo( + track_info = TrackInfo( id=track_data.get('id', ''), title=track_data.get('title', ''), artist=track_data.get('artist', ''), diff --git a/core/plex_client.py b/core/plex_client.py index f9c1a24d..1cf771e1 100644 --- a/core/plex_client.py +++ b/core/plex_client.py @@ -4,7 +4,6 @@ from plexapi.audio import Track as PlexTrack, Album as PlexAlbum, Artist as Plex from plexapi.playlist import Playlist as PlexPlaylist from plexapi.exceptions import PlexApiException, NotFound from typing import List, Optional, Dict, Any -from dataclasses import dataclass import requests from datetime import datetime, timedelta import re @@ -12,70 +11,15 @@ from utils.logging_config import get_logger from config.settings import config_manager import threading -logger = get_logger("plex_client") - -@dataclass -class PlexTrackInfo: - id: str - title: str - artist: str - album: str - duration: int - track_number: Optional[int] = None - year: Optional[int] = None - rating: Optional[float] = None - - @classmethod - def from_plex_track(cls, track: PlexTrack) -> 'PlexTrackInfo': - # Gracefully handle tracks that might be missing artist or album metadata in Plex - try: - artist_title = track.artist().title if track.artist() else "Unknown Artist" - except (NotFound, AttributeError): - artist_title = "Unknown Artist" - - try: - album_title = track.album().title if track.album() else "Unknown Album" - except (NotFound, AttributeError): - album_title = "Unknown Album" - - return cls( - id=str(track.ratingKey), - title=track.title, - artist=artist_title, - album=album_title, - duration=track.duration, - track_number=track.trackNumber, - year=track.year, - rating=track.userRating - ) - -@dataclass -class PlexPlaylistInfo: - id: str - title: str - description: Optional[str] - duration: int - leaf_count: int - tracks: List[PlexTrackInfo] - - @classmethod - def from_plex_playlist(cls, playlist: PlexPlaylist) -> 'PlexPlaylistInfo': - tracks = [] - for item in playlist.items(): - if isinstance(item, PlexTrack): - tracks.append(PlexTrackInfo.from_plex_track(item)) - - return cls( - id=str(playlist.ratingKey), - title=playlist.title, - description=playlist.summary, - duration=playlist.duration, - leaf_count=playlist.leafCount, - tracks=tracks - ) - +# Shared dataclasses live in the neutral media_server package now — +# every server client used to define a near-identical XTrackInfo / +# XPlaylistInfo. Lifted to one canonical type so consumers (matching +# engine, sync service, etc.) get a single import. +from core.media_server.types import TrackInfo, PlaylistInfo from core.media_server.contract import MediaServerClient +logger = get_logger("plex_client") + class PlexClient(MediaServerClient): def __init__(self): @@ -271,7 +215,7 @@ class PlexClient(MediaServerClient): """Check if both server is connected AND music library is selected.""" return self.server is not None and self.music_library is not None - def get_all_playlists(self) -> List[PlexPlaylistInfo]: + def get_all_playlists(self) -> List[PlaylistInfo]: if not self.ensure_connection(): logger.error("Not connected to Plex server") return [] @@ -281,7 +225,7 @@ class PlexClient(MediaServerClient): try: for playlist in self.server.playlists(): if getattr(playlist, 'playlistType', None) == 'audio': - playlist_info = PlexPlaylistInfo.from_plex_playlist(playlist) + playlist_info = PlaylistInfo.from_plex_playlist(playlist) playlists.append(playlist_info) logger.info(f"Retrieved {len(playlists)} audio playlists") @@ -291,14 +235,14 @@ class PlexClient(MediaServerClient): logger.error(f"Error fetching playlists: {e}") return [] - def get_playlist_by_name(self, name: str) -> Optional[PlexPlaylistInfo]: + def get_playlist_by_name(self, name: str) -> Optional[PlaylistInfo]: if not self.ensure_connection(): return None try: playlist = self.server.playlist(name) if getattr(playlist, 'playlistType', None) == 'audio': - return PlexPlaylistInfo.from_plex_playlist(playlist) + return PlaylistInfo.from_plex_playlist(playlist) return None except NotFound: @@ -314,14 +258,14 @@ class PlexClient(MediaServerClient): return False try: - # Handle both PlexTrackInfo objects and actual Plex track objects + # Handle both TrackInfo objects and actual Plex track objects plex_tracks = [] for track in tracks: if hasattr(track, 'ratingKey'): # This is already a Plex track object plex_tracks.append(track) elif hasattr(track, '_original_plex_track'): - # This is a PlexTrackInfo object with stored original track reference + # This is a TrackInfo object with stored original track reference original_track = track._original_plex_track if original_track is not None: plex_tracks.append(original_track) @@ -329,7 +273,7 @@ class PlexClient(MediaServerClient): else: logger.warning(f"Stored track reference is None for: {track.title} by {track.artist}") elif hasattr(track, 'title'): - # Fallback: This is a PlexTrackInfo object, need to find the actual track + # Fallback: This is a TrackInfo object, need to find the actual track plex_track = self._find_track(track.title, track.artist, track.album) if plex_track: plex_tracks.append(plex_track) @@ -451,7 +395,7 @@ class PlexClient(MediaServerClient): logger.error(f"Error copying playlist '{source_name}' to '{target_name}': {e}") return False - def update_playlist(self, playlist_name: str, tracks: List[PlexTrackInfo]) -> bool: + def update_playlist(self, playlist_name: str, tracks: List[TrackInfo]) -> bool: if not self.ensure_connection(): return False @@ -522,7 +466,7 @@ class PlexClient(MediaServerClient): logger.error(f"Error searching for track '{title}' by '{artist}': {e}") return None - def search_tracks(self, title: str, artist: str, limit: int = 15) -> List[PlexTrackInfo]: + def search_tracks(self, title: str, artist: str, limit: int = 15) -> List[TrackInfo]: """ Searches for tracks using an efficient, multi-stage "early exit" strategy. It stops and returns results as soon as candidates are found. @@ -557,7 +501,7 @@ class PlexClient(MediaServerClient): # --- Early Exit: If Stage 1 found results, stop here --- if candidate_tracks: logger.info(f"Found {len(candidate_tracks)} candidates in Stage 1. Exiting early.") - tracks = [PlexTrackInfo.from_plex_track(track) for track in candidate_tracks[:limit]] + tracks = [TrackInfo.from_plex_track(track) for track in candidate_tracks[:limit]] # Store references to original tracks for playlist creation for i, track_info in enumerate(tracks): if i < len(candidate_tracks): @@ -576,7 +520,7 @@ class PlexClient(MediaServerClient): # --- Early Exit: If Stage 2 found results, stop here --- if candidate_tracks: logger.info(f"Found {len(candidate_tracks)} candidates in Stage 2. Exiting early.") - tracks = [PlexTrackInfo.from_plex_track(track) for track in candidate_tracks[:limit]] + tracks = [TrackInfo.from_plex_track(track) for track in candidate_tracks[:limit]] # Store references to original tracks for playlist creation for i, track_info in enumerate(tracks): if i < len(candidate_tracks): @@ -590,7 +534,7 @@ class PlexClient(MediaServerClient): # Removed to prevent false positives where tracks with same title # but different artists are incorrectly matched - tracks = [PlexTrackInfo.from_plex_track(track) for track in candidate_tracks[:limit]] + tracks = [TrackInfo.from_plex_track(track) for track in candidate_tracks[:limit]] # Store references to original tracks for playlist creation for i, track_info in enumerate(tracks): diff --git a/services/sync_service.py b/services/sync_service.py index 119aff87..39fb0930 100644 --- a/services/sync_service.py +++ b/services/sync_service.py @@ -4,7 +4,8 @@ from dataclasses import dataclass from datetime import datetime from utils.logging_config import get_logger from core.spotify_client import SpotifyClient, Playlist as SpotifyPlaylist, Track as SpotifyTrack -from core.plex_client import PlexClient, PlexTrackInfo +from core.plex_client import PlexClient +from core.media_server.types import TrackInfo from core.jellyfin_client import JellyfinClient from core.navidrome_client import NavidromeClient from core.soulseek_client import SoulseekClient @@ -456,7 +457,7 @@ class PlaylistSyncService: self.clear_progress_callback(playlist.name) self._cancelled = False - async def _find_track_in_media_server(self, spotify_track: SpotifyTrack) -> Tuple[Optional[PlexTrackInfo], float]: + async def _find_track_in_media_server(self, spotify_track: SpotifyTrack) -> Tuple[Optional[TrackInfo], float]: """Find a track using the same improved database matching as Download Missing Tracks modal""" try: # Check active media server connection