danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

MS Gap 1: Lift shared TrackInfo + PlaylistInfo to neutral types module

Browse source 
Plex / Jellyfin / Navidrome each defined a near-identical
XTrackInfo (id / title / artist / album / duration / track_number /
year / rating) and XPlaylistInfo (id / title / description /
duration / leaf_count / tracks). Three classes that grew up by
copy-paste — not a real contract difference.

Lifted both to core/media_server/types.py as canonical TrackInfo +
PlaylistInfo. Per-server constructors live as classmethods on the
unified class (TrackInfo.from_plex_track, PlaylistInfo.from_plex_playlist)
matching the metadata Album.from_X_dict pattern Cin's POC uses.
Heavy plexapi imports stay lazy under TYPE_CHECKING.

- core/plex_client.py / jellyfin_client.py / navidrome_client.py:
  per-server XTrackInfo / XPlaylistInfo dataclass definitions
  removed; each module now imports TrackInfo + PlaylistInfo from
  the neutral package and uses the shared name internally.
- core/matching_engine.py: was annotating callers with PlexTrackInfo
  even though sync_service hands it Jellyfin / Navidrome instances
  at runtime when those servers are active. Annotation is now the
  unified TrackInfo, so signatures match the actual contract.
- services/sync_service.py: same import + annotation update.
This commit is contained in:
Broque Thomas 2026-05-05 18:25:28 -07:00
parent a6bb5f5b43
commit 2ebaf2e6e3
6 changed files with 177 additions and 130 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
core/jellyfin_client.py
View file

@ -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:

8
core/matching_engine.py
View file

@ -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

130
core/media_server/types.py Normal file
View file

@ -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,
)

38
core/navidrome_client.py
View file

@ -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', ''),

94
core/plex_client.py
View file

@ -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):

5
services/sync_service.py
View file

@ -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