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.
SoulSync/core/downloads/playlist_folder.py

129 lines
4.1 KiB

"""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',
]