774d0f6c65
The ranked-target list is now the single source of truth for which formats download, in the user's exact priority order, for ALL sources — no hardcoded format hierarchy decides anything. A candidate passes only if it matches a ranked target; if nothing matches, the existing Use-Fallback toggle decides. - source_map: new shared format_from_extension() + AUDIO_EXTENSIONS — one source of truth for extension→format used by every extension-based source, so adding a format lights it up everywhere. Soulseek now classifies through it (opus/wav/aiff were previously dropped as 'unknown'). - file_ops.probe_audio_quality (generic import-time guard, all sources): add WMA; detect ALAC from the real codec (an .m4a is AAC or ALAC). - soulseek: drop the AAC-specific opt-in gate — AAC now follows the same universal rule as every format. - model.tier_score: documented as ONLY a same-format tiebreak + fallback order, never cross-format priority (the list owns that); add opus/alac bases. - UI: ranked-target editor offers all formats (FLAC/ALAC/WAV·AIFF lossless with bit-depth+sample-rate; MP3/AAC/OGG/Opus/WMA lossy with min-bitrate). - tests: AAC retargeted to the universal model; new coverage for format_from_extension and matches_target across all formats. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2197 lines
98 KiB
Python
2197 lines
98 KiB
Python
import requests
|
||
import asyncio
|
||
import aiohttp
|
||
import os
|
||
from typing import List, Optional, Dict, Any
|
||
from dataclasses import dataclass
|
||
import time
|
||
from pathlib import Path
|
||
from utils.logging_config import get_logger
|
||
from config.settings import config_manager
|
||
from core.imports.filename import parse_filename_metadata
|
||
# Shared download-result dataclasses + plugin contract live in the
|
||
# neutral plugin package — every source uses the same types, so they
|
||
# belong there rather than this soulseek-specific module.
|
||
from core.download_plugins.types import (
|
||
AlbumResult,
|
||
DownloadStatus,
|
||
SearchResult,
|
||
TrackResult,
|
||
)
|
||
from core.download_plugins.album_bundle import (
|
||
copy_audio_files_atomically,
|
||
get_poll_interval,
|
||
get_poll_timeout,
|
||
)
|
||
from core.download_plugins.base import DownloadSourcePlugin
|
||
from core.quality.model import QualityTarget, filter_and_rank, v2_qualities_to_ranked_targets
|
||
from core.quality.source_map import AUDIO_EXTENSIONS, format_from_extension
|
||
from utils.async_helpers import run_async
|
||
|
||
logger = get_logger("soulseek_client")
|
||
|
||
|
||
# slskd HTTP timeouts. Issue #499: long-running download sessions
|
||
# (~2-3hr) wedged because ``aiohttp.ClientSession()`` was constructed
|
||
# with no timeout — when slskd hung on a request (overloaded, network
|
||
# blip, internal stall), the HTTP call blocked indefinitely. The
|
||
# download worker thread blocked with it. Once the
|
||
# ``ThreadPoolExecutor(max_workers=3)`` had all 3 threads wedged,
|
||
# no further downloads could start and the user had to restart the
|
||
# container.
|
||
#
|
||
# Every slskd API call is metadata-level (search submission, status
|
||
# polls, download enqueue, transfer state queries) — none stream files.
|
||
# slskd handles file transfer via its own peer-to-peer infrastructure
|
||
# entirely outside our HTTP requests. So generous-but-bounded timeouts
|
||
# are safe and won't kill legitimate operations.
|
||
#
|
||
# Failures surface as caught exceptions in the existing
|
||
# ``except Exception`` blocks → logged + return None → caller treats
|
||
# as a normal failure (same as a 5xx response). No new error path.
|
||
_SLSKD_DEFAULT_TIMEOUT = aiohttp.ClientTimeout(
|
||
total=120, # hard ceiling — no single slskd call should take >2min
|
||
connect=15, # TCP connect to slskd
|
||
sock_read=60, # per-chunk read; slskd shouldn't go silent for >60s
|
||
)
|
||
|
||
|
||
# Search-rate-limit defaults. Pre-fix these were hardcoded magic numbers
|
||
# inside `SoulseekClient.__init__`. Lifted to module level so they're
|
||
# greppable + bumpable in one place, and so the reddit-reported case
|
||
# (Bell Canada anti-abuse trips on slskd peer-connection bursts) can
|
||
# tune them via `soulseek.search_*` config without touching code.
|
||
_DEFAULT_MAX_PER_WINDOW = 35
|
||
_DEFAULT_WINDOW_SECONDS = 220
|
||
_DEFAULT_MIN_DELAY_SECONDS = 0 # 0 = disabled (preserves prior behavior)
|
||
|
||
|
||
def compute_search_wait_seconds(
|
||
timestamps: List[float],
|
||
last_search_at: float,
|
||
now: float,
|
||
*,
|
||
max_per_window: int,
|
||
window_seconds: float,
|
||
min_delay_seconds: float,
|
||
) -> float:
|
||
"""Pure scheduler for the slskd search throttle.
|
||
|
||
Returns how many seconds the caller should sleep before issuing
|
||
the next search. ``timestamps`` is the list of recent search
|
||
timestamps already pruned to the current window (caller's job).
|
||
``last_search_at`` is the timestamp of the most recent search
|
||
(0.0 if there hasn't been one). ``now`` is the current monotonic /
|
||
wall-clock time (caller chooses — pure function only does math).
|
||
|
||
Two independent gates, return the larger:
|
||
|
||
1. **Sliding-window cap** — when ``len(timestamps) >= max_per_window``,
|
||
sleep until the oldest timestamp ages out of the window. Same
|
||
semantics as the pre-fix hardcoded behavior.
|
||
|
||
2. **Min-delay between searches** — when ``min_delay_seconds > 0``,
|
||
sleep until at least that many seconds have passed since
|
||
``last_search_at``. Smooths bursts even when the window isn't
|
||
full — this is the actual fix for the Reddit-reported case where
|
||
Bell Canada's anti-abuse trips on the rapid peer-connection
|
||
bursts that 35 back-to-back searches generate.
|
||
|
||
Returns 0.0 (no wait) when ``min_delay_seconds`` is 0 / negative
|
||
AND the window isn't full. Pure: no I/O, no side effects, no
|
||
mutation of the inputs.
|
||
"""
|
||
window_wait = 0.0
|
||
if max_per_window > 0 and len(timestamps) >= max_per_window:
|
||
oldest = timestamps[0]
|
||
window_wait = max(0.0, oldest + window_seconds - now)
|
||
|
||
delay_wait = 0.0
|
||
if min_delay_seconds > 0 and last_search_at > 0:
|
||
elapsed = now - last_search_at
|
||
delay_wait = max(0.0, min_delay_seconds - elapsed)
|
||
|
||
return max(window_wait, delay_wait)
|
||
|
||
|
||
class SoulseekClient(DownloadSourcePlugin):
|
||
def __init__(self):
|
||
self.base_url: Optional[str] = None
|
||
self.api_key: Optional[str] = None
|
||
self.download_path: Path = Path("./downloads")
|
||
self.active_searches: Dict[str, bool] = {} # search_id -> still_active
|
||
|
||
# Rate limiting for searches. Cap + window stay hardcoded —
|
||
# nobody has reported issues with the 35/220 defaults. The
|
||
# min-delay knob is the actual fix for the Reddit-reported
|
||
# case (Bell Canada anti-abuse cuts the WAN after rapid
|
||
# peer-connection bursts) — smooths bursts even when the
|
||
# sliding-window cap isn't hit. 0 = disabled (preserves prior
|
||
# behavior).
|
||
self.search_timestamps: List[float] = []
|
||
self._last_search_at: float = 0.0
|
||
self.max_searches_per_window = _DEFAULT_MAX_PER_WINDOW
|
||
self.rate_limit_window = _DEFAULT_WINDOW_SECONDS
|
||
self.search_min_delay_seconds = float(
|
||
config_manager.get('soulseek.search_min_delay_seconds', _DEFAULT_MIN_DELAY_SECONDS)
|
||
or _DEFAULT_MIN_DELAY_SECONDS
|
||
)
|
||
|
||
self._setup_client()
|
||
|
||
def _setup_client(self):
|
||
config = config_manager.get_soulseek_config()
|
||
|
||
if not config.get('slskd_url'):
|
||
logger.warning("Soulseek slskd URL not configured")
|
||
return
|
||
|
||
# Apply Docker URL resolution if running in container
|
||
slskd_url = config.get('slskd_url')
|
||
import os
|
||
if os.path.exists('/.dockerenv') and 'localhost' in slskd_url:
|
||
slskd_url = slskd_url.replace('localhost', 'host.docker.internal')
|
||
logger.info(f"Docker detected, using {slskd_url} for slskd connection")
|
||
|
||
self.base_url = slskd_url.rstrip('/')
|
||
self.api_key = config.get('api_key', '')
|
||
|
||
# Handle download path with Docker translation
|
||
download_path_str = config.get('download_path', './downloads')
|
||
if os.path.exists('/.dockerenv') and len(download_path_str) >= 3 and download_path_str[1] == ':' and download_path_str[0].isalpha():
|
||
# Convert Windows path (E:/path) to WSL mount path (/mnt/e/path)
|
||
drive_letter = download_path_str[0].lower()
|
||
rest_of_path = download_path_str[2:].replace('\\', '/') # Remove E: and convert backslashes
|
||
download_path_str = f"/host/mnt/{drive_letter}{rest_of_path}"
|
||
logger.info(f"Docker detected, using {download_path_str} for downloads")
|
||
|
||
self.download_path = Path(download_path_str)
|
||
try:
|
||
self.download_path.mkdir(parents=True, exist_ok=True)
|
||
except OSError as e:
|
||
logger.warning(f"Could not verify download path {download_path_str}: {e}")
|
||
|
||
logger.info(f"Soulseek client configured with slskd at {self.base_url}")
|
||
|
||
def _clean_old_timestamps(self):
|
||
"""Remove timestamps older than the rate limit window"""
|
||
current_time = time.time()
|
||
cutoff_time = current_time - self.rate_limit_window
|
||
self.search_timestamps = [ts for ts in self.search_timestamps if ts > cutoff_time]
|
||
|
||
async def _wait_for_rate_limit(self):
|
||
"""Wait if necessary to respect search rate limits.
|
||
|
||
Delegates the wait math to ``compute_search_wait_seconds`` so
|
||
the throttle logic is testable independently of asyncio.sleep
|
||
and the singleton client. Two gates apply (max wins): sliding-
|
||
window cap on searches per N seconds, plus optional min-delay
|
||
between consecutive searches (the burst-smoother).
|
||
"""
|
||
self._clean_old_timestamps()
|
||
wait_time = compute_search_wait_seconds(
|
||
self.search_timestamps,
|
||
self._last_search_at,
|
||
time.time(),
|
||
max_per_window=self.max_searches_per_window,
|
||
window_seconds=self.rate_limit_window,
|
||
min_delay_seconds=self.search_min_delay_seconds,
|
||
)
|
||
if wait_time > 0:
|
||
logger.info(
|
||
f"Search rate limit: waiting {wait_time:.1f}s "
|
||
f"({len(self.search_timestamps)}/{self.max_searches_per_window} in window, "
|
||
f"min_delay={self.search_min_delay_seconds:.1f}s)"
|
||
)
|
||
await asyncio.sleep(wait_time)
|
||
self._clean_old_timestamps()
|
||
|
||
# Record this search attempt
|
||
now = time.time()
|
||
self.search_timestamps.append(now)
|
||
self._last_search_at = now
|
||
|
||
def get_rate_limit_status(self) -> Dict[str, Any]:
|
||
"""Get current rate limiting status"""
|
||
self._clean_old_timestamps()
|
||
return {
|
||
'searches_in_window': len(self.search_timestamps),
|
||
'max_searches_per_window': self.max_searches_per_window,
|
||
'window_seconds': self.rate_limit_window,
|
||
'searches_remaining': max(0, self.max_searches_per_window - len(self.search_timestamps))
|
||
}
|
||
|
||
def _get_headers(self) -> Dict[str, str]:
|
||
headers = {'Content-Type': 'application/json'}
|
||
if self.api_key:
|
||
# Use X-API-Key authentication (Bearer tokens are session-based JWT tokens)
|
||
headers['X-API-Key'] = self.api_key
|
||
return headers
|
||
|
||
async def _make_request(self, method: str, endpoint: str, **kwargs) -> Optional[Dict[str, Any]]:
|
||
if not self.base_url:
|
||
logger.debug("Soulseek client not configured")
|
||
return None
|
||
|
||
url = f"{self.base_url}/api/v0/{endpoint}"
|
||
|
||
# Create a fresh session for each thread/event loop to avoid conflicts.
|
||
# Bounded timeout (issue #499) prevents the worker thread from
|
||
# wedging if slskd hangs.
|
||
session = None
|
||
try:
|
||
session = aiohttp.ClientSession(timeout=_SLSKD_DEFAULT_TIMEOUT)
|
||
|
||
headers = self._get_headers()
|
||
|
||
if 'json' in kwargs:
|
||
logger.debug(f"JSON payload: {kwargs['json']}")
|
||
|
||
async with session.request(
|
||
method,
|
||
url,
|
||
headers=headers,
|
||
**kwargs
|
||
) as response:
|
||
response_text = await response.text()
|
||
|
||
|
||
if response.status in [200, 201, 204]: # Accept 200 OK, 201 Created, and 204 No Content
|
||
self._last_401_logged = False # Reset on success
|
||
self._last_unreachable_logged = False # Same reset for unreachable-host suppression
|
||
try:
|
||
if response_text.strip(): # Only parse if there's content
|
||
return await response.json()
|
||
else:
|
||
# Return empty dict for successful requests with no content (like 201 Created)
|
||
return {}
|
||
except:
|
||
# If response_text was already consumed, parse it manually
|
||
import json
|
||
if response_text.strip():
|
||
return json.loads(response_text)
|
||
else:
|
||
return {}
|
||
else:
|
||
# Enhanced error logging for better debugging
|
||
error_detail = response_text if response_text.strip() else "No error details provided"
|
||
|
||
# Reduce noise for expected 404s (e.g. status checks for YouTube downloads)
|
||
# and repeated 401s (slskd not running / bad credentials)
|
||
if response.status == 404:
|
||
logger.debug(f"API request returned 404 (Not Found) for {url}")
|
||
elif response.status == 401:
|
||
if not getattr(self, '_last_401_logged', False):
|
||
logger.warning("slskd authentication failed (401) — check API key. Suppressing further 401 errors.")
|
||
self._last_401_logged = True
|
||
logger.debug(f"API request 401 for {url}")
|
||
else:
|
||
self._last_401_logged = False
|
||
logger.error(f"API request failed: HTTP {response.status} ({response.reason}) - {error_detail}")
|
||
logger.debug(f"Failed request: {method} {url}")
|
||
|
||
return None
|
||
|
||
except asyncio.TimeoutError:
|
||
# Issue #499: explicit handling so the worker thread unblocks
|
||
# instead of staying wedged on the HTTP call.
|
||
logger.warning(
|
||
f"slskd request timed out after {_SLSKD_DEFAULT_TIMEOUT.total}s: "
|
||
f"{method} {url} — slskd may be overloaded or unreachable"
|
||
)
|
||
return None
|
||
except aiohttp.ClientConnectorError as e:
|
||
# Issue #649: slskd_url is configured but the host is unreachable
|
||
# (slskd not running, wrong port, DNS / Docker bridge issue).
|
||
# Status polling at /api/downloads/status fans out to every plugin
|
||
# including soulseek even when the user has soulseek toggled out
|
||
# of their active download sources, so each frontend poll
|
||
# produced an ERROR log line — visible spam during any
|
||
# non-soulseek download. Suppress repeats to debug; emit one
|
||
# WARNING with actionable context, then reset on any successful
|
||
# response (slskd came back up).
|
||
if not getattr(self, '_last_unreachable_logged', False):
|
||
logger.warning(
|
||
f"slskd unreachable at {self.base_url}: {e}. "
|
||
f"Either start slskd or clear `soulseek.slskd_url` in settings "
|
||
f"if you don't use Soulseek. Suppressing further connection errors."
|
||
)
|
||
self._last_unreachable_logged = True
|
||
logger.debug(f"slskd connection failed: {method} {url}: {e}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"Error making API request: {e}")
|
||
return None
|
||
finally:
|
||
# Always clean up the session
|
||
if session:
|
||
try:
|
||
await session.close()
|
||
except Exception as _e:
|
||
logger.debug("aiohttp session close: %s", _e)
|
||
|
||
async def _make_direct_request(self, method: str, endpoint: str, **kwargs) -> Optional[Dict[str, Any]]:
|
||
"""Make a direct request to slskd without /api/v0/ prefix (for endpoints that work directly)"""
|
||
if not self.base_url:
|
||
logger.debug("Soulseek client not configured")
|
||
return None
|
||
|
||
url = f"{self.base_url}/{endpoint}"
|
||
|
||
# Create a fresh session for each thread/event loop to avoid conflicts.
|
||
# Bounded timeout (issue #499) prevents the worker thread from
|
||
# wedging if slskd hangs.
|
||
session = None
|
||
try:
|
||
session = aiohttp.ClientSession(timeout=_SLSKD_DEFAULT_TIMEOUT)
|
||
|
||
headers = self._get_headers()
|
||
|
||
if 'json' in kwargs:
|
||
logger.debug(f"JSON payload: {kwargs['json']}")
|
||
|
||
async with session.request(
|
||
method,
|
||
url,
|
||
headers=headers,
|
||
**kwargs
|
||
) as response:
|
||
response_text = await response.text()
|
||
|
||
|
||
if response.status == 200:
|
||
try:
|
||
return await response.json()
|
||
except:
|
||
# If response_text was already consumed, parse it manually
|
||
import json
|
||
return json.loads(response_text)
|
||
else:
|
||
logger.error(f"Direct API request failed: {response.status} - {response_text}")
|
||
return None
|
||
|
||
except asyncio.TimeoutError:
|
||
logger.warning(
|
||
f"slskd direct request timed out after {_SLSKD_DEFAULT_TIMEOUT.total}s: "
|
||
f"{method} {url} — slskd may be overloaded or unreachable"
|
||
)
|
||
return None
|
||
except aiohttp.ClientConnectorError as e:
|
||
# Issue #649 — same suppression as _make_request. Direct
|
||
# request is a less common path but uses the same base_url,
|
||
# so the same unreachable-host condition fires here.
|
||
if not getattr(self, '_last_unreachable_logged', False):
|
||
logger.warning(
|
||
f"slskd unreachable at {self.base_url}: {e}. "
|
||
f"Either start slskd or clear `soulseek.slskd_url` in settings "
|
||
f"if you don't use Soulseek. Suppressing further connection errors."
|
||
)
|
||
self._last_unreachable_logged = True
|
||
logger.debug(f"slskd direct connection failed: {method} {url}: {e}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"Error making direct API request: {e}")
|
||
return None
|
||
finally:
|
||
# Always clean up the session
|
||
if session:
|
||
try:
|
||
await session.close()
|
||
except Exception as _e:
|
||
logger.debug("aiohttp direct session close: %s", _e)
|
||
|
||
def _process_search_responses(self, responses_data: List[Dict[str, Any]]) -> tuple[List[TrackResult], List[AlbumResult]]:
|
||
"""Process search response data into TrackResult and AlbumResult objects"""
|
||
from collections import defaultdict
|
||
import re
|
||
|
||
all_tracks = []
|
||
albums_by_path = defaultdict(list)
|
||
|
||
|
||
|
||
# Audio file extensions to filter for
|
||
audio_extensions = AUDIO_EXTENSIONS
|
||
|
||
for response_data in responses_data:
|
||
username = response_data.get('username', '')
|
||
files = response_data.get('files', [])
|
||
|
||
|
||
for file_data in files:
|
||
filename = file_data.get('filename', '')
|
||
size = file_data.get('size', 0)
|
||
|
||
file_ext = Path(filename).suffix.lower().lstrip('.')
|
||
|
||
# Only process audio files
|
||
if f'.{file_ext}' not in audio_extensions:
|
||
continue
|
||
|
||
# Source-agnostic extension → format (shared with every other
|
||
# extension-based source). Ranked targets do the rest.
|
||
quality = format_from_extension(file_ext)
|
||
|
||
# Create TrackResult
|
||
# Convert duration from seconds to milliseconds (slskd returns seconds, Spotify uses ms)
|
||
raw_duration = file_data.get('length')
|
||
duration_ms = raw_duration * 1000 if raw_duration else None
|
||
|
||
slskd_attrs = {a['type']: a['value'] for a in file_data.get('attributes', [])}
|
||
track = TrackResult(
|
||
username=username,
|
||
filename=filename,
|
||
size=size,
|
||
bitrate=file_data.get('bitRate') or slskd_attrs.get(0),
|
||
duration=duration_ms,
|
||
quality=quality,
|
||
free_upload_slots=response_data.get('freeUploadSlots', 0),
|
||
upload_speed=response_data.get('uploadSpeed', 0),
|
||
queue_length=response_data.get('queueLength', 0),
|
||
sample_rate=slskd_attrs.get(4),
|
||
bit_depth=slskd_attrs.get(5),
|
||
)
|
||
|
||
all_tracks.append(track)
|
||
|
||
# Group tracks by album path for album detection
|
||
album_path = self._extract_album_path(filename)
|
||
if album_path:
|
||
albums_by_path[(username, album_path)].append(track)
|
||
|
||
# Create AlbumResults from grouped tracks
|
||
album_results = self._create_album_results(albums_by_path)
|
||
|
||
# Keep individual tracks that weren't grouped into albums
|
||
album_track_filenames = set()
|
||
for album in album_results:
|
||
for track in album.tracks:
|
||
album_track_filenames.add(track.filename)
|
||
|
||
# Individual tracks are those not part of any album
|
||
individual_tracks = [track for track in all_tracks if track.filename not in album_track_filenames]
|
||
|
||
|
||
return individual_tracks, album_results
|
||
|
||
def _extract_album_path(self, filename: str) -> Optional[str]:
|
||
"""Extract potential album directory path from filename"""
|
||
# Handle both Windows (\) and Unix (/) path separators
|
||
if '/' not in filename and '\\' not in filename:
|
||
return None
|
||
|
||
# Normalize path separators to forward slashes for consistent processing
|
||
normalized_path = filename.replace('\\', '/')
|
||
path_parts = normalized_path.split('/')
|
||
|
||
if len(path_parts) < 2:
|
||
return None
|
||
|
||
# Take the directory containing the file as potential album path
|
||
album_dir = path_parts[-2] # Directory containing the file
|
||
|
||
# Skip system directories that start with @ or are too short
|
||
if album_dir.startswith('@') or len(album_dir) < 2:
|
||
return None
|
||
|
||
# Return the full path up to the album directory (keeping forward slashes)
|
||
return '/'.join(path_parts[:-1])
|
||
|
||
|
||
def _create_album_results(self, albums_by_path: dict) -> List[AlbumResult]:
|
||
"""Create AlbumResult objects from grouped tracks"""
|
||
import re
|
||
from collections import Counter
|
||
|
||
album_results = []
|
||
|
||
for (username, album_path), tracks in albums_by_path.items():
|
||
# Only create albums for paths with multiple tracks (2+ tracks)
|
||
if len(tracks) < 2:
|
||
continue
|
||
|
||
# Calculate album metadata
|
||
total_size = sum(track.size for track in tracks)
|
||
quality_counts = Counter(track.quality for track in tracks)
|
||
dominant_quality = quality_counts.most_common(1)[0][0]
|
||
|
||
# Extract album title from path
|
||
album_title = self._extract_album_title(album_path)
|
||
|
||
# Try to determine artist from tracks or path
|
||
artist = self._determine_album_artist(tracks, album_path)
|
||
|
||
# Extract year if present
|
||
year = self._extract_year(album_path, album_title)
|
||
|
||
# Use user metrics from first track (they should be the same for all tracks from same user)
|
||
first_track = tracks[0]
|
||
|
||
album = AlbumResult(
|
||
username=username,
|
||
album_path=album_path,
|
||
album_title=album_title,
|
||
artist=artist,
|
||
track_count=len(tracks),
|
||
total_size=total_size,
|
||
tracks=sorted(tracks, key=lambda t: t.track_number or 0), # Sort by track number
|
||
dominant_quality=dominant_quality,
|
||
year=year,
|
||
free_upload_slots=first_track.free_upload_slots,
|
||
upload_speed=first_track.upload_speed,
|
||
queue_length=first_track.queue_length
|
||
)
|
||
|
||
album_results.append(album)
|
||
|
||
return album_results
|
||
|
||
def _extract_album_title(self, album_path: str) -> str:
|
||
"""Extract album title from directory path"""
|
||
import re
|
||
|
||
# Get the last directory name as album title
|
||
album_dir = album_path.split('/')[-1]
|
||
|
||
# Clean up common patterns
|
||
# Remove leading numbers and separators
|
||
cleaned = re.sub(r'^\d+\s*[-\.\s]+', '', album_dir)
|
||
|
||
# Remove year patterns at the end: (2023), [2023], - 2023
|
||
cleaned = re.sub(r'\s*[-\(\[]?\d{4}[-\)\]]?\s*$', '', cleaned)
|
||
|
||
# Remove common separators and extra spaces
|
||
cleaned = re.sub(r'\s+', ' ', cleaned).strip()
|
||
|
||
return cleaned if cleaned else album_dir
|
||
|
||
def _determine_album_artist(self, tracks: List[TrackResult], album_path: str) -> Optional[str]:
|
||
"""Determine album artist from track artists or path"""
|
||
from collections import Counter
|
||
|
||
# Get artist from tracks
|
||
track_artists = [track.artist for track in tracks if track.artist]
|
||
if track_artists:
|
||
# Use most common artist
|
||
artist_counts = Counter(track_artists)
|
||
return artist_counts.most_common(1)[0][0]
|
||
|
||
# Try to extract from path
|
||
import re
|
||
album_dir = album_path.split('/')[-1]
|
||
|
||
# Look for "Artist - Album" pattern
|
||
artist_match = re.match(r'^(.+?)\s*[-–]\s*(.+)$', album_dir)
|
||
if artist_match:
|
||
potential_artist = artist_match.group(1).strip()
|
||
if len(potential_artist) > 1:
|
||
return potential_artist
|
||
|
||
return None
|
||
|
||
def _extract_year(self, album_path: str, album_title: str) -> Optional[str]:
|
||
"""Extract year from album path or title"""
|
||
import re
|
||
|
||
# Look for 4-digit year in parentheses, brackets, or after dash
|
||
text_to_search = f"{album_path} {album_title}"
|
||
year_patterns = [
|
||
r'\((\d{4})\)', # (2023)
|
||
r'\[(\d{4})\]', # [2023]
|
||
r'\s-(\d{4})$', # - 2023 at end
|
||
r'\s(\d{4})\s', # 2023 with spaces
|
||
r'\s(\d{4})$' # 2023 at end
|
||
]
|
||
|
||
for pattern in year_patterns:
|
||
match = re.search(pattern, text_to_search)
|
||
if match:
|
||
year = match.group(1)
|
||
# Validate year range (1900-2030)
|
||
if 1900 <= int(year) <= 2030:
|
||
return year
|
||
|
||
return None
|
||
|
||
async def search(self, query: str, timeout: int = None, progress_callback=None) -> tuple[List[TrackResult], List[AlbumResult]]:
|
||
if not self.base_url:
|
||
logger.debug("Soulseek client not configured")
|
||
return [], []
|
||
|
||
# Get timeout from config if not specified
|
||
from config.settings import config_manager
|
||
if timeout is None:
|
||
timeout = config_manager.get('soulseek.search_timeout', 60)
|
||
|
||
# Apply rate limiting before search
|
||
await self._wait_for_rate_limit()
|
||
|
||
try:
|
||
logger.info(f"Starting search for: '{query}' (slskd timeout: {timeout}s)")
|
||
|
||
# Get minimum peer upload speed from config (stored as Mbps, API expects bytes/sec)
|
||
min_speed_mbps = config_manager.get('soulseek.min_peer_upload_speed', 0) or 0
|
||
min_speed_bytes = int(min_speed_mbps) * 125000 # 1 Mbps = 125000 bytes/sec
|
||
|
||
search_data = {
|
||
'searchText': query,
|
||
'timeout': timeout * 1000, # slskd expects milliseconds
|
||
'filterResponses': True,
|
||
'minimumResponseFileCount': 1,
|
||
'minimumPeerUploadSpeed': min_speed_bytes
|
||
}
|
||
|
||
logger.debug(f"Search data: {search_data}")
|
||
logger.debug(f"Making POST request to: {self.base_url}/api/v0/searches")
|
||
|
||
response = await self._make_request('POST', 'searches', json=search_data)
|
||
if not response:
|
||
logger.error("No response from search POST request")
|
||
return [], []
|
||
|
||
# Handle both dict and list responses from slskd API
|
||
search_id = None
|
||
if isinstance(response, dict):
|
||
search_id = response.get('id')
|
||
elif isinstance(response, list) and len(response) > 0:
|
||
search_id = response[0].get('id') if isinstance(response[0], dict) else None
|
||
|
||
if not search_id:
|
||
logger.error("No search ID returned from POST request")
|
||
logger.debug(f"Full response (type: {type(response)}): {response}")
|
||
return [], []
|
||
|
||
logger.info(f"Search initiated with ID: {search_id}")
|
||
|
||
# Track this search as active
|
||
self.active_searches[search_id] = True
|
||
|
||
# Get timeout buffer from config
|
||
from config.settings import config_manager
|
||
timeout_buffer = config_manager.get('soulseek.search_timeout_buffer', 15)
|
||
|
||
# Poll for results - process and emit results immediately when found
|
||
all_responses = []
|
||
all_tracks = []
|
||
all_albums = []
|
||
poll_interval = 1 # Check every 1 second for responsive updates
|
||
|
||
# IMPORTANT: Poll for LONGER than slskd searches to catch all results
|
||
# slskd timeout: how long slskd searches for
|
||
# polling timeout: how long WE wait for slskd to finish (with buffer)
|
||
polling_timeout = timeout + timeout_buffer
|
||
max_polls = int(polling_timeout / poll_interval)
|
||
|
||
logger.info(f"Polling for up to {polling_timeout}s (slskd timeout: {timeout}s + buffer: {timeout_buffer}s)")
|
||
|
||
for poll_count in range(max_polls):
|
||
# Check if search was cancelled
|
||
if search_id not in self.active_searches:
|
||
logger.info(f"Search {search_id} was cancelled, stopping")
|
||
return [], []
|
||
|
||
logger.debug(f"Polling for results (attempt {poll_count + 1}/{max_polls}) - elapsed: {poll_count * poll_interval:.1f}s")
|
||
|
||
# Get current search responses
|
||
responses_data = await self._make_request('GET', f'searches/{search_id}/responses')
|
||
if responses_data and isinstance(responses_data, list):
|
||
# Check if we got new responses
|
||
new_response_count = len(responses_data) - len(all_responses)
|
||
if new_response_count > 0:
|
||
# Process only the new responses
|
||
new_responses = responses_data[len(all_responses):]
|
||
all_responses = responses_data
|
||
|
||
logger.info(f"Found {new_response_count} new responses ({len(all_responses)} total) at {poll_count * poll_interval:.1f}s")
|
||
|
||
# Process new responses immediately
|
||
new_tracks, new_albums = self._process_search_responses(new_responses)
|
||
|
||
# Add to cumulative results
|
||
all_tracks.extend(new_tracks)
|
||
all_albums.extend(new_albums)
|
||
|
||
# Sort by quality score for better display order
|
||
all_tracks.sort(key=lambda x: x.quality_score, reverse=True)
|
||
all_albums.sort(key=lambda x: x.quality_score, reverse=True)
|
||
|
||
# Call progress callback with processed results immediately
|
||
if progress_callback:
|
||
try:
|
||
progress_callback(all_tracks, all_albums, len(all_responses))
|
||
except Exception as e:
|
||
logger.error(f"Error in progress callback: {e}")
|
||
|
||
logger.info(f"Processed results: {len(all_tracks)} tracks, {len(all_albums)} albums")
|
||
|
||
# Early termination if we have enough responses
|
||
if len(all_responses) >= 30: # Stop after 30 responses for better performance
|
||
logger.info(f"Early termination: Found {len(all_responses)} responses, stopping search")
|
||
break
|
||
elif len(all_responses) > 0:
|
||
logger.debug(f"No new responses, total still: {len(all_responses)}")
|
||
else:
|
||
logger.debug(f"Still waiting for responses... ({poll_count * poll_interval:.1f}s elapsed)")
|
||
|
||
# Wait before next poll (unless this is the last attempt)
|
||
if poll_count < max_polls - 1:
|
||
await asyncio.sleep(poll_interval)
|
||
|
||
logger.info(f"Search completed. Final results: {len(all_tracks)} tracks and {len(all_albums)} albums for query: {query}")
|
||
return all_tracks, all_albums
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error searching: {e}")
|
||
return [], []
|
||
finally:
|
||
# Remove from active searches when done
|
||
if 'search_id' in locals() and search_id in self.active_searches:
|
||
del self.active_searches[search_id]
|
||
|
||
async def download(self, username: str, filename: str, file_size: int = 0) -> Optional[str]:
|
||
if not self.base_url:
|
||
logger.debug("Soulseek client not configured")
|
||
return None
|
||
|
||
try:
|
||
logger.debug(f"Attempting to download: {filename} from {username} (size: {file_size})")
|
||
|
||
# Use the exact format observed in the web interface
|
||
# Payload: [{filename: "...", size: 123}] - array of files
|
||
# Try adding path parameter to see if slskd supports custom download paths
|
||
download_data = [
|
||
{
|
||
"filename": filename,
|
||
"size": file_size,
|
||
"path": str(self.download_path) # Try custom download path
|
||
}
|
||
]
|
||
|
||
logger.debug(f"Using web interface API format: {download_data}")
|
||
|
||
# Use the correct endpoint pattern from web interface: /api/v0/transfers/downloads/{username}
|
||
endpoint = f'transfers/downloads/{username}'
|
||
logger.debug(f"Trying web interface endpoint: {endpoint}")
|
||
|
||
try:
|
||
response = await self._make_request('POST', endpoint, json=download_data)
|
||
if response is not None: # 201 Created might return download info
|
||
logger.info(f"[SUCCESS] Started download: {filename} from {username}")
|
||
# Try to extract download ID from response if available
|
||
if isinstance(response, dict) and 'id' in response:
|
||
logger.debug(f"Got download ID from response: {response['id']}")
|
||
return response['id']
|
||
elif isinstance(response, list) and len(response) > 0 and 'id' in response[0]:
|
||
logger.debug(f"Got download ID from response list: {response[0]['id']}")
|
||
return response[0]['id']
|
||
else:
|
||
# Fallback to filename if no ID in response
|
||
logger.debug(f"No ID in response, using filename as fallback: {response}")
|
||
return filename
|
||
else:
|
||
logger.debug("Web interface endpoint returned no response")
|
||
|
||
except Exception as e:
|
||
logger.debug(f"Web interface endpoint failed: {e}")
|
||
|
||
# Fallback: Try alternative patterns if the main one fails
|
||
logger.debug("Web interface endpoint failed, trying alternatives...")
|
||
|
||
# Try different username-based endpoint patterns
|
||
username_endpoints_to_try = [
|
||
f'transfers/{username}/enqueue',
|
||
f'users/{username}/downloads',
|
||
f'users/{username}/enqueue'
|
||
]
|
||
|
||
# Try with array format first
|
||
for endpoint in username_endpoints_to_try:
|
||
logger.debug(f"Trying endpoint: {endpoint} with array format")
|
||
|
||
try:
|
||
response = await self._make_request('POST', endpoint, json=download_data)
|
||
if response is not None:
|
||
logger.info(f"[SUCCESS] Started download: {filename} from {username} using endpoint: {endpoint}")
|
||
# Try to extract download ID from response if available
|
||
if isinstance(response, dict) and 'id' in response:
|
||
logger.debug(f"Got download ID from response: {response['id']}")
|
||
return response['id']
|
||
elif isinstance(response, list) and len(response) > 0 and 'id' in response[0]:
|
||
logger.debug(f"Got download ID from response list: {response[0]['id']}")
|
||
return response[0]['id']
|
||
else:
|
||
# Fallback to filename if no ID in response
|
||
logger.debug(f"No ID in response, using filename as fallback: {response}")
|
||
return filename
|
||
else:
|
||
logger.debug(f"Endpoint {endpoint} returned no response")
|
||
|
||
except Exception as e:
|
||
logger.debug(f"Endpoint {endpoint} failed: {e}")
|
||
continue
|
||
|
||
# Try with old format as final fallback
|
||
logger.debug("Array format failed, trying old object format")
|
||
fallback_data = {
|
||
"files": [
|
||
{
|
||
"filename": filename,
|
||
"size": file_size
|
||
}
|
||
]
|
||
}
|
||
|
||
for endpoint in username_endpoints_to_try:
|
||
logger.debug(f"Trying endpoint: {endpoint} with object format")
|
||
|
||
try:
|
||
response = await self._make_request('POST', endpoint, json=fallback_data)
|
||
if response is not None:
|
||
logger.info(f"[SUCCESS] Started download: {filename} from {username} using fallback endpoint: {endpoint}")
|
||
# Try to extract download ID from response if available
|
||
if isinstance(response, dict) and 'id' in response:
|
||
logger.debug(f"Got download ID from response: {response['id']}")
|
||
return response['id']
|
||
elif isinstance(response, list) and len(response) > 0 and 'id' in response[0]:
|
||
logger.debug(f"Got download ID from response list: {response[0]['id']}")
|
||
return response[0]['id']
|
||
else:
|
||
# Fallback to filename if no ID in response
|
||
logger.debug(f"No ID in response, using filename as fallback: {response}")
|
||
return filename
|
||
else:
|
||
logger.debug(f"Fallback endpoint {endpoint} returned no response")
|
||
|
||
except Exception as e:
|
||
logger.debug(f"Fallback endpoint {endpoint} failed: {e}")
|
||
continue
|
||
|
||
logger.error(f"All download endpoints failed for {filename} from {username}")
|
||
return None
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error starting download: {e}")
|
||
return None
|
||
|
||
async def get_download_status(self, download_id: str) -> Optional[DownloadStatus]:
|
||
if not self.base_url:
|
||
return None
|
||
|
||
try:
|
||
response = await self._make_request('GET', f'transfers/downloads/{download_id}')
|
||
if not response:
|
||
return None
|
||
|
||
# Handle both dict and list responses (slskd API can vary)
|
||
download_data = None
|
||
if isinstance(response, dict):
|
||
download_data = response
|
||
elif isinstance(response, list) and len(response) > 0 and isinstance(response[0], dict):
|
||
download_data = response[0]
|
||
|
||
if not download_data:
|
||
logger.error(f"Invalid response format for download status (type: {type(response)})")
|
||
return None
|
||
|
||
return DownloadStatus(
|
||
id=download_data.get('id', ''),
|
||
filename=download_data.get('filename', ''),
|
||
username=download_data.get('username', ''),
|
||
state=download_data.get('state', ''),
|
||
progress=download_data.get('percentComplete', 0.0),
|
||
size=download_data.get('size', 0),
|
||
transferred=download_data.get('bytesTransferred', 0),
|
||
speed=download_data.get('averageSpeed', 0),
|
||
time_remaining=download_data.get('timeRemaining')
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error getting download status: {e}")
|
||
return None
|
||
|
||
async def get_all_downloads(self) -> List[DownloadStatus]:
|
||
if not self.base_url:
|
||
return []
|
||
|
||
try:
|
||
# FIXED: Skip the 404 endpoint and go straight to the working one
|
||
response = await self._make_request('GET', 'transfers/downloads')
|
||
|
||
if not response:
|
||
return []
|
||
|
||
downloads = []
|
||
|
||
# FIXED: Parse the nested response structure correctly
|
||
# Response format: [{"username": "user", "directories": [{"files": [...]}]}]
|
||
for user_data in response:
|
||
username = user_data.get('username', '')
|
||
directories = user_data.get('directories', [])
|
||
|
||
for directory in directories:
|
||
files = directory.get('files', [])
|
||
|
||
for file_data in files:
|
||
# Parse progress from the state if available
|
||
progress = 0.0
|
||
if file_data.get('state', '').lower().startswith('completed'):
|
||
progress = 100.0
|
||
elif 'progress' in file_data:
|
||
progress = float(file_data.get('progress', 0.0))
|
||
|
||
status = DownloadStatus(
|
||
id=file_data.get('id', ''),
|
||
filename=file_data.get('filename', ''),
|
||
username=username,
|
||
state=file_data.get('state', ''),
|
||
progress=progress,
|
||
size=file_data.get('size', 0),
|
||
transferred=file_data.get('bytesTransferred', 0), # May not exist in API
|
||
speed=file_data.get('averageSpeed', 0), # May not exist in API
|
||
time_remaining=file_data.get('timeRemaining')
|
||
)
|
||
downloads.append(status)
|
||
|
||
logger.debug(f"Parsed {len(downloads)} downloads from API response")
|
||
return downloads
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error getting downloads: {e}")
|
||
return []
|
||
|
||
async def cancel_download(self, download_id: str, username: str = None, remove: bool = False) -> bool:
|
||
if not self.base_url:
|
||
return False
|
||
|
||
# If username is not provided, try to extract it from stored transfer data
|
||
if not username:
|
||
logger.debug(f"No username provided for download_id {download_id}, attempting to find it")
|
||
try:
|
||
downloads = await self.get_all_downloads()
|
||
for download in downloads:
|
||
if download.id == download_id:
|
||
username = download.username
|
||
logger.debug(f"Found username {username} for download_id {download_id}")
|
||
break
|
||
|
||
if not username:
|
||
logger.error(f"Could not find username for download_id {download_id}")
|
||
return False
|
||
except Exception as e:
|
||
logger.error(f"Error finding username for download: {e}")
|
||
return False
|
||
|
||
try:
|
||
from urllib.parse import quote
|
||
# URL-encode download_id to handle backslashes and special characters
|
||
encoded_id = quote(download_id, safe='')
|
||
|
||
# Try multiple API formats as slskd API may vary between versions
|
||
endpoints_to_try = [
|
||
# Format 1: With username and remove parameter (original format)
|
||
f'transfers/downloads/{username}/{encoded_id}?remove={str(remove).lower()}',
|
||
# Format 2: Simple format with just download_id (used in sync.py)
|
||
f'transfers/downloads/{encoded_id}',
|
||
# Format 3: Alternative format without remove parameter
|
||
f'transfers/downloads/{username}/{encoded_id}'
|
||
]
|
||
|
||
action = "Removing" if remove else "Cancelling"
|
||
|
||
for i, endpoint in enumerate(endpoints_to_try):
|
||
logger.debug(f"{action} download (attempt {i+1}/3) with endpoint: {endpoint}")
|
||
response = await self._make_request('DELETE', endpoint)
|
||
if response is not None:
|
||
logger.info(f"Successfully cancelled download using endpoint format {i+1}")
|
||
return True
|
||
else:
|
||
logger.debug(f"Endpoint format {i+1} failed: {endpoint}")
|
||
|
||
# Fallback: if download_id looks like a filename (contains path separators),
|
||
# list all transfers, find by filename, and cancel with the real transfer ID
|
||
if '\\' in download_id or '/' in download_id:
|
||
logger.debug("Download ID looks like a filename, trying filename-based lookup fallback")
|
||
try:
|
||
downloads = await self.get_all_downloads()
|
||
target_basename = os.path.basename(download_id.replace('\\', '/'))
|
||
for download in downloads:
|
||
dl_basename = os.path.basename(download.filename.replace('\\', '/'))
|
||
if dl_basename == target_basename and download.username == username:
|
||
real_id = quote(str(download.id), safe='')
|
||
fallback_endpoint = f'transfers/downloads/{username}/{real_id}?remove={str(remove).lower()}'
|
||
logger.debug(f"Found matching transfer with real ID, trying: {fallback_endpoint}")
|
||
response = await self._make_request('DELETE', fallback_endpoint)
|
||
if response is not None:
|
||
logger.info("Successfully cancelled download via filename fallback")
|
||
return True
|
||
except Exception as fallback_error:
|
||
logger.debug(f"Filename fallback failed: {fallback_error}")
|
||
|
||
logger.error(f"All cancel endpoint formats failed for download_id: {download_id}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error cancelling download: {e}")
|
||
return False
|
||
|
||
async def signal_download_completion(self, download_id: str, username: str, remove: bool = True) -> bool:
|
||
"""Signal the Soulseek API that a download has completed or been cancelled
|
||
|
||
Args:
|
||
download_id: The ID of the download
|
||
username: The uploader username
|
||
remove: True to remove from transfer list (completion), False to just cancel
|
||
|
||
Returns:
|
||
bool: True if signal was successful, False otherwise
|
||
"""
|
||
if not self.base_url:
|
||
logger.debug("Soulseek client not configured")
|
||
return False
|
||
|
||
try:
|
||
# Use the API endpoint format: /transfers/downloads/{username}/{download_id}?remove={true/false}
|
||
endpoint = f'transfers/downloads/{username}/{download_id}?remove={str(remove).lower()}'
|
||
action = "Signaling completion" if remove else "Signaling cancellation"
|
||
logger.debug(f"{action} for download {download_id} from {username}")
|
||
|
||
response = await self._make_request('DELETE', endpoint)
|
||
success = response is not None
|
||
|
||
if success:
|
||
logger.info(f"Successfully signaled download {action.lower()}: {download_id}")
|
||
else:
|
||
logger.warning(f"Failed to signal download {action.lower()}: {download_id}")
|
||
|
||
return success
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error signaling download completion: {e}")
|
||
return False
|
||
|
||
async def browse_user_directory(self, username: str, directory: str, timeout: int = 10) -> Optional[List[Dict[str, Any]]]:
|
||
"""Browse a specific directory on a Soulseek user's share.
|
||
|
||
Args:
|
||
username: The Soulseek username to browse
|
||
directory: The directory path to list
|
||
timeout: Request timeout in seconds
|
||
|
||
Returns:
|
||
List of file dicts from the directory, or None on failure
|
||
"""
|
||
if not self.base_url:
|
||
return None
|
||
try:
|
||
response = await self._make_request('POST', f'users/{username}/directory',
|
||
json={"directory": directory})
|
||
if not response:
|
||
logger.warning(f"Browse got empty/None response for {username}:{directory}")
|
||
return None
|
||
# Log raw response keys to debug field naming
|
||
if isinstance(response, dict):
|
||
logger.info(f"Browse response keys: {list(response.keys())}")
|
||
# Try multiple possible key names (slskd API may use 'files' or 'directories')
|
||
files = response.get('files', [])
|
||
if not files:
|
||
# Some slskd versions nest files under directories
|
||
dirs = response.get('directories', [])
|
||
if dirs and isinstance(dirs, list) and len(dirs) > 0:
|
||
files = dirs[0].get('files', []) if isinstance(dirs[0], dict) else []
|
||
if not files:
|
||
logger.info(f"Browse raw response (truncated): {str(response)[:500]}")
|
||
elif isinstance(response, list):
|
||
logger.info(f"Browse response is a list with {len(response)} items")
|
||
# Response is likely a list of directory objects, each containing 'files'
|
||
if len(response) > 0:
|
||
first_item = response[0]
|
||
logger.info(f"Browse first item type={type(first_item).__name__}, keys={list(first_item.keys()) if isinstance(first_item, dict) else 'N/A'}")
|
||
if isinstance(first_item, dict) and 'files' in first_item:
|
||
files = first_item.get('files', [])
|
||
logger.info(f"Extracted {len(files)} files from directory object")
|
||
else:
|
||
# Log the item to understand its structure
|
||
logger.info(f"Browse first item (truncated): {str(first_item)[:500]}")
|
||
files = response
|
||
else:
|
||
files = []
|
||
else:
|
||
files = []
|
||
logger.info(f"Browse found {len(files)} files in {username}:{directory}")
|
||
return files
|
||
except Exception as e:
|
||
logger.warning(f"Error browsing {username}:{directory}: {e}")
|
||
return None
|
||
|
||
def parse_browse_results_to_tracks(self, username: str, files: List[Dict[str, Any]],
|
||
upload_speed: int = 0, free_slots: int = 0,
|
||
queue_length: int = 0,
|
||
directory: str = '') -> List['TrackResult']:
|
||
"""Convert browse API file results into TrackResult objects.
|
||
|
||
Args:
|
||
username: The source username
|
||
files: Raw file dicts from browse API
|
||
upload_speed: User's upload speed
|
||
free_slots: User's free upload slots
|
||
queue_length: User's queue length
|
||
directory: The directory path these files came from (prepended to bare filenames)
|
||
|
||
Returns:
|
||
List of TrackResult objects for audio files
|
||
"""
|
||
audio_extensions = AUDIO_EXTENSIONS
|
||
results = []
|
||
if files:
|
||
logger.debug(f"Browse raw file sample: {files[0]}")
|
||
for file_data in files:
|
||
filename = file_data.get('filename', '')
|
||
# If filename is bare (no path separators), prepend the directory path
|
||
# so the matching engine can find artist/album context in the full path
|
||
if directory and '\\' not in filename and '/' not in filename:
|
||
sep = '\\' if '\\' in directory else '/'
|
||
filename = f"{directory}{sep}{filename}"
|
||
ext = Path(filename).suffix.lower()
|
||
if ext not in audio_extensions:
|
||
continue
|
||
quality = format_from_extension(ext)
|
||
raw_duration = file_data.get('length')
|
||
duration_ms = raw_duration * 1000 if raw_duration else None
|
||
slskd_attrs = {a['type']: a['value'] for a in file_data.get('attributes', [])}
|
||
results.append(TrackResult(
|
||
username=username, filename=filename, size=file_data.get('size', 0),
|
||
bitrate=file_data.get('bitRate') or slskd_attrs.get(0),
|
||
duration=duration_ms, quality=quality,
|
||
free_upload_slots=free_slots, upload_speed=upload_speed, queue_length=queue_length,
|
||
sample_rate=slskd_attrs.get(4),
|
||
bit_depth=slskd_attrs.get(5),
|
||
))
|
||
return results
|
||
|
||
async def cancel_all_downloads(self) -> bool:
|
||
"""Cancel and remove ALL downloads (active + completed) from slskd.
|
||
|
||
Lists all current downloads and cancels each one individually,
|
||
since slskd has no bulk cancel endpoint.
|
||
|
||
Returns:
|
||
bool: True if successful, False otherwise
|
||
"""
|
||
if not self.base_url:
|
||
logger.debug("Soulseek client not configured")
|
||
return False
|
||
|
||
try:
|
||
# Get all current downloads grouped by user
|
||
response = await self._make_request('GET', 'transfers/downloads')
|
||
if not response:
|
||
logger.info("No downloads to cancel")
|
||
return True
|
||
|
||
from urllib.parse import quote
|
||
cancelled = 0
|
||
failed = 0
|
||
|
||
for user_data in response:
|
||
username = user_data.get('username', '')
|
||
if not username:
|
||
continue
|
||
for directory in user_data.get('directories', []):
|
||
for file_data in directory.get('files', []):
|
||
file_id = file_data.get('id', '')
|
||
if not file_id:
|
||
continue
|
||
encoded_id = quote(str(file_id), safe='')
|
||
endpoint = f'transfers/downloads/{username}/{encoded_id}?remove=true'
|
||
result = await self._make_request('DELETE', endpoint)
|
||
if result is not None:
|
||
cancelled += 1
|
||
else:
|
||
failed += 1
|
||
|
||
if failed:
|
||
logger.warning(f"Cancelled {cancelled} downloads, {failed} failed")
|
||
else:
|
||
logger.info(f"Successfully cancelled {cancelled} downloads from slskd")
|
||
|
||
return failed == 0 or cancelled > 0
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error cancelling all downloads: {e}")
|
||
return False
|
||
|
||
async def clear_all_completed_downloads(self) -> bool:
|
||
"""Clear all completed/finished downloads from slskd backend
|
||
|
||
Uses the /api/v0/transfers/downloads/all/completed endpoint to remove
|
||
all downloads with completed, cancelled, or failed status from slskd.
|
||
|
||
Returns:
|
||
bool: True if clearing was successful, False otherwise
|
||
"""
|
||
if not self.base_url:
|
||
logger.debug("Soulseek client not configured")
|
||
return False
|
||
|
||
try:
|
||
endpoint = 'transfers/downloads/all/completed'
|
||
logger.debug(f"Clearing all completed downloads with endpoint: {endpoint}")
|
||
response = await self._make_request('DELETE', endpoint)
|
||
success = response is not None
|
||
|
||
if success:
|
||
logger.info("Successfully cleared all completed downloads from slskd")
|
||
else:
|
||
logger.error("Failed to clear completed downloads from slskd")
|
||
|
||
return success
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error clearing completed downloads: {e}")
|
||
return False
|
||
|
||
async def get_all_searches(self) -> List[dict]:
|
||
"""Get all search history from slskd
|
||
|
||
Returns:
|
||
List[dict]: List of search objects from slskd API, empty list if error
|
||
"""
|
||
if not self.base_url:
|
||
logger.debug("Soulseek client not configured")
|
||
return []
|
||
|
||
try:
|
||
endpoint = 'searches'
|
||
logger.debug(f"Getting all searches with endpoint: {endpoint}")
|
||
response = await self._make_request('GET', endpoint)
|
||
|
||
if response is not None:
|
||
searches = response if isinstance(response, list) else []
|
||
logger.info(f"Retrieved {len(searches)} searches from slskd")
|
||
return searches
|
||
else:
|
||
logger.error("Failed to retrieve searches from slskd")
|
||
return []
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error retrieving searches: {e}")
|
||
return []
|
||
|
||
async def delete_search(self, search_id: str) -> bool:
|
||
"""Delete a specific search from slskd history
|
||
|
||
Args:
|
||
search_id: The ID of the search to delete
|
||
|
||
Returns:
|
||
bool: True if deletion was successful, False otherwise
|
||
"""
|
||
if not self.base_url:
|
||
logger.debug("Soulseek client not configured")
|
||
return False
|
||
|
||
try:
|
||
endpoint = f'searches/{search_id}'
|
||
logger.debug(f"Deleting search {search_id} with endpoint: {endpoint}")
|
||
response = await self._make_request('DELETE', endpoint)
|
||
success = response is not None
|
||
|
||
if success:
|
||
logger.debug(f"Successfully deleted search {search_id}")
|
||
else:
|
||
# Don't log warnings for failed deletions - they're often just 404s for already-removed searches
|
||
logger.debug(f"Search deletion returned false (likely already removed): {search_id}")
|
||
|
||
return success
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error deleting search {search_id}: {e}")
|
||
return False
|
||
|
||
async def clear_all_searches(self) -> bool:
|
||
"""Clear all search history from slskd
|
||
|
||
Returns:
|
||
bool: True if all searches were cleared successfully, False otherwise
|
||
"""
|
||
if not self.base_url:
|
||
logger.debug("Soulseek client not configured")
|
||
return False
|
||
|
||
try:
|
||
# Get all searches first
|
||
searches = await self.get_all_searches()
|
||
|
||
if not searches:
|
||
logger.info("No searches found to clear")
|
||
return True
|
||
|
||
logger.info(f"Clearing {len(searches)} searches from slskd...")
|
||
|
||
# Delete each search individually
|
||
deleted_count = 0
|
||
failed_count = 0
|
||
|
||
for search in searches:
|
||
search_id = search.get('id')
|
||
if search_id:
|
||
success = await self.delete_search(search_id)
|
||
if success:
|
||
deleted_count += 1
|
||
else:
|
||
failed_count += 1
|
||
else:
|
||
logger.warning("Search found without ID, skipping")
|
||
failed_count += 1
|
||
|
||
logger.info(f"Search cleanup complete: {deleted_count} deleted, {failed_count} failed")
|
||
return failed_count == 0
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error clearing all searches: {e}")
|
||
return False
|
||
|
||
async def maintain_search_history(self, max_searches: int = 50) -> bool:
|
||
"""Maintain a rolling window of recent searches by deleting oldest when over limit
|
||
|
||
Args:
|
||
max_searches: Maximum number of searches to keep (default: 50)
|
||
|
||
Returns:
|
||
bool: True if maintenance was successful, False otherwise
|
||
"""
|
||
if not self.base_url:
|
||
logger.debug("Soulseek client not configured, skipping search maintenance")
|
||
return False
|
||
|
||
try:
|
||
# Get all searches (should be ordered by creation time, oldest first)
|
||
searches = await self.get_all_searches()
|
||
|
||
if len(searches) <= max_searches:
|
||
logger.debug(f"Search count ({len(searches)}) within limit ({max_searches}), no maintenance needed")
|
||
return True
|
||
|
||
# Calculate how many to delete
|
||
excess_count = len(searches) - max_searches
|
||
oldest_searches = searches[:excess_count] # Get the oldest ones
|
||
|
||
logger.info(f"Maintaining search history: deleting {excess_count} oldest searches (keeping {max_searches})")
|
||
|
||
# Delete the oldest searches
|
||
deleted_count = 0
|
||
failed_count = 0
|
||
|
||
for search in oldest_searches:
|
||
search_id = search.get('id')
|
||
if search_id:
|
||
success = await self.delete_search(search_id)
|
||
if success:
|
||
deleted_count += 1
|
||
else:
|
||
failed_count += 1
|
||
else:
|
||
logger.warning("Search found without ID during maintenance, skipping")
|
||
failed_count += 1
|
||
|
||
logger.info(f"Search maintenance complete: {deleted_count} deleted, {failed_count} failed")
|
||
return failed_count == 0
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error during search history maintenance: {e}")
|
||
return False
|
||
|
||
async def maintain_search_history_with_buffer(self, keep_searches: int = 50, trigger_threshold: int = 200) -> bool:
|
||
"""Maintain search history with a buffer - only clean when searches exceed threshold
|
||
|
||
Args:
|
||
keep_searches: Number of searches to keep after cleanup (default: 50)
|
||
trigger_threshold: Only trigger cleanup when search count exceeds this (default: 200)
|
||
|
||
Returns:
|
||
bool: True if maintenance was successful or not needed, False otherwise
|
||
"""
|
||
if not self.base_url:
|
||
logger.debug("Soulseek client not configured, skipping search maintenance")
|
||
return False
|
||
|
||
try:
|
||
# Get all searches
|
||
searches = await self.get_all_searches()
|
||
|
||
if len(searches) <= trigger_threshold:
|
||
logger.debug(f"Search count ({len(searches)}) below trigger threshold ({trigger_threshold}), no maintenance needed")
|
||
return True
|
||
|
||
# Calculate how many to delete (keep only the most recent ones)
|
||
excess_count = len(searches) - keep_searches
|
||
oldest_searches = searches[:excess_count] # Get the oldest ones to delete
|
||
|
||
logger.info(f"Search buffer exceeded: {len(searches)} searches > {trigger_threshold} threshold. Deleting {excess_count} oldest searches (keeping {keep_searches})")
|
||
|
||
# Delete the oldest searches
|
||
deleted_count = 0
|
||
failed_count = 0
|
||
|
||
for search in oldest_searches:
|
||
search_id = search.get('id')
|
||
if search_id:
|
||
success = await self.delete_search(search_id)
|
||
if success:
|
||
deleted_count += 1
|
||
else:
|
||
failed_count += 1
|
||
else:
|
||
logger.warning("Search found without ID during maintenance, skipping")
|
||
failed_count += 1
|
||
|
||
logger.info(f"Search buffer maintenance complete: {deleted_count} deleted, {failed_count} failed, {keep_searches} searches remaining")
|
||
return failed_count == 0
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error during search history buffer maintenance: {e}")
|
||
return False
|
||
|
||
async def search_and_download_best(self, query: str) -> Optional[str]:
|
||
results = await self.search(query)
|
||
|
||
if not results:
|
||
logger.warning(f"No results found for: {query}")
|
||
return None
|
||
|
||
# Use quality profile filtering
|
||
filtered_results = self.filter_results_by_quality_preference(results)
|
||
|
||
if not filtered_results:
|
||
logger.warning(f"No suitable quality results found for: {query}")
|
||
return None
|
||
|
||
best_result = filtered_results[0]
|
||
quality_info = f"{best_result.quality.upper()}"
|
||
if best_result.bitrate:
|
||
quality_info += f" {best_result.bitrate}kbps"
|
||
|
||
logger.info(f"Downloading: {best_result.filename} ({quality_info}) from {best_result.username}")
|
||
return await self.download(best_result.username, best_result.filename, best_result.size)
|
||
|
||
def download_album_to_staging(
|
||
self,
|
||
album_name: str,
|
||
artist_name: str,
|
||
staging_dir: str,
|
||
progress_callback=None,
|
||
*,
|
||
preferred_source: Optional[Dict[str, Any]] = None,
|
||
preferred_tracks: Optional[List[TrackResult]] = None,
|
||
) -> Dict[str, Any]:
|
||
"""One-shot Soulseek album download.
|
||
|
||
Search for one album folder, enqueue files from that single
|
||
``username + folder_path``, wait for slskd to report completion,
|
||
then copy completed files into the private album-bundle staging
|
||
directory. If the folder cannot be selected or enqueued cleanly,
|
||
callers may fall back to the existing per-track Soulseek flow.
|
||
Once files are staged, the per-track staging matcher owns final
|
||
import, same as torrent / usenet album bundles.
|
||
"""
|
||
result: Dict[str, Any] = {
|
||
'success': False,
|
||
'files': [],
|
||
'error': None,
|
||
'fallback': True,
|
||
'partial': False,
|
||
}
|
||
if not self.is_configured():
|
||
result['error'] = 'Soulseek source not configured'
|
||
return result
|
||
|
||
def _emit(state: str, **extra) -> None:
|
||
if progress_callback:
|
||
try:
|
||
progress_callback({'state': state, **extra})
|
||
except Exception as cb_exc:
|
||
logger.debug("[Soulseek album] progress callback failed: %s", cb_exc)
|
||
|
||
picked = None
|
||
folder_tracks = list(preferred_tracks or [])
|
||
username = (preferred_source or {}).get('username', '') if preferred_source else ''
|
||
folder_path = (preferred_source or {}).get('folder_path', '') if preferred_source else ''
|
||
if username and folder_path:
|
||
logger.info(
|
||
"[Soulseek album] Using preflight-selected folder %s:%s",
|
||
username,
|
||
folder_path,
|
||
)
|
||
_emit('searching', query=f"{artist_name} {album_name}".strip(), release=folder_path)
|
||
else:
|
||
query = f"{artist_name} {album_name}".strip()
|
||
_emit('searching', query=query)
|
||
try:
|
||
_, albums = run_async(self.search(query, timeout=30))
|
||
except Exception as exc:
|
||
result['error'] = f'Soulseek album search failed: {exc}'
|
||
return result
|
||
|
||
if not albums:
|
||
result['error'] = 'No complete Soulseek album folders found'
|
||
return result
|
||
|
||
picked = self._pick_album_bundle_folder(albums, album_name, artist_name)
|
||
if picked is None:
|
||
result['error'] = 'No suitable Soulseek album folder after filtering'
|
||
return result
|
||
|
||
folder_path = getattr(picked, 'album_path', '') or ''
|
||
username = getattr(picked, 'username', '') or ''
|
||
if not username or not folder_path:
|
||
result['error'] = 'No suitable Soulseek album folder after filtering'
|
||
return result
|
||
|
||
# On the preflight-reuse path ``picked`` is None — the master
|
||
# already selected the folder so we never call _pick_album_bundle_folder.
|
||
# Read the track count off the preferred_tracks list in that
|
||
# case so the log line doesn't misleadingly report "0 tracks".
|
||
_log_track_count = (
|
||
getattr(picked, 'track_count', 0) if picked is not None
|
||
else len(folder_tracks)
|
||
)
|
||
_log_quality = getattr(picked, 'dominant_quality', '') if picked is not None else ''
|
||
logger.info(
|
||
"[Soulseek album] Picked %s:%s (%s tracks, quality=%s)",
|
||
username,
|
||
folder_path,
|
||
_log_track_count,
|
||
_log_quality,
|
||
)
|
||
_emit(
|
||
'queued',
|
||
release=getattr(picked, 'album_title', folder_path) if picked else folder_path,
|
||
count=getattr(picked, 'track_count', 0) if picked else len(folder_tracks),
|
||
)
|
||
|
||
if not folder_tracks:
|
||
try:
|
||
browse_files = run_async(self.browse_user_directory(username, folder_path))
|
||
except Exception as exc:
|
||
result['error'] = f'Soulseek folder browse failed: {exc}'
|
||
return result
|
||
|
||
if not browse_files:
|
||
result['error'] = 'Could not browse selected Soulseek album folder'
|
||
return result
|
||
|
||
folder_tracks = self.parse_browse_results_to_tracks(
|
||
username,
|
||
browse_files,
|
||
directory=folder_path,
|
||
)
|
||
folder_tracks = self.filter_results_by_quality_preference(folder_tracks)
|
||
if not folder_tracks:
|
||
result['error'] = 'Selected Soulseek album folder contained no audio files'
|
||
return result
|
||
|
||
transfer_keys: Dict[tuple, TrackResult] = {}
|
||
_emit(
|
||
'downloading',
|
||
release=getattr(picked, 'album_title', folder_path) if picked else folder_path,
|
||
count=len(folder_tracks),
|
||
)
|
||
for track in folder_tracks:
|
||
try:
|
||
download_id = run_async(self.download(track.username, track.filename, track.size))
|
||
except Exception as exc:
|
||
logger.warning("[Soulseek album] Failed to enqueue %s: %s", track.filename, exc)
|
||
continue
|
||
if download_id:
|
||
transfer_keys[(track.username, track.filename)] = track
|
||
|
||
if not transfer_keys:
|
||
result['error'] = 'No Soulseek album files could be enqueued'
|
||
return result
|
||
|
||
result['fallback'] = False
|
||
completed = self._poll_album_bundle_downloads(transfer_keys, _emit)
|
||
if not completed:
|
||
# The selected folder yielded ZERO usable files — every transfer
|
||
# failed / aborted / stalled (a dead or unwilling peer). Don't hard-
|
||
# fail the batch: fall back to the per-track flow, which searches ALL
|
||
# sources per track and can pull each from a live peer. We reuse that
|
||
# proven multi-source robustness instead of looping candidate folders
|
||
# here. (Per-track only fires for a genuinely-missing album anyway.)
|
||
result['error'] = ('Soulseek album folder produced no usable files '
|
||
'(peer failed/aborted/stalled) — falling back to per-track')
|
||
result['fallback'] = True
|
||
return result
|
||
|
||
_emit('staging', release=getattr(picked, 'album_title', folder_path) if picked else folder_path)
|
||
# remove_source=True: clean slskd's completed files once staged so they
|
||
# don't pile up in the download folder (#796). Soulseek has no seeding,
|
||
# unlike the torrent/usenet bundle paths which keep their originals.
|
||
copied = copy_audio_files_atomically(completed, Path(staging_dir), remove_source=True)
|
||
if not copied:
|
||
result['error'] = 'No Soulseek album files copied to staging'
|
||
return result
|
||
|
||
partial = len(copied) < len(transfer_keys)
|
||
if partial:
|
||
logger.warning(
|
||
"[Soulseek album] Staged partial album for '%s': %d/%d files",
|
||
album_name,
|
||
len(copied),
|
||
len(transfer_keys),
|
||
)
|
||
else:
|
||
logger.info("[Soulseek album] Staged %d files for '%s'", len(copied), album_name)
|
||
_emit('staged', count=len(copied))
|
||
result['success'] = True
|
||
result['files'] = copied
|
||
result['partial'] = partial
|
||
result['expected_count'] = len(transfer_keys)
|
||
result['completed_count'] = len(copied)
|
||
return result
|
||
|
||
def _pick_album_bundle_folder(
|
||
self,
|
||
albums: List[AlbumResult],
|
||
album_name: str,
|
||
artist_name: str,
|
||
) -> Optional[AlbumResult]:
|
||
scored = []
|
||
for album in albums:
|
||
tracks = self.filter_results_by_quality_preference(list(getattr(album, 'tracks', []) or []))
|
||
if not tracks:
|
||
continue
|
||
album_text = f"{getattr(album, 'album_title', '')} {getattr(album, 'album_path', '')}"
|
||
artist_text = f"{getattr(album, 'artist', '')} {getattr(album, 'album_path', '')}"
|
||
album_score = self._bundle_similarity(album_name, album_text)
|
||
artist_score = self._bundle_similarity(artist_name, artist_text)
|
||
track_count = int(getattr(album, 'track_count', 0) or len(tracks))
|
||
count_score = 1.0 if track_count >= 3 else 0.35
|
||
score = (
|
||
album_score * 0.42
|
||
+ artist_score * 0.22
|
||
+ count_score * 0.12
|
||
+ min(1.0, len(tracks) / max(1, track_count)) * 0.12
|
||
+ float(getattr(album, 'quality_score', 0.0) or 0.0) * 0.12
|
||
)
|
||
scored.append((score, len(tracks), album))
|
||
if not scored:
|
||
return None
|
||
scored.sort(key=lambda row: (row[0], row[1], getattr(row[2], 'quality_score', 0.0)), reverse=True)
|
||
best_score, _, best = scored[0]
|
||
if best_score < 0.58:
|
||
logger.warning("[Soulseek album] Best folder score %.3f below threshold", best_score)
|
||
return None
|
||
return best
|
||
|
||
@staticmethod
|
||
def _bundle_similarity(expected: Any, actual: Any) -> float:
|
||
import re
|
||
from difflib import SequenceMatcher
|
||
left = re.sub(r'[^a-z0-9]+', ' ', str(expected or '').lower()).strip()
|
||
right = re.sub(r'[^a-z0-9]+', ' ', str(actual or '').lower()).strip()
|
||
if not left or not right:
|
||
return 0.0
|
||
if left == right:
|
||
return 1.0
|
||
left_words = set(left.split())
|
||
right_words = set(right.split())
|
||
if left_words and left_words.issubset(right_words):
|
||
return 0.92
|
||
if right_words and right_words.issubset(left_words):
|
||
return 0.86
|
||
if left in right or right in left:
|
||
return min(len(left), len(right)) / max(len(left), len(right))
|
||
return SequenceMatcher(None, left, right).ratio()
|
||
|
||
def _poll_album_bundle_downloads(self, transfer_keys: Dict[tuple, TrackResult], emit) -> List[Path]:
|
||
deadline = time.monotonic() + get_poll_timeout()
|
||
interval = get_poll_interval()
|
||
completed_paths: Dict[tuple, Path] = {}
|
||
failed_states: Dict[tuple, str] = {}
|
||
# Track keys where slskd reports the transfer Completed /
|
||
# Succeeded but the local file finder can't yet locate the
|
||
# file on disk. Usually transient (slskd writes the file
|
||
# after announcing completion); becomes a hard failure when
|
||
# ALL remaining keys land here long enough — that's the
|
||
# symptom from issue #715 (Billy Ocean bundle hung 22 min
|
||
# after slskd finished). The grace window keeps the
|
||
# transient case from triggering; the all-stuck check
|
||
# short-circuits when there's no chance of progress.
|
||
unresolved_since: Dict[tuple, float] = {}
|
||
# Seconds an "slskd Completed but locally unresolved" key
|
||
# has to stay stuck before we give up on it.
|
||
_unresolved_grace = 45.0
|
||
# Bundle-level stall guard. The #715 grace above only covers
|
||
# "slskd says Completed but the file isn't on disk yet". It does NOT
|
||
# cover a transfer the peer stalls on — stuck InProgress / Queued, or
|
||
# dropped by slskd entirely — which is never failed, never completed,
|
||
# and never marked unresolved, so it blocks BOTH the all-terminal
|
||
# finish check AND the grace exit, and the poll spun to the full
|
||
# ``get_poll_timeout()`` deadline (the Slipknot hang). If NOTHING
|
||
# progresses — no transfer completes/fails and no pending transfer's
|
||
# byte count moves — for this long, the folder has stalled: mark the
|
||
# stuck transfers failed so the bundle resolves with whatever
|
||
# completed (the per-track matcher then handles the missing tracks).
|
||
# Conservative on purpose: only trips when EVERYTHING is frozen, so a
|
||
# slow-but-progressing or still-queued-then-starting transfer is safe.
|
||
_stall_grace = 180.0
|
||
_last_progress_marker = None
|
||
_last_progress_at = time.monotonic()
|
||
while time.monotonic() < deadline:
|
||
try:
|
||
downloads = run_async(self.get_all_downloads())
|
||
except Exception as exc:
|
||
logger.warning("[Soulseek album] Poll error: %s", exc)
|
||
downloads = []
|
||
|
||
by_key = {}
|
||
for dl in downloads:
|
||
exact_key = (dl.username, dl.filename)
|
||
by_key[exact_key] = dl
|
||
basename_key = (
|
||
dl.username,
|
||
os.path.basename((dl.filename or '').replace('\\', '/')),
|
||
)
|
||
by_key.setdefault(basename_key, dl)
|
||
for key, track in transfer_keys.items():
|
||
if key in completed_paths or key in failed_states:
|
||
continue
|
||
dl = by_key.get(key) or by_key.get((
|
||
key[0],
|
||
os.path.basename((key[1] or '').replace('\\', '/')),
|
||
))
|
||
state = (getattr(dl, 'state', '') or '') if dl else ''
|
||
# NOTE: check failure tokens BEFORE the 'Completed' branch — slskd
|
||
# reports terminal failures as "Completed, <reason>" (e.g.
|
||
# "Completed, Aborted" / "Completed, Cancelled" when a peer accepts
|
||
# then drops every transfer at 0 bytes). Those contain "Completed",
|
||
# so without catching the failure reason first they'd be misread as
|
||
# "completed but file missing" (the #715 download_path path).
|
||
if any(token in state for token in
|
||
('Errored', 'Failed', 'Rejected', 'TimedOut', 'Aborted', 'Cancelled')):
|
||
failed_states[key] = state or 'Failed'
|
||
logger.warning(
|
||
"[Soulseek album] Transfer failed from selected folder: %s (%s)",
|
||
os.path.basename((track.filename or '').replace('\\', '/')),
|
||
failed_states[key],
|
||
)
|
||
continue
|
||
if dl and ('Completed' in state or 'Succeeded' in state):
|
||
if dl.size and dl.transferred and dl.transferred < dl.size:
|
||
continue
|
||
path = self._resolve_downloaded_album_file(track.filename)
|
||
if path:
|
||
completed_paths[key] = path
|
||
unresolved_since.pop(key, None)
|
||
else:
|
||
# First time we see slskd report this key as
|
||
# completed-but-locally-missing, stamp it.
|
||
# Subsequent iterations keep the original
|
||
# stamp so the grace window is real wall-time.
|
||
unresolved_since.setdefault(key, time.monotonic())
|
||
logger.debug(
|
||
"[Soulseek album] Transfer completed but local file not found yet: %s",
|
||
track.filename,
|
||
)
|
||
emit(
|
||
'downloading',
|
||
progress=round(len(completed_paths) / max(1, len(transfer_keys)) * 100, 1),
|
||
count=len(completed_paths),
|
||
failed=len(failed_states),
|
||
)
|
||
|
||
# Bundle-level stall detection (see ``_stall_grace`` above). Advance
|
||
# the progress marker on ANY forward motion — a transfer reaching a
|
||
# terminal state, or a still-pending transfer downloading more bytes.
|
||
# If the marker is frozen for ``_stall_grace``, the peer has stalled;
|
||
# mark the stuck transfers failed so the finish/all-failed checks
|
||
# below resolve the bundle instead of spinning to the deadline.
|
||
now = time.monotonic()
|
||
stall_pending = [
|
||
k for k in transfer_keys
|
||
if k not in completed_paths and k not in failed_states
|
||
]
|
||
pending_bytes = 0
|
||
for k in stall_pending:
|
||
dl = by_key.get(k) or by_key.get((
|
||
k[0], os.path.basename((k[1] or '').replace('\\', '/')),
|
||
))
|
||
pending_bytes += (getattr(dl, 'transferred', 0) or 0) if dl else 0
|
||
marker = (len(completed_paths) + len(failed_states), pending_bytes)
|
||
if marker != _last_progress_marker:
|
||
_last_progress_marker = marker
|
||
_last_progress_at = now
|
||
elif stall_pending and (now - _last_progress_at) >= _stall_grace:
|
||
logger.warning(
|
||
"[Soulseek album] No progress for %.0fs — peer stalled on %d "
|
||
"transfer(s) (stuck / queued / dropped). Marking them failed and "
|
||
"resolving with what completed; missing tracks fall back to "
|
||
"per-track. Stalled: %s",
|
||
_stall_grace, len(stall_pending),
|
||
[transfer_keys[k].filename for k in stall_pending[:5]],
|
||
)
|
||
for k in stall_pending:
|
||
failed_states[k] = 'Stalled'
|
||
|
||
if completed_paths and len(completed_paths) + len(failed_states) == len(transfer_keys):
|
||
logger.warning(
|
||
"[Soulseek album] Selected folder finished with %d completed and %d failed transfer(s)",
|
||
len(completed_paths),
|
||
len(failed_states),
|
||
)
|
||
return list(completed_paths.values())
|
||
if not completed_paths and len(failed_states) == len(transfer_keys):
|
||
logger.warning("[Soulseek album] All %d transfer(s) failed from selected folder", len(failed_states))
|
||
return []
|
||
if len(completed_paths) == len(transfer_keys):
|
||
return list(completed_paths.values())
|
||
|
||
# Early exit when every remaining key is "slskd done +
|
||
# locally unresolved past grace". Pre-fix this was the
|
||
# silent timeout path from issue #715 — slskd finished
|
||
# downloading the whole album, but no local file ever
|
||
# resolved, so the poll spun until ``get_poll_timeout()``
|
||
# elapsed (default 30+ minutes) before failing the batch.
|
||
now = time.monotonic()
|
||
still_pending = [
|
||
k for k in transfer_keys
|
||
if k not in completed_paths and k not in failed_states
|
||
]
|
||
if still_pending and all(
|
||
k in unresolved_since and (now - unresolved_since[k]) >= _unresolved_grace
|
||
for k in still_pending
|
||
):
|
||
logger.error(
|
||
"[Soulseek album] %d transfer(s) reported Completed by slskd "
|
||
"but no local file could be resolved after %.0fs — likely a "
|
||
"``soulseek.download_path`` mismatch (Docker volume / "
|
||
"username-prefixed slskd config). Files this poll attempted: %s",
|
||
len(still_pending),
|
||
_unresolved_grace,
|
||
[transfer_keys[k].filename for k in still_pending[:5]],
|
||
)
|
||
return list(completed_paths.values())
|
||
|
||
time.sleep(interval)
|
||
pending = len(transfer_keys) - len(completed_paths) - len(failed_states)
|
||
if completed_paths:
|
||
logger.warning(
|
||
"[Soulseek album] Timed out with partial album: %d completed, %d failed, %d pending",
|
||
len(completed_paths),
|
||
len(failed_states),
|
||
pending,
|
||
)
|
||
return list(completed_paths.values())
|
||
logger.error(
|
||
"[Soulseek album] Timed out waiting for %d album files (%d failed, %d pending)",
|
||
len(transfer_keys),
|
||
len(failed_states),
|
||
pending,
|
||
)
|
||
return []
|
||
|
||
def _resolve_downloaded_album_file(self, remote_filename: str) -> Optional[Path]:
|
||
# Pre-fix this tried three hardcoded candidate paths and
|
||
# silently returned None on anything else — including the
|
||
# common slskd config that nests downloads under
|
||
# ``<download_dir>/<username>/<filename>``. That mismatch
|
||
# caused issue #715: bundle downloads on those setups
|
||
# timed out 22 minutes after slskd reported every transfer
|
||
# Completed because the resolver never located a single
|
||
# local file.
|
||
#
|
||
# Now delegates to the shared robust finder which recursively
|
||
# walks the download dir by basename + path-confirms via the
|
||
# remote directory components. Same logic the per-track flow
|
||
# has used since 2.5.9.
|
||
from core.downloads.file_finder import find_completed_audio_file
|
||
basename = os.path.basename((remote_filename or '').replace('\\', '/'))
|
||
if not basename:
|
||
return None
|
||
# Fast path: the three hardcoded candidates still cover the
|
||
# default slskd-flat layout cheaply, and avoid an os.walk
|
||
# for the common case. Walk only when none hit.
|
||
candidates = [
|
||
self.download_path / remote_filename,
|
||
self.download_path / basename,
|
||
]
|
||
normalized_parts = [p for p in remote_filename.replace('\\', '/').split('/') if p]
|
||
if normalized_parts:
|
||
candidates.append(self.download_path.joinpath(*normalized_parts))
|
||
for candidate in candidates:
|
||
try:
|
||
if candidate.exists() and candidate.is_file():
|
||
return candidate
|
||
except OSError:
|
||
continue
|
||
|
||
found_path, _location = find_completed_audio_file(
|
||
str(self.download_path), remote_filename,
|
||
)
|
||
return Path(found_path) if found_path else None
|
||
|
||
async def check_connection(self) -> bool:
|
||
"""Check if slskd is running and connected to the Soulseek network"""
|
||
if not self.base_url:
|
||
return False
|
||
|
||
try:
|
||
# Primary check: server/state tells us if slskd is connected to the Soulseek network
|
||
state = await self._make_request('GET', 'server/state')
|
||
if state is not None:
|
||
is_connected = state.get('isConnected') or state.get('IsConnected', False)
|
||
is_logged_in = state.get('isLoggedIn') or state.get('IsLoggedIn', False)
|
||
if not (is_connected and is_logged_in):
|
||
logger.debug(f"Soulseek not fully connected: isConnected={is_connected}, isLoggedIn={is_logged_in}")
|
||
return is_connected and is_logged_in
|
||
|
||
# Fallback: if server/state endpoint unavailable (older slskd), check API reachability
|
||
logger.debug("server/state endpoint unavailable, falling back to session check")
|
||
response = await self._make_request('GET', 'session')
|
||
return response is not None
|
||
except Exception as e:
|
||
logger.debug(f"Connection check failed: {e}")
|
||
return False
|
||
|
||
@staticmethod
|
||
def _calculate_effective_kbps(size_bytes: int, duration_ms: Optional[int]) -> Optional[float]:
|
||
"""Calculate effective bitrate in kbps from file size and duration."""
|
||
if not duration_ms or duration_ms <= 0 or not size_bytes or size_bytes <= 0:
|
||
return None
|
||
duration_seconds = duration_ms / 1000.0
|
||
return (size_bytes * 8) / duration_seconds / 1000.0
|
||
|
||
# Internal fallback size limits (MB) when duration is unavailable — generous to catch only extreme outliers
|
||
_FALLBACK_SIZE_LIMITS = {
|
||
'flac': (1, 500),
|
||
'mp3_320': (1, 50),
|
||
'mp3_256': (1, 40),
|
||
'mp3_192': (1, 30),
|
||
'aac': (1, 50),
|
||
'other': (0, 500),
|
||
}
|
||
|
||
def _drop_quarantined_sources(self, results: List[TrackResult]) -> List[TrackResult]:
|
||
"""Filter out candidates whose `(username, filename)` is on the
|
||
quarantine record. Issue #652.
|
||
|
||
Reads quarantine sidecars fresh each call so newly-quarantined
|
||
sources are honored immediately on the next search — no client
|
||
state to invalidate. Filesystem cost is bounded (one listdir +
|
||
N small JSON reads) and dwarfed by the Soulseek search itself.
|
||
|
||
Returns the input list unchanged when the quarantine directory
|
||
is absent, empty, or unreadable — i.e. defaults to today's
|
||
behaviour if anything goes wrong on the dedup path.
|
||
"""
|
||
try:
|
||
from core.imports.quarantine import get_quarantined_source_keys
|
||
download_path = config_manager.get('soulseek.download_path', './downloads')
|
||
quarantine_dir = os.path.join(download_path, 'ss_quarantine')
|
||
blocked = get_quarantined_source_keys(quarantine_dir)
|
||
except Exception as exc:
|
||
logger.debug("quarantine dedup: failed to load source keys, skipping filter: %s", exc)
|
||
return results
|
||
|
||
if not blocked:
|
||
return results
|
||
|
||
kept: List[TrackResult] = []
|
||
skipped = 0
|
||
for candidate in results:
|
||
key = (candidate.username or '', candidate.filename or '')
|
||
if key in blocked:
|
||
skipped += 1
|
||
continue
|
||
kept.append(candidate)
|
||
|
||
if skipped:
|
||
logger.info(
|
||
f"Quarantine dedup: dropped {skipped} candidate(s) matching previously-quarantined sources; "
|
||
f"{len(kept)} remain"
|
||
)
|
||
return kept
|
||
|
||
def filter_results_by_quality_preference(self, results: List[TrackResult]) -> List[TrackResult]:
|
||
"""Filter and rank candidates using the global quality target list.
|
||
|
||
Replaces the old bucket+heuristic approach with ``core.quality.model``
|
||
so every download source shares the same ranking logic.
|
||
|
||
Issue #652: also drops candidates whose ``(username, filename)``
|
||
matches a previously-quarantined download to break infinite retry loops.
|
||
"""
|
||
from database.music_database import MusicDatabase
|
||
|
||
if not results:
|
||
return []
|
||
|
||
# Issue #652: drop candidates on the quarantine record BEFORE ranking,
|
||
# so a previously-quarantined source can't win the quality picker by
|
||
# superior bitrate and re-trigger the same failed download in a loop.
|
||
results = self._drop_quarantined_sources(results)
|
||
if not results:
|
||
return []
|
||
|
||
db = MusicDatabase()
|
||
profile = db.get_quality_profile()
|
||
|
||
# Build ranked target list — v3 profiles carry it directly;
|
||
# v2 profiles are converted on the fly (no DB write needed here).
|
||
raw_targets = profile.get('ranked_targets')
|
||
if not raw_targets and 'qualities' in profile:
|
||
raw_targets = v2_qualities_to_ranked_targets(profile['qualities'])
|
||
|
||
targets = [QualityTarget.from_dict(t) for t in (raw_targets or [])]
|
||
fallback_enabled = profile.get('fallback_enabled', True)
|
||
|
||
# Every format (AAC included) follows the SAME universal rule: a
|
||
# candidate passes only if it matches a ranked target; if nothing
|
||
# matches, the fallback toggle decides. No per-format special-casing.
|
||
|
||
logger.debug(
|
||
"Quality Filter: profile='%s', %d targets, %d candidates",
|
||
profile.get('preset', 'custom'), len(targets), len(results),
|
||
)
|
||
|
||
ranked = filter_and_rank(results, targets, fallback_enabled=fallback_enabled)
|
||
|
||
if ranked:
|
||
best_label = ranked[0].audio_quality.label()
|
||
logger.info("Quality Filter: returning %d candidate(s), best=%s", len(ranked), best_label)
|
||
else:
|
||
logger.warning("Quality Filter: no candidates passed quality constraints")
|
||
|
||
return ranked
|
||
|
||
async def get_session_info(self) -> Optional[Dict[str, Any]]:
|
||
"""Get slskd session information including version"""
|
||
if not self.base_url:
|
||
return None
|
||
|
||
try:
|
||
response = await self._make_request('GET', 'session')
|
||
if response:
|
||
logger.info(f"slskd session info: {response}")
|
||
return response
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"Error getting session info: {e}")
|
||
return None
|
||
|
||
async def explore_api_endpoints(self) -> Dict[str, Any]:
|
||
"""Explore available API endpoints to find the correct download endpoint"""
|
||
if not self.base_url:
|
||
return {}
|
||
|
||
try:
|
||
logger.info("Exploring slskd API endpoints...")
|
||
|
||
# Try to get Swagger/OpenAPI documentation
|
||
swagger_url = f"{self.base_url}/swagger/v1/swagger.json"
|
||
|
||
session = aiohttp.ClientSession(timeout=_SLSKD_DEFAULT_TIMEOUT)
|
||
try:
|
||
headers = self._get_headers()
|
||
async with session.get(swagger_url, headers=headers) as response:
|
||
if response.status == 200:
|
||
swagger_data = await response.json()
|
||
logger.info("Found Swagger documentation")
|
||
|
||
# Look for download/transfer related endpoints
|
||
paths = swagger_data.get('paths', {})
|
||
download_endpoints = {}
|
||
|
||
for path, methods in paths.items():
|
||
if any(keyword in path.lower() for keyword in ['download', 'transfer', 'enqueue']):
|
||
download_endpoints[path] = methods
|
||
logger.info(f"Found endpoint: {path} with methods: {list(methods.keys())}")
|
||
|
||
return {
|
||
'swagger_available': True,
|
||
'download_endpoints': download_endpoints,
|
||
'base_url': self.base_url
|
||
}
|
||
else:
|
||
logger.debug(f"Swagger endpoint returned {response.status}")
|
||
except Exception as e:
|
||
logger.debug(f"Could not access Swagger docs: {e}")
|
||
finally:
|
||
await session.close()
|
||
|
||
# If Swagger is not available, try common endpoints manually
|
||
logger.info("Swagger not available, testing common endpoints...")
|
||
|
||
common_endpoints = [
|
||
'transfers',
|
||
'downloads',
|
||
'transfers/downloads',
|
||
'api/transfers',
|
||
'api/downloads'
|
||
]
|
||
|
||
available_endpoints = {}
|
||
|
||
for endpoint in common_endpoints:
|
||
try:
|
||
response = await self._make_request('GET', endpoint)
|
||
if response is not None:
|
||
available_endpoints[endpoint] = 'GET available'
|
||
logger.info(f"[OK] Endpoint available: {endpoint}")
|
||
else:
|
||
# Try different endpoints without /api/v0 prefix
|
||
simple_url = f"{self.base_url}/{endpoint}"
|
||
session = aiohttp.ClientSession(timeout=_SLSKD_DEFAULT_TIMEOUT)
|
||
try:
|
||
headers = self._get_headers()
|
||
async with session.get(simple_url, headers=headers) as resp:
|
||
if resp.status in [200, 405]: # 405 means endpoint exists but wrong method
|
||
available_endpoints[f"direct_{endpoint}"] = f"Status: {resp.status}"
|
||
logger.info(f"[OK] Direct endpoint available: {simple_url} (Status: {resp.status})")
|
||
except Exception as _e:
|
||
logger.debug("direct endpoint probe %s: %s", endpoint, _e)
|
||
finally:
|
||
await session.close()
|
||
|
||
except Exception as e:
|
||
logger.debug(f"Endpoint {endpoint} failed: {e}")
|
||
|
||
return {
|
||
'swagger_available': False,
|
||
'available_endpoints': available_endpoints,
|
||
'base_url': self.base_url
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error exploring API endpoints: {e}")
|
||
return {'error': str(e)}
|
||
|
||
def is_configured(self) -> bool:
|
||
"""Check if slskd is configured (has base_url)"""
|
||
return self.base_url is not None
|
||
|
||
async def cancel_all_searches(self):
|
||
"""Cancel all active searches"""
|
||
if not self.active_searches:
|
||
return
|
||
|
||
logger.info(f"Cancelling {len(self.active_searches)} active searches...")
|
||
for search_id in list(self.active_searches.keys()):
|
||
try:
|
||
# Delete the search via API
|
||
await self._make_request('DELETE', f'searches/{search_id}')
|
||
logger.debug(f"Cancelled search {search_id}")
|
||
except Exception as e:
|
||
logger.warning(f"Could not cancel search {search_id}: {e}")
|
||
|
||
# Mark all searches as cancelled
|
||
self.active_searches.clear()
|
||
|
||
async def close(self):
|
||
# Cancel any active searches before closing
|
||
await self.cancel_all_searches()
|
||
|
||
def __del__(self):
|
||
# No persistent session to clean up
|
||
pass
|