Expose mirrored playlist source-ref health

Return normalized source_ref metadata from mirrored playlist APIs so the UI no longer has to infer editable refresh links from description fields. Accept Spotify embed URLs during source-ref repair and add coverage for source-ref health reporting.
This commit is contained in:
Broque Thomas 2026-05-24 19:32:39 -07:00
parent 73bd2db547
commit 547e499121
4 changed files with 83 additions and 2 deletions

View file

@ -13,7 +13,7 @@ from __future__ import annotations
import hashlib
import re
from dataclasses import dataclass
from typing import Optional
from typing import Mapping, Optional
from urllib.parse import parse_qs, urlparse
@ -26,6 +26,14 @@ class MirroredSourceRef:
description: Optional[str]
@dataclass(frozen=True)
class MirroredSourceRefView:
source_ref: str
source_ref_kind: str
source_ref_status: str
source_ref_error: Optional[str] = None
def normalize_mirrored_source_ref(
source: str,
source_ref: str,
@ -75,6 +83,29 @@ def require_refresh_url(source: str, description: str, playlist_name: str = "")
return description
def describe_mirrored_source_ref(playlist: Mapping[str, object]) -> MirroredSourceRefView:
"""Build a UI/API friendly view of a mirrored playlist's refresh ref."""
source = str(playlist.get("source") or "").strip().lower()
source_playlist_id = str(playlist.get("source_playlist_id") or "").strip()
description = str(playlist.get("description") or "").strip()
name = str(playlist.get("name") or "")
if source in {"spotify_public", "youtube"}:
if description.startswith(("http://", "https://")):
return MirroredSourceRefView(description, "url", "ok")
try:
require_refresh_url(source, description, name)
except ValueError as exc:
return MirroredSourceRefView(
source_playlist_id,
"url",
"missing",
str(exc),
)
return MirroredSourceRefView(source_playlist_id, "id", "ok" if source_playlist_id else "missing")
def _canonical_spotify_url(source_ref: str) -> str:
parsed = _parse_spotify_ref(source_ref)
if parsed:
@ -94,7 +125,7 @@ def _parse_spotify_ref(source_ref: str) -> Optional[dict]:
return {"type": uri_match.group(1), "id": uri_match.group(2)}
url_match = re.search(
r"https?://open\.spotify\.com/(playlist|album)/([A-Za-z0-9]+)",
r"https?://open\.spotify\.com/(?:embed/)?(playlist|album)/([A-Za-z0-9]+)",
source_ref,
)
if url_match:

View file

@ -1,6 +1,7 @@
import pytest
from core.playlists.source_refs import (
describe_mirrored_source_ref,
normalize_mirrored_source_ref,
require_refresh_url,
)
@ -23,6 +24,16 @@ def test_spotify_public_raw_id_defaults_to_playlist_url():
assert out.description == "https://open.spotify.com/playlist/37i9dQZF1DXcBWIGoYBM5M"
def test_spotify_public_embed_url_canonicalizes_to_open_url():
out = normalize_mirrored_source_ref(
"spotify_public",
"https://open.spotify.com/embed/playlist/37i9dQZF1DX0kbJZpiYdZl?pi=abc",
)
assert out.source_playlist_id == "d29b105efc8e"
assert out.description == "https://open.spotify.com/playlist/37i9dQZF1DX0kbJZpiYdZl"
def test_youtube_url_stores_hash_and_canonical_url():
out = normalize_mirrored_source_ref(
"youtube",
@ -43,3 +54,29 @@ def test_direct_id_sources_preserve_existing_description():
def test_hash_backed_refresh_requires_url():
with pytest.raises(ValueError, match="missing its original source URL"):
require_refresh_url("spotify_public", "", "Release Radar")
def test_describe_hash_backed_ref_reports_missing_url():
view = describe_mirrored_source_ref({
"source": "spotify_public",
"source_playlist_id": "hash",
"description": "",
"name": "Release Radar",
})
assert view.source_ref == "hash"
assert view.source_ref_kind == "url"
assert view.source_ref_status == "missing"
assert "missing its original source URL" in view.source_ref_error
def test_describe_hash_backed_ref_uses_url_when_present():
view = describe_mirrored_source_ref({
"source": "spotify_public",
"source_playlist_id": "hash",
"description": "https://open.spotify.com/playlist/abc",
})
assert view.source_ref == "https://open.spotify.com/playlist/abc"
assert view.source_ref_kind == "url"
assert view.source_ref_status == "ok"

View file

@ -32183,6 +32183,7 @@ def mirror_playlist_endpoint():
def get_mirrored_playlists_endpoint():
"""List all mirrored playlists for the active profile."""
try:
from core.playlists.source_refs import describe_mirrored_source_ref
database = get_database()
profile_id = get_current_profile_id()
playlists = database.get_mirrored_playlists(profile_id=profile_id)
@ -32192,6 +32193,11 @@ def get_mirrored_playlists_endpoint():
pl['total_count'] = counts['total']
pl['wishlisted_count'] = counts['wishlisted']
pl['in_library_count'] = counts['in_library']
source_ref = describe_mirrored_source_ref(pl)
pl['source_ref'] = source_ref.source_ref
pl['source_ref_kind'] = source_ref.source_ref_kind
pl['source_ref_status'] = source_ref.source_ref_status
pl['source_ref_error'] = source_ref.source_ref_error
return jsonify(playlists)
except Exception as e:
logger.error(f"Error getting mirrored playlists: {e}")
@ -32201,10 +32207,16 @@ def get_mirrored_playlists_endpoint():
def get_mirrored_playlist_endpoint(playlist_id):
"""Get a mirrored playlist with its tracks."""
try:
from core.playlists.source_refs import describe_mirrored_source_ref
database = get_database()
playlist = database.get_mirrored_playlist(playlist_id)
if not playlist:
return jsonify({"error": "Playlist not found"}), 404
source_ref = describe_mirrored_source_ref(playlist)
playlist['source_ref'] = source_ref.source_ref
playlist['source_ref_kind'] = source_ref.source_ref_kind
playlist['source_ref_status'] = source_ref.source_ref_status
playlist['source_ref_error'] = source_ref.source_ref_error
playlist['tracks'] = database.get_mirrored_playlist_tracks(playlist_id)
return jsonify(playlist)
except Exception as e:

View file

@ -535,6 +535,7 @@ function renderMirroredCard(p, container) {
}
function getMirroredSourceRef(p) {
if (p && p.source_ref) return String(p.source_ref);
const desc = (p && p.description) ? String(p.description).trim() : '';
if ((p.source === 'spotify_public' || p.source === 'youtube') && /^https?:\/\//i.test(desc)) {
return desc;