mirror of https://github.com/Nezreka/SoulSync.git
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
390 lines
16 KiB
390 lines
16 KiB
"""
|
|
Centralized serializers for the SoulSync API v1.
|
|
|
|
All serializers accept a sqlite3.Row, a dict, or a dataclass instance
|
|
and normalize the output to a plain dict. This allows the same serializer
|
|
to be used whether the data comes from raw queries or existing methods.
|
|
"""
|
|
|
|
import json
|
|
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional, Set
|
|
|
|
|
|
def _to_dict(obj) -> dict:
|
|
"""Convert a sqlite3.Row, dataclass, or dict to a plain dict."""
|
|
if isinstance(obj, dict):
|
|
return obj
|
|
if hasattr(obj, "keys"): # sqlite3.Row
|
|
return {k: obj[k] for k in obj.keys()}
|
|
if hasattr(obj, "__dataclass_fields__"):
|
|
from dataclasses import asdict
|
|
return asdict(obj)
|
|
raise TypeError(f"Cannot serialize {type(obj)}")
|
|
|
|
|
|
def _parse_genres(raw) -> list:
|
|
"""Parse genres from JSON string, list, or comma-separated string."""
|
|
if isinstance(raw, list):
|
|
return raw
|
|
if isinstance(raw, str):
|
|
try:
|
|
parsed = json.loads(raw)
|
|
return parsed if isinstance(parsed, list) else []
|
|
except (json.JSONDecodeError, TypeError):
|
|
return [g.strip() for g in raw.split(",") if g.strip()]
|
|
return []
|
|
|
|
|
|
def _isoformat(val) -> Optional[str]:
|
|
"""Safely convert datetime or string to ISO format string."""
|
|
if val is None:
|
|
return None
|
|
if isinstance(val, datetime):
|
|
return val.isoformat()
|
|
if isinstance(val, str):
|
|
return val
|
|
return str(val)
|
|
|
|
|
|
def _bool_or_none(val):
|
|
"""Convert to bool, returning None if val is None."""
|
|
if val is None:
|
|
return None
|
|
return bool(val)
|
|
|
|
|
|
def filter_fields(data: dict, fields: Optional[Set[str]]) -> dict:
|
|
"""If fields set is provided, return only those keys."""
|
|
if not fields:
|
|
return data
|
|
return {k: v for k, v in data.items() if k in fields}
|
|
|
|
|
|
# ── Library Entity Serializers ────────────────────────────────
|
|
|
|
|
|
def serialize_artist(obj, fields: Optional[Set[str]] = None) -> dict:
|
|
"""Full artist serialization — all columns."""
|
|
d = _to_dict(obj)
|
|
result = {
|
|
"id": d.get("id"),
|
|
"name": d.get("name"),
|
|
"thumb_url": d.get("thumb_url"),
|
|
"banner_url": d.get("banner_url"),
|
|
"genres": _parse_genres(d.get("genres")),
|
|
"summary": d.get("summary"),
|
|
"style": d.get("style"),
|
|
"mood": d.get("mood"),
|
|
"label": d.get("label"),
|
|
"server_source": d.get("server_source"),
|
|
"created_at": _isoformat(d.get("created_at")),
|
|
"updated_at": _isoformat(d.get("updated_at")),
|
|
# External IDs
|
|
"musicbrainz_id": d.get("musicbrainz_id"),
|
|
"spotify_artist_id": d.get("spotify_artist_id"),
|
|
"itunes_artist_id": d.get("itunes_artist_id"),
|
|
"audiodb_id": d.get("audiodb_id"),
|
|
"deezer_id": d.get("deezer_id"),
|
|
"tidal_id": d.get("tidal_id"),
|
|
"qobuz_id": d.get("qobuz_id"),
|
|
"genius_id": d.get("genius_id"),
|
|
# Match statuses
|
|
"musicbrainz_match_status": d.get("musicbrainz_match_status"),
|
|
"spotify_match_status": d.get("spotify_match_status"),
|
|
"itunes_match_status": d.get("itunes_match_status"),
|
|
"audiodb_match_status": d.get("audiodb_match_status"),
|
|
"deezer_match_status": d.get("deezer_match_status"),
|
|
"lastfm_match_status": d.get("lastfm_match_status"),
|
|
"genius_match_status": d.get("genius_match_status"),
|
|
"tidal_match_status": d.get("tidal_match_status"),
|
|
"qobuz_match_status": d.get("qobuz_match_status"),
|
|
# Last attempted timestamps
|
|
"musicbrainz_last_attempted": _isoformat(d.get("musicbrainz_last_attempted")),
|
|
"spotify_last_attempted": _isoformat(d.get("spotify_last_attempted")),
|
|
"itunes_last_attempted": _isoformat(d.get("itunes_last_attempted")),
|
|
"audiodb_last_attempted": _isoformat(d.get("audiodb_last_attempted")),
|
|
"deezer_last_attempted": _isoformat(d.get("deezer_last_attempted")),
|
|
"lastfm_last_attempted": _isoformat(d.get("lastfm_last_attempted")),
|
|
"genius_last_attempted": _isoformat(d.get("genius_last_attempted")),
|
|
"tidal_last_attempted": _isoformat(d.get("tidal_last_attempted")),
|
|
"qobuz_last_attempted": _isoformat(d.get("qobuz_last_attempted")),
|
|
# Last.fm metadata
|
|
"lastfm_listeners": d.get("lastfm_listeners"),
|
|
"lastfm_playcount": d.get("lastfm_playcount"),
|
|
"lastfm_tags": d.get("lastfm_tags"),
|
|
"lastfm_similar": d.get("lastfm_similar"),
|
|
"lastfm_bio": d.get("lastfm_bio"),
|
|
"lastfm_url": d.get("lastfm_url"),
|
|
# Genius metadata
|
|
"genius_description": d.get("genius_description"),
|
|
"genius_alt_names": d.get("genius_alt_names"),
|
|
"genius_url": d.get("genius_url"),
|
|
}
|
|
# Preserve extra keys from enriched queries (album_count, track_count, is_watched)
|
|
for extra_key in ("album_count", "track_count", "is_watched", "image_url"):
|
|
if extra_key in d:
|
|
result[extra_key] = d[extra_key]
|
|
return filter_fields(result, fields)
|
|
|
|
|
|
def serialize_album(obj, fields: Optional[Set[str]] = None) -> dict:
|
|
"""Full album serialization — all columns."""
|
|
d = _to_dict(obj)
|
|
result = {
|
|
"id": d.get("id"),
|
|
"artist_id": d.get("artist_id"),
|
|
"title": d.get("title"),
|
|
"year": d.get("year"),
|
|
"thumb_url": d.get("thumb_url"),
|
|
"genres": _parse_genres(d.get("genres")),
|
|
"track_count": d.get("track_count"),
|
|
"duration": d.get("duration"),
|
|
"style": d.get("style"),
|
|
"mood": d.get("mood"),
|
|
"label": d.get("label"),
|
|
"explicit": _bool_or_none(d.get("explicit")),
|
|
"record_type": d.get("record_type"),
|
|
"server_source": d.get("server_source"),
|
|
"created_at": _isoformat(d.get("created_at")),
|
|
"updated_at": _isoformat(d.get("updated_at")),
|
|
"upc": d.get("upc"),
|
|
"copyright": d.get("copyright"),
|
|
# External IDs
|
|
"musicbrainz_release_id": d.get("musicbrainz_release_id"),
|
|
"spotify_album_id": d.get("spotify_album_id"),
|
|
"itunes_album_id": d.get("itunes_album_id"),
|
|
"audiodb_id": d.get("audiodb_id"),
|
|
"deezer_id": d.get("deezer_id"),
|
|
"tidal_id": d.get("tidal_id"),
|
|
"qobuz_id": d.get("qobuz_id"),
|
|
# Match statuses
|
|
"musicbrainz_match_status": d.get("musicbrainz_match_status"),
|
|
"spotify_match_status": d.get("spotify_match_status"),
|
|
"itunes_match_status": d.get("itunes_match_status"),
|
|
"audiodb_match_status": d.get("audiodb_match_status"),
|
|
"deezer_match_status": d.get("deezer_match_status"),
|
|
"lastfm_match_status": d.get("lastfm_match_status"),
|
|
"tidal_match_status": d.get("tidal_match_status"),
|
|
"qobuz_match_status": d.get("qobuz_match_status"),
|
|
# Last attempted timestamps
|
|
"musicbrainz_last_attempted": _isoformat(d.get("musicbrainz_last_attempted")),
|
|
"spotify_last_attempted": _isoformat(d.get("spotify_last_attempted")),
|
|
"itunes_last_attempted": _isoformat(d.get("itunes_last_attempted")),
|
|
"audiodb_last_attempted": _isoformat(d.get("audiodb_last_attempted")),
|
|
"deezer_last_attempted": _isoformat(d.get("deezer_last_attempted")),
|
|
"lastfm_last_attempted": _isoformat(d.get("lastfm_last_attempted")),
|
|
"tidal_last_attempted": _isoformat(d.get("tidal_last_attempted")),
|
|
"qobuz_last_attempted": _isoformat(d.get("qobuz_last_attempted")),
|
|
# Last.fm metadata
|
|
"lastfm_listeners": d.get("lastfm_listeners"),
|
|
"lastfm_playcount": d.get("lastfm_playcount"),
|
|
"lastfm_tags": d.get("lastfm_tags"),
|
|
"lastfm_wiki": d.get("lastfm_wiki"),
|
|
"lastfm_url": d.get("lastfm_url"),
|
|
}
|
|
return filter_fields(result, fields)
|
|
|
|
|
|
def serialize_track(obj, fields: Optional[Set[str]] = None) -> dict:
|
|
"""Full track serialization — all columns."""
|
|
d = _to_dict(obj)
|
|
result = {
|
|
"id": d.get("id"),
|
|
"album_id": d.get("album_id"),
|
|
"artist_id": d.get("artist_id"),
|
|
"title": d.get("title"),
|
|
"track_number": d.get("track_number"),
|
|
"duration": d.get("duration"),
|
|
"file_path": d.get("file_path"),
|
|
"bitrate": d.get("bitrate"),
|
|
"bpm": d.get("bpm"),
|
|
"explicit": _bool_or_none(d.get("explicit")),
|
|
"style": d.get("style"),
|
|
"mood": d.get("mood"),
|
|
"repair_status": d.get("repair_status"),
|
|
"repair_last_checked": _isoformat(d.get("repair_last_checked")),
|
|
"server_source": d.get("server_source"),
|
|
"created_at": _isoformat(d.get("created_at")),
|
|
"updated_at": _isoformat(d.get("updated_at")),
|
|
"isrc": d.get("isrc"),
|
|
"copyright": d.get("copyright"),
|
|
# External IDs
|
|
"musicbrainz_recording_id": d.get("musicbrainz_recording_id"),
|
|
"spotify_track_id": d.get("spotify_track_id"),
|
|
"itunes_track_id": d.get("itunes_track_id"),
|
|
"audiodb_id": d.get("audiodb_id"),
|
|
"deezer_id": d.get("deezer_id"),
|
|
"tidal_id": d.get("tidal_id"),
|
|
"qobuz_id": d.get("qobuz_id"),
|
|
"genius_id": d.get("genius_id"),
|
|
# Match statuses
|
|
"musicbrainz_match_status": d.get("musicbrainz_match_status"),
|
|
"spotify_match_status": d.get("spotify_match_status"),
|
|
"itunes_match_status": d.get("itunes_match_status"),
|
|
"audiodb_match_status": d.get("audiodb_match_status"),
|
|
"deezer_match_status": d.get("deezer_match_status"),
|
|
"lastfm_match_status": d.get("lastfm_match_status"),
|
|
"genius_match_status": d.get("genius_match_status"),
|
|
"tidal_match_status": d.get("tidal_match_status"),
|
|
"qobuz_match_status": d.get("qobuz_match_status"),
|
|
# Last attempted timestamps
|
|
"musicbrainz_last_attempted": _isoformat(d.get("musicbrainz_last_attempted")),
|
|
"spotify_last_attempted": _isoformat(d.get("spotify_last_attempted")),
|
|
"itunes_last_attempted": _isoformat(d.get("itunes_last_attempted")),
|
|
"audiodb_last_attempted": _isoformat(d.get("audiodb_last_attempted")),
|
|
"deezer_last_attempted": _isoformat(d.get("deezer_last_attempted")),
|
|
"lastfm_last_attempted": _isoformat(d.get("lastfm_last_attempted")),
|
|
"genius_last_attempted": _isoformat(d.get("genius_last_attempted")),
|
|
"tidal_last_attempted": _isoformat(d.get("tidal_last_attempted")),
|
|
"qobuz_last_attempted": _isoformat(d.get("qobuz_last_attempted")),
|
|
# Last.fm metadata
|
|
"lastfm_listeners": d.get("lastfm_listeners"),
|
|
"lastfm_playcount": d.get("lastfm_playcount"),
|
|
"lastfm_tags": d.get("lastfm_tags"),
|
|
"lastfm_url": d.get("lastfm_url"),
|
|
# Genius metadata
|
|
"genius_lyrics": d.get("genius_lyrics"),
|
|
"genius_description": d.get("genius_description"),
|
|
"genius_url": d.get("genius_url"),
|
|
}
|
|
# Preserve extra keys from joined queries (artist_name, album_title)
|
|
for extra_key in ("artist_name", "album_title"):
|
|
if extra_key in d:
|
|
result[extra_key] = d[extra_key]
|
|
return filter_fields(result, fields)
|
|
|
|
|
|
# ── Watchlist / Wishlist Serializers ──────────────────────────
|
|
|
|
|
|
def serialize_watchlist_artist(obj, fields: Optional[Set[str]] = None) -> dict:
|
|
"""Full watchlist artist serialization — all columns including all content filters."""
|
|
d = _to_dict(obj)
|
|
result = {
|
|
"id": d.get("id"),
|
|
"spotify_artist_id": d.get("spotify_artist_id"),
|
|
"itunes_artist_id": d.get("itunes_artist_id"),
|
|
"artist_name": d.get("artist_name"),
|
|
"image_url": d.get("image_url"),
|
|
"date_added": _isoformat(d.get("date_added")),
|
|
"last_scan_timestamp": _isoformat(d.get("last_scan_timestamp")),
|
|
"created_at": _isoformat(d.get("created_at")),
|
|
"updated_at": _isoformat(d.get("updated_at")),
|
|
"profile_id": d.get("profile_id"),
|
|
# Content type filters — ALL of them
|
|
"include_albums": bool(d.get("include_albums", True)),
|
|
"include_eps": bool(d.get("include_eps", True)),
|
|
"include_singles": bool(d.get("include_singles", True)),
|
|
"include_live": bool(d.get("include_live", False)),
|
|
"include_remixes": bool(d.get("include_remixes", False)),
|
|
"include_acoustic": bool(d.get("include_acoustic", False)),
|
|
"include_compilations": bool(d.get("include_compilations", False)),
|
|
}
|
|
return filter_fields(result, fields)
|
|
|
|
|
|
def serialize_wishlist_track(obj, fields: Optional[Set[str]] = None) -> dict:
|
|
"""Standardized wishlist track serialization."""
|
|
d = _to_dict(obj)
|
|
spotify_data = d.get("spotify_data", {})
|
|
if isinstance(spotify_data, str):
|
|
try:
|
|
spotify_data = json.loads(spotify_data)
|
|
except (json.JSONDecodeError, TypeError):
|
|
spotify_data = {}
|
|
|
|
source_info = d.get("source_info")
|
|
if isinstance(source_info, str):
|
|
try:
|
|
source_info = json.loads(source_info)
|
|
except (json.JSONDecodeError, TypeError):
|
|
source_info = None
|
|
|
|
result = {
|
|
"id": d.get("id"),
|
|
"spotify_track_id": d.get("spotify_track_id"),
|
|
"track_name": spotify_data.get("name", "Unknown") if isinstance(spotify_data, dict) else "Unknown",
|
|
"artist_name": ", ".join(
|
|
a.get("name", "") for a in spotify_data.get("artists", [])
|
|
) if isinstance(spotify_data, dict) and isinstance(spotify_data.get("artists"), list) else "",
|
|
"album_name": (
|
|
spotify_data.get("album", {}).get("name")
|
|
if isinstance(spotify_data, dict) and isinstance(spotify_data.get("album"), dict)
|
|
else None
|
|
),
|
|
"spotify_data": spotify_data,
|
|
"failure_reason": d.get("failure_reason"),
|
|
"retry_count": d.get("retry_count", 0),
|
|
"last_attempted": _isoformat(d.get("last_attempted")),
|
|
"date_added": _isoformat(d.get("date_added")),
|
|
"source_type": d.get("source_type"),
|
|
"source_info": source_info,
|
|
"profile_id": d.get("profile_id"),
|
|
}
|
|
return filter_fields(result, fields)
|
|
|
|
|
|
# ── Discovery Serializers ─────────────────────────────────────
|
|
|
|
|
|
def serialize_discovery_track(obj, fields: Optional[Set[str]] = None) -> dict:
|
|
"""Discovery pool track serialization."""
|
|
d = _to_dict(obj)
|
|
result = {
|
|
"id": d.get("id"),
|
|
"spotify_track_id": d.get("spotify_track_id"),
|
|
"spotify_album_id": d.get("spotify_album_id"),
|
|
"spotify_artist_id": d.get("spotify_artist_id"),
|
|
"itunes_track_id": d.get("itunes_track_id"),
|
|
"itunes_album_id": d.get("itunes_album_id"),
|
|
"itunes_artist_id": d.get("itunes_artist_id"),
|
|
"source": d.get("source"),
|
|
"track_name": d.get("track_name"),
|
|
"artist_name": d.get("artist_name"),
|
|
"album_name": d.get("album_name"),
|
|
"album_cover_url": d.get("album_cover_url"),
|
|
"duration_ms": d.get("duration_ms"),
|
|
"popularity": d.get("popularity"),
|
|
"release_date": d.get("release_date"),
|
|
"is_new_release": bool(d.get("is_new_release", False)),
|
|
"artist_genres": _parse_genres(d.get("artist_genres")),
|
|
"added_date": _isoformat(d.get("added_date")),
|
|
}
|
|
return filter_fields(result, fields)
|
|
|
|
|
|
def serialize_similar_artist(obj, fields: Optional[Set[str]] = None) -> dict:
|
|
"""Similar artist serialization."""
|
|
d = _to_dict(obj)
|
|
result = {
|
|
"id": d.get("id"),
|
|
"source_artist_id": d.get("source_artist_id"),
|
|
"similar_artist_spotify_id": d.get("similar_artist_spotify_id"),
|
|
"similar_artist_itunes_id": d.get("similar_artist_itunes_id"),
|
|
"similar_artist_name": d.get("similar_artist_name"),
|
|
"similarity_rank": d.get("similarity_rank"),
|
|
"occurrence_count": d.get("occurrence_count"),
|
|
"last_updated": _isoformat(d.get("last_updated")),
|
|
"last_featured": _isoformat(d.get("last_featured")),
|
|
}
|
|
return filter_fields(result, fields)
|
|
|
|
|
|
def serialize_recent_release(obj, fields: Optional[Set[str]] = None) -> dict:
|
|
"""Recent release serialization."""
|
|
d = _to_dict(obj)
|
|
result = {
|
|
"id": d.get("id"),
|
|
"watchlist_artist_id": d.get("watchlist_artist_id"),
|
|
"album_spotify_id": d.get("album_spotify_id"),
|
|
"album_itunes_id": d.get("album_itunes_id"),
|
|
"source": d.get("source"),
|
|
"album_name": d.get("album_name"),
|
|
"release_date": d.get("release_date"),
|
|
"album_cover_url": d.get("album_cover_url"),
|
|
"track_count": d.get("track_count"),
|
|
"added_date": _isoformat(d.get("added_date")),
|
|
}
|
|
return filter_fields(result, fields)
|