soulsync/core/downloads/playlist_folder.py
BoulderBadgeDad a977d28144 Fix #780: Deezer/non-Spotify organize-by-playlist resolved the wrong row
resolve_mirrored_playlist tried the mirrored-playlists primary key FIRST for
any all-digit ref. Deezer upstream ids are all-numeric, so a Deezer playlist id
was mistaken for the PK and the organize-by-playlist toggle resolved a wrong row
(or nothing) — the toggle silently wouldn't save / 'Open in Mirrored' missed.

Resolve by (source, source_playlist_id) first, fall back to PK only when the
source lookup misses. Thread the batch/wishlist source through the download-path
callers so numeric upstream ids resolve correctly there too. Spotify (base62
ids) is unaffected.

Seam tests: numeric Deezer id resolves by source (not PK), spotify alphanumeric
by source, PK fallback still works, profile-scoped, empty refs -> None.
2026-06-03 20:41:04 -07:00

128 lines
4.1 KiB
Python

"""Playlist-folder layout helpers for download analysis and existence checks."""
from __future__ import annotations
import os
from typing import Any, Dict, List, Optional
from core.downloads.file_finder import AUDIO_EXTENSIONS
from core.imports.paths import (
_get_config_manager,
docker_resolve_path,
get_file_path_from_template,
sanitize_filename,
)
def _first_artist_name(artists: Any) -> str:
if not artists:
return ''
first = artists[0]
if isinstance(first, dict):
return str(first.get('name', '') or '').strip()
return str(first).strip()
def candidate_playlist_folder_paths(
playlist_name: str,
artist: str,
title: str,
) -> List[str]:
"""Return absolute candidate paths for a track in playlist-folder layout."""
if not playlist_name or not title:
return []
artist_name = (artist or 'Unknown Artist').strip()
track_name = title.strip()
transfer_dir = docker_resolve_path(
_get_config_manager().get('soulseek.transfer_path', './Transfer')
)
template_context = {
'artist': artist_name,
'albumartist': artist_name,
'album': track_name,
'title': track_name,
'playlist_name': playlist_name,
'track_number': 1,
'disc_number': 1,
'year': '',
'quality': '',
'albumtype': '',
'_artists_list': [{'name': artist_name}],
}
candidates: List[str] = []
folder_path, filename_base = get_file_path_from_template(template_context, 'playlist_path')
if folder_path and filename_base:
base = os.path.join(transfer_dir, folder_path, filename_base)
for ext in AUDIO_EXTENSIONS:
candidates.append(base + ext)
else:
playlist_name_sanitized = sanitize_filename(playlist_name)
playlist_dir = os.path.join(transfer_dir, playlist_name_sanitized)
artist_name_sanitized = sanitize_filename(artist_name)
track_name_sanitized = sanitize_filename(track_name)
stem = f'{artist_name_sanitized} - {track_name_sanitized}'
for ext in AUDIO_EXTENSIONS:
candidates.append(os.path.join(playlist_dir, stem + ext))
return candidates
def track_exists_in_playlist_folder(
playlist_name: str,
artist: str,
title: str,
) -> bool:
"""Return True if any audio file exists at the playlist-folder path for this track."""
for path in candidate_playlist_folder_paths(playlist_name, artist, title):
if os.path.isfile(path):
return True
return False
def track_exists_in_playlist_folder_from_track_data(
playlist_name: str,
track_data: Dict[str, Any],
) -> bool:
"""Check playlist-folder existence using Spotify-style track payload."""
title = track_data.get('name', '') or track_data.get('track_name', '')
artist = _first_artist_name(track_data.get('artists', []))
if not artist:
artist = str(track_data.get('artist_name', '') or '').strip()
return track_exists_in_playlist_folder(playlist_name, artist, title)
def resolve_playlist_folder_mode_for_batch(
db: Any,
*,
playlist_id: str,
playlist_name: str,
batch_playlist_folder_mode: bool,
profile_id: int = 1,
source: str = 'spotify',
) -> tuple[bool, str]:
"""Merge batch flag with persisted mirrored-playlist preference."""
if batch_playlist_folder_mode:
return True, playlist_name
if not hasattr(db, 'resolve_mirrored_playlist'):
return False, playlist_name
# Pass the batch's source so numeric upstream ids (e.g. Deezer) resolve by
# source instead of colliding with the mirrored-playlists primary key.
mirrored = db.resolve_mirrored_playlist(
playlist_id, profile_id=profile_id, default_source=source or 'spotify'
)
if mirrored and mirrored.get('organize_by_playlist'):
return True, mirrored.get('name') or playlist_name
return False, playlist_name
__all__ = [
'candidate_playlist_folder_paths',
'track_exists_in_playlist_folder',
'track_exists_in_playlist_folder_from_track_data',
'resolve_playlist_folder_mode_for_batch',
]