diff --git a/core/download_plugins/registry.py b/core/download_plugins/registry.py index ad01c2e9..2ca5ebb8 100644 --- a/core/download_plugins/registry.py +++ b/core/download_plugins/registry.py @@ -40,6 +40,8 @@ from core.download_plugins.base import DownloadSourcePlugin # orchestrator did. from core.amazon_download_client import AmazonDownloadClient from core.deezer_download_client import DeezerDownloadClient +from core.download_plugins.torrent import TorrentDownloadPlugin +from core.download_plugins.usenet import UsenetDownloadPlugin from core.hifi_client import HiFiClient from core.lidarr_download_client import LidarrDownloadClient from core.qobuz_client import QobuzClient @@ -190,5 +192,7 @@ def build_default_registry() -> DownloadPluginRegistry: aliases=('deezer_dl',))) registry.register(PluginSpec(name='lidarr', factory=LidarrDownloadClient, display_name='Lidarr')) registry.register(PluginSpec(name='soundcloud',factory=SoundcloudClient, display_name='SoundCloud')) + registry.register(PluginSpec(name='torrent', factory=TorrentDownloadPlugin, display_name='Torrent (Prowlarr)')) + registry.register(PluginSpec(name='usenet', factory=UsenetDownloadPlugin, display_name='Usenet (Prowlarr)')) return registry diff --git a/core/download_plugins/torrent.py b/core/download_plugins/torrent.py new file mode 100644 index 00000000..a5487422 --- /dev/null +++ b/core/download_plugins/torrent.py @@ -0,0 +1,472 @@ +"""TorrentDownloadPlugin — composes Prowlarr search + torrent client +adapter + archive_pipeline into a uniform download source. + +Pipeline: +1. ``search(query)`` calls ``ProwlarrClient.search`` filtered to + ``protocol='torrent'`` results, projects releases into + ``TrackResult`` / ``AlbumResult`` shaped objects the existing + search UI already understands. Encodes the indexer's + ``downloadUrl`` (or magnet URI) into the filename so + ``download()`` can recover it. +2. ``download(username, filename, ...)`` decodes the URL, asks the + active torrent adapter (qBittorrent, Transmission, or Deluge per + user's settings) to add it, spawns a background thread that + polls the adapter for completion. +3. On completion the thread walks the adapter-reported save path + via ``archive_pipeline.collect_audio_after_extraction`` and + marks the download succeeded with the first audio file as the + primary ``file_path`` (matches Lidarr's single-track-pick + contract — picking which specific track to import happens in + post-processing, not here). + +Limitations: +- ``save_path`` is the torrent client's view of the disk. If + SoulSync runs on a different host than qBit / Trans / Deluge, + the post-processing pipeline can't see those files. The plugin + works fine for the all-on-one-box case (most users); remote + setups will need a future sync step (rclone / SMB / Docker + bind mount). +- Track-level metadata isn't available until after download. + Search results carry only the release title + indexer metadata; + individual track names are populated when the matching pipeline + walks the extracted audio files. +""" + +from __future__ import annotations + +import asyncio +import re +import threading +import time +import uuid +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from config.settings import config_manager +from core.archive_pipeline import collect_audio_after_extraction +from core.download_plugins.base import DownloadSourcePlugin +from core.download_plugins.types import AlbumResult, DownloadStatus, TrackResult +from core.prowlarr_client import ( + DEFAULT_MUSIC_CATEGORIES, + ProwlarrClient, + ProwlarrSearchResult, +) +from core.torrent_clients import get_active_adapter as get_active_torrent_adapter +from utils.async_helpers import run_async +from utils.logging_config import get_logger + +logger = get_logger("download_plugins.torrent") + + +# Separator used to encode the download URL inside the filename +# field. Same convention Lidarr / YouTube use for embedding their +# own opaque identifiers — ``||``. +_FILENAME_SEP = '||' + +# Adapter states that count as the download being on-disk and +# safe to walk. ``seeding`` and ``completed`` both mean the +# bits are there; the user can pause seeding manually if they +# don't want to keep sharing. +_COMPLETE_STATES = frozenset(['seeding', 'completed']) + +# Max seconds the poll thread keeps watching one download before +# giving up. 6 hours covers slow private trackers without leaking +# threads forever on dead torrents. +_POLL_TIMEOUT_SECONDS = 6 * 60 * 60 + +# Poll cadence — torrent state changes slowly; no point hammering +# the WebUI more than once a second. +_POLL_INTERVAL_SECONDS = 2.0 + + +class TorrentDownloadPlugin(DownloadSourcePlugin): + """Torrent download source backed by Prowlarr + an active + torrent client adapter.""" + + def __init__(self) -> None: + self._prowlarr = ProwlarrClient() + # Track every download we've kicked off. Keyed by our own + # uuid — NOT the adapter's hash — because the orchestrator + # owns the lifecycle and we need a stable id even before + # the adapter has assigned one. + self.active_downloads: Dict[str, Dict[str, Any]] = {} + self._lock = threading.Lock() + self.shutdown_check = None + + def set_shutdown_check(self, check_callable): + self.shutdown_check = check_callable + + def reload_settings(self) -> None: + self._prowlarr.reload_settings() + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + def is_configured(self) -> bool: + if not self._prowlarr.is_configured(): + return False + adapter = get_active_torrent_adapter() + return bool(adapter and adapter.is_configured()) + + async def check_connection(self) -> bool: + if not self._prowlarr.is_configured(): + return False + adapter = get_active_torrent_adapter() + if not adapter or not adapter.is_configured(): + return False + # Probe both sides. A torrent download is useless if either + # the indexer or the downloader is unreachable. + prowlarr_ok = await self._prowlarr.check_connection() + if not prowlarr_ok: + return False + return await adapter.check_connection() + + # ------------------------------------------------------------------ + # Search + # ------------------------------------------------------------------ + + async def search( + self, + query: str, + timeout: Optional[int] = None, + progress_callback=None, + ) -> Tuple[List[TrackResult], List[AlbumResult]]: + if not self._prowlarr.is_configured(): + return ([], []) + try: + indexer_ids = _parse_indexer_id_filter() + results = await self._prowlarr.search( + query, + categories=DEFAULT_MUSIC_CATEGORIES, + indexer_ids=indexer_ids, + ) + except Exception as e: + logger.error("Torrent plugin search failed: %s", e) + return ([], []) + return self._project_results(results) + + def _project_results( + self, results: List[ProwlarrSearchResult] + ) -> Tuple[List[TrackResult], List[AlbumResult]]: + """Turn Prowlarr releases into TrackResult / AlbumResult + shaped objects. One TrackResult + one AlbumResult per + release — Prowlarr search hits are at the release level, + not the track level, so we can't synthesise track listings + without downloading the actual torrent.""" + tracks: List[TrackResult] = [] + albums: List[AlbumResult] = [] + for result in results: + if result.protocol != 'torrent': + continue + download_url = result.magnet_uri or result.download_url + if not download_url: + continue + filename = f"{download_url}{_FILENAME_SEP}{result.title}" + quality = _guess_quality_from_title(result.title) + tr = TrackResult( + username='torrent', + filename=filename, + size=result.size, + bitrate=None, + duration=None, + quality=quality, + # Torrent results don't have per-uploader slot / queue + # data the way Soulseek does. Fill with neutral values + # so the quality_score doesn't punish them artificially. + free_upload_slots=max(1, result.seeders or 0), + upload_speed=0, + queue_length=0, + artist=None, + title=result.title, + album=None, + track_number=None, + _source_metadata={ + 'indexer': result.indexer_name, + 'indexer_id': result.indexer_id, + 'seeders': result.seeders, + 'leechers': result.leechers, + 'grabs': result.grabs, + 'protocol': 'torrent', + }, + ) + tracks.append(tr) + albums.append(AlbumResult( + username='torrent', + album_path=f"torrent/{result.guid}", + album_title=result.title, + artist=None, + track_count=1, # unknown until download finishes + total_size=result.size, + tracks=[tr], + dominant_quality=quality, + year=None, + )) + return tracks, albums + + # ------------------------------------------------------------------ + # Download + # ------------------------------------------------------------------ + + async def download( + self, + username: str, + filename: str, + file_size: int = 0, + ) -> Optional[str]: + if not self.is_configured(): + return None + download_url, display_name = _decode_filename(filename) + if not download_url: + logger.error("Torrent download missing URL in filename: %r", filename) + return None + + download_id = str(uuid.uuid4()) + with self._lock: + self.active_downloads[download_id] = { + 'id': download_id, + 'filename': filename, + 'username': 'torrent', + 'display_name': display_name, + 'state': 'Initializing', + 'progress': 0.0, + 'size': file_size, + 'transferred': 0, + 'speed': 0, + 'file_path': None, + 'torrent_hash': None, + 'error': None, + } + + thread = threading.Thread( + target=self._download_thread, + args=(download_id, download_url, display_name), + daemon=True, + name=f'torrent-dl-{download_id[:8]}', + ) + thread.start() + return download_id + + def _download_thread(self, download_id: str, download_url: str, display_name: str) -> None: + """Background worker: hand the URL to the active adapter, + poll until done, then walk the resulting directory.""" + adapter = get_active_torrent_adapter() + if adapter is None or not adapter.is_configured(): + self._mark_error(download_id, "No torrent client configured") + return + + try: + torrent_hash = run_async(adapter.add_torrent(download_url)) + except Exception as e: + self._mark_error(download_id, f"add_torrent failed: {e}") + return + if not torrent_hash: + self._mark_error(download_id, "Torrent client refused the URL") + return + + with self._lock: + row = self.active_downloads.get(download_id) + if row is not None: + row['torrent_hash'] = torrent_hash + row['state'] = 'InProgress, Downloading' + + deadline = time.monotonic() + _POLL_TIMEOUT_SECONDS + last_save_path: Optional[str] = None + while time.monotonic() < deadline: + if self.shutdown_check and self.shutdown_check(): + return + try: + status = run_async(adapter.get_status(torrent_hash)) + except Exception as e: + logger.warning("Torrent poll error for %s: %s", torrent_hash, e) + status = None + + if status is None: + # Adapter forgot about the torrent — probably user-removed. + self._mark_error(download_id, "Torrent disappeared from client") + return + + with self._lock: + row = self.active_downloads.get(download_id) + if row is not None: + row['progress'] = status.progress * 100.0 + row['transferred'] = status.downloaded + row['speed'] = status.download_speed + row['size'] = status.size or row.get('size', 0) + row['state'] = _adapter_state_to_display(status.state) + row['error'] = status.error + if status.save_path: + last_save_path = status.save_path + + if status.state in _COMPLETE_STATES: + self._finalize_download(download_id, last_save_path) + return + if status.state == 'error': + self._mark_error(download_id, status.error or "Torrent client reported error") + return + + time.sleep(_POLL_INTERVAL_SECONDS) + + self._mark_error(download_id, "Torrent download timed out") + + def _finalize_download(self, download_id: str, save_path: Optional[str]) -> None: + """Adapter said complete. Walk the directory + pick the + first audio file as the canonical ``file_path``.""" + if not save_path: + self._mark_error(download_id, "Torrent completed but no save_path reported") + return + try: + audio_files = collect_audio_after_extraction(Path(save_path)) + except Exception as e: + self._mark_error(download_id, f"Post-extract walk failed: {e}") + return + if not audio_files: + self._mark_error(download_id, f"No audio files found in {save_path}") + return + primary = audio_files[0] + with self._lock: + row = self.active_downloads.get(download_id) + if row is not None: + row['state'] = 'Completed, Succeeded' + row['progress'] = 100.0 + row['file_path'] = str(primary) + logger.info("Torrent download complete: %s -> %s (%d audio files)", + download_id[:8], primary.name, len(audio_files)) + + def _mark_error(self, download_id: str, message: str) -> None: + logger.error("Torrent download %s failed: %s", download_id[:8], message) + with self._lock: + row = self.active_downloads.get(download_id) + if row is not None: + row['state'] = 'Completed, Errored' + row['error'] = message + + # ------------------------------------------------------------------ + # Status / lifecycle + # ------------------------------------------------------------------ + + async def get_all_downloads(self) -> List[DownloadStatus]: + with self._lock: + rows = list(self.active_downloads.values()) + return [_row_to_status(r) for r in rows] + + async def get_download_status(self, download_id: str) -> Optional[DownloadStatus]: + with self._lock: + row = self.active_downloads.get(download_id) + if row is None: + return None + return _row_to_status(row) + + async def cancel_download( + self, + download_id: str, + username: Optional[str] = None, + remove: bool = False, + ) -> bool: + adapter = get_active_torrent_adapter() + with self._lock: + row = self.active_downloads.get(download_id) + torrent_hash = row.get('torrent_hash') if row else None + if adapter and torrent_hash: + try: + await adapter.remove(torrent_hash, delete_files=remove) + except Exception as e: + logger.warning("Torrent cancel via adapter failed: %s", e) + with self._lock: + if remove: + self.active_downloads.pop(download_id, None) + else: + row = self.active_downloads.get(download_id) + if row is not None: + row['state'] = 'Cancelled' + return True + + async def clear_all_completed_downloads(self) -> bool: + with self._lock: + for did in list(self.active_downloads.keys()): + state = self.active_downloads[did].get('state', '') + if state.startswith('Completed') or state == 'Cancelled': + self.active_downloads.pop(did, None) + return True + + +# --------------------------------------------------------------------------- +# Module-level helpers (pure functions — easy to unit-test) +# --------------------------------------------------------------------------- + + +def _decode_filename(filename: str) -> Tuple[Optional[str], str]: + """Pull the encoded download URL out of the ``filename`` string. + Returns ``(url, display_name)``. ``url`` is None when the string + has no separator.""" + if not filename or _FILENAME_SEP not in filename: + return (None, filename or '') + url, display = filename.split(_FILENAME_SEP, 1) + return (url, display) + + +def _guess_quality_from_title(title: str) -> str: + """Read the quality hint from a release title — most music + torrents put the encoding right in the name (FLAC, MP3 320, + etc.). Falls back to ``'mp3'`` so quality_score doesn't crash.""" + if not title: + return 'mp3' + lower = title.lower() + if 'flac' in lower: + return 'flac' + if re.search(r'\b24[\s-]?bit\b', lower) or 'hi-?res' in lower: + return 'flac' + if 'aac' in lower: + return 'aac' + if 'ogg' in lower: + return 'ogg' + return 'mp3' + + +def _parse_indexer_id_filter() -> List[int]: + """Read the comma-separated indexer-ID allowlist from config. + Empty list = search every enabled indexer.""" + raw = (config_manager.get('prowlarr.indexer_ids', '') or '').strip() + if not raw: + return [] + out: List[int] = [] + for chunk in raw.split(','): + chunk = chunk.strip() + if not chunk: + continue + try: + out.append(int(chunk)) + except ValueError: + continue + return out + + +def _adapter_state_to_display(state: str) -> str: + """Translate the adapter-uniform state strings into the + ``'InProgress, Downloading'`` / ``'Completed, Succeeded'`` + style the existing UI expects (matches Soulseek + Lidarr).""" + mapping = { + 'queued': 'Queued', + 'downloading': 'InProgress, Downloading', + 'stalled': 'InProgress, Stalled', + 'seeding': 'Completed, Succeeded', + 'completed': 'Completed, Succeeded', + 'paused': 'Paused', + 'error': 'Completed, Errored', + } + return mapping.get(state, state.title()) + + +def _row_to_status(row: Dict[str, Any]) -> DownloadStatus: + return DownloadStatus( + id=row['id'], + filename=row['filename'], + username=row['username'], + state=row.get('state', 'Unknown'), + progress=float(row.get('progress', 0.0)), + size=int(row.get('size', 0)), + transferred=int(row.get('transferred', 0)), + speed=int(row.get('speed', 0)), + time_remaining=None, + file_path=row.get('file_path'), + ) diff --git a/core/download_plugins/usenet.py b/core/download_plugins/usenet.py new file mode 100644 index 00000000..fca9b01d --- /dev/null +++ b/core/download_plugins/usenet.py @@ -0,0 +1,332 @@ +"""UsenetDownloadPlugin — composes Prowlarr search + usenet client +adapter + archive_pipeline into a uniform download source. + +Mirrors ``TorrentDownloadPlugin`` in shape and lifecycle (see that +module's docstring for the full pipeline rationale). Differences: + +- Search filters Prowlarr results to ``protocol='usenet'``. +- ``add_nzb`` replaces ``add_torrent``; for NZBs we usually have + a direct HTTP URL the indexer exposes via Prowlarr. +- Usenet clients (SABnzbd, NZBGet) typically auto-extract during + post-processing, so ``archive_pipeline.collect_audio_after_extraction`` + usually has nothing to extract and just walks loose files. +""" + +from __future__ import annotations + +import threading +import time +import uuid +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from core.archive_pipeline import collect_audio_after_extraction +from core.download_plugins.base import DownloadSourcePlugin +from core.download_plugins.torrent import ( + _adapter_state_to_display, + _decode_filename, + _guess_quality_from_title, + _parse_indexer_id_filter, + _row_to_status, + _COMPLETE_STATES, + _FILENAME_SEP, + _POLL_INTERVAL_SECONDS, + _POLL_TIMEOUT_SECONDS, +) +from core.download_plugins.types import AlbumResult, DownloadStatus, TrackResult +from core.prowlarr_client import ( + DEFAULT_MUSIC_CATEGORIES, + ProwlarrClient, + ProwlarrSearchResult, +) +from core.usenet_clients import get_active_adapter as get_active_usenet_adapter +from utils.async_helpers import run_async +from utils.logging_config import get_logger + +logger = get_logger("download_plugins.usenet") + + +class UsenetDownloadPlugin(DownloadSourcePlugin): + """Usenet download source backed by Prowlarr + an active usenet + client adapter (SABnzbd or NZBGet).""" + + def __init__(self) -> None: + self._prowlarr = ProwlarrClient() + self.active_downloads: Dict[str, Dict[str, Any]] = {} + self._lock = threading.Lock() + self.shutdown_check = None + + def set_shutdown_check(self, check_callable): + self.shutdown_check = check_callable + + def reload_settings(self) -> None: + self._prowlarr.reload_settings() + + def is_configured(self) -> bool: + if not self._prowlarr.is_configured(): + return False + adapter = get_active_usenet_adapter() + return bool(adapter and adapter.is_configured()) + + async def check_connection(self) -> bool: + if not self._prowlarr.is_configured(): + return False + adapter = get_active_usenet_adapter() + if not adapter or not adapter.is_configured(): + return False + if not await self._prowlarr.check_connection(): + return False + return await adapter.check_connection() + + # ------------------------------------------------------------------ + # Search + # ------------------------------------------------------------------ + + async def search( + self, + query: str, + timeout: Optional[int] = None, + progress_callback=None, + ) -> Tuple[List[TrackResult], List[AlbumResult]]: + if not self._prowlarr.is_configured(): + return ([], []) + try: + indexer_ids = _parse_indexer_id_filter() + results = await self._prowlarr.search( + query, + categories=DEFAULT_MUSIC_CATEGORIES, + indexer_ids=indexer_ids, + ) + except Exception as e: + logger.error("Usenet plugin search failed: %s", e) + return ([], []) + return self._project_results(results) + + def _project_results( + self, results: List[ProwlarrSearchResult] + ) -> Tuple[List[TrackResult], List[AlbumResult]]: + tracks: List[TrackResult] = [] + albums: List[AlbumResult] = [] + for result in results: + if result.protocol != 'usenet': + continue + if not result.download_url: + continue + filename = f"{result.download_url}{_FILENAME_SEP}{result.title}" + quality = _guess_quality_from_title(result.title) + tr = TrackResult( + username='usenet', + filename=filename, + size=result.size, + bitrate=None, + duration=None, + quality=quality, + # Usenet doesn't expose per-uploader concurrency the way + # Soulseek does; fill in neutral non-punishing values. + free_upload_slots=1, + upload_speed=0, + queue_length=0, + artist=None, + title=result.title, + album=None, + track_number=None, + _source_metadata={ + 'indexer': result.indexer_name, + 'indexer_id': result.indexer_id, + 'grabs': result.grabs, + 'protocol': 'usenet', + }, + ) + tracks.append(tr) + albums.append(AlbumResult( + username='usenet', + album_path=f"usenet/{result.guid}", + album_title=result.title, + artist=None, + track_count=1, + total_size=result.size, + tracks=[tr], + dominant_quality=quality, + year=None, + )) + return tracks, albums + + # ------------------------------------------------------------------ + # Download + # ------------------------------------------------------------------ + + async def download( + self, + username: str, + filename: str, + file_size: int = 0, + ) -> Optional[str]: + if not self.is_configured(): + return None + nzb_url, display_name = _decode_filename(filename) + if not nzb_url: + logger.error("Usenet download missing URL in filename: %r", filename) + return None + + download_id = str(uuid.uuid4()) + with self._lock: + self.active_downloads[download_id] = { + 'id': download_id, + 'filename': filename, + 'username': 'usenet', + 'display_name': display_name, + 'state': 'Initializing', + 'progress': 0.0, + 'size': file_size, + 'transferred': 0, + 'speed': 0, + 'file_path': None, + 'job_id': None, + 'error': None, + } + + thread = threading.Thread( + target=self._download_thread, + args=(download_id, nzb_url), + daemon=True, + name=f'usenet-dl-{download_id[:8]}', + ) + thread.start() + return download_id + + def _download_thread(self, download_id: str, nzb_url: str) -> None: + adapter = get_active_usenet_adapter() + if adapter is None or not adapter.is_configured(): + self._mark_error(download_id, "No usenet client configured") + return + + try: + job_id = run_async(adapter.add_nzb(nzb_url)) + except Exception as e: + self._mark_error(download_id, f"add_nzb failed: {e}") + return + if not job_id: + self._mark_error(download_id, "Usenet client refused the NZB") + return + + with self._lock: + row = self.active_downloads.get(download_id) + if row is not None: + row['job_id'] = job_id + row['state'] = 'InProgress, Downloading' + + deadline = time.monotonic() + _POLL_TIMEOUT_SECONDS + last_save_path: Optional[str] = None + while time.monotonic() < deadline: + if self.shutdown_check and self.shutdown_check(): + return + try: + status = run_async(adapter.get_status(job_id)) + except Exception as e: + logger.warning("Usenet poll error for %s: %s", job_id, e) + status = None + + if status is None: + self._mark_error(download_id, "Usenet job disappeared from client") + return + + with self._lock: + row = self.active_downloads.get(download_id) + if row is not None: + row['progress'] = status.progress * 100.0 + row['transferred'] = status.downloaded + row['speed'] = status.download_speed + row['size'] = status.size or row.get('size', 0) + row['state'] = _adapter_state_to_display(status.state) + row['error'] = status.error + if status.save_path: + last_save_path = status.save_path + + if status.state in _COMPLETE_STATES: + self._finalize_download(download_id, last_save_path) + return + if status.state == 'failed': + self._mark_error(download_id, status.error or "Usenet client reported failure") + return + + time.sleep(_POLL_INTERVAL_SECONDS) + + self._mark_error(download_id, "Usenet download timed out") + + def _finalize_download(self, download_id: str, save_path: Optional[str]) -> None: + if not save_path: + self._mark_error(download_id, "Usenet job completed but no save_path reported") + return + try: + audio_files = collect_audio_after_extraction(Path(save_path)) + except Exception as e: + self._mark_error(download_id, f"Post-extract walk failed: {e}") + return + if not audio_files: + self._mark_error(download_id, f"No audio files found in {save_path}") + return + primary = audio_files[0] + with self._lock: + row = self.active_downloads.get(download_id) + if row is not None: + row['state'] = 'Completed, Succeeded' + row['progress'] = 100.0 + row['file_path'] = str(primary) + logger.info("Usenet download complete: %s -> %s (%d audio files)", + download_id[:8], primary.name, len(audio_files)) + + def _mark_error(self, download_id: str, message: str) -> None: + logger.error("Usenet download %s failed: %s", download_id[:8], message) + with self._lock: + row = self.active_downloads.get(download_id) + if row is not None: + row['state'] = 'Completed, Errored' + row['error'] = message + + # ------------------------------------------------------------------ + # Status / lifecycle + # ------------------------------------------------------------------ + + async def get_all_downloads(self) -> List[DownloadStatus]: + with self._lock: + rows = list(self.active_downloads.values()) + return [_row_to_status(r) for r in rows] + + async def get_download_status(self, download_id: str) -> Optional[DownloadStatus]: + with self._lock: + row = self.active_downloads.get(download_id) + if row is None: + return None + return _row_to_status(row) + + async def cancel_download( + self, + download_id: str, + username: Optional[str] = None, + remove: bool = False, + ) -> bool: + adapter = get_active_usenet_adapter() + with self._lock: + row = self.active_downloads.get(download_id) + job_id = row.get('job_id') if row else None + if adapter and job_id: + try: + await adapter.remove(job_id, delete_files=remove) + except Exception as e: + logger.warning("Usenet cancel via adapter failed: %s", e) + with self._lock: + if remove: + self.active_downloads.pop(download_id, None) + else: + row = self.active_downloads.get(download_id) + if row is not None: + row['state'] = 'Cancelled' + return True + + async def clear_all_completed_downloads(self) -> bool: + with self._lock: + for did in list(self.active_downloads.keys()): + state = self.active_downloads[did].get('state', '') + if state.startswith('Completed') or state == 'Cancelled': + self.active_downloads.pop(did, None) + return True diff --git a/tests/test_torrent_usenet_plugins.py b/tests/test_torrent_usenet_plugins.py new file mode 100644 index 00000000..4a9818a2 --- /dev/null +++ b/tests/test_torrent_usenet_plugins.py @@ -0,0 +1,346 @@ +"""Tests for ``core/download_plugins/torrent.py`` and ``usenet.py``. + +Both plugins compose a Prowlarr client + an adapter + the archive +pipeline. The tests mock the Prowlarr client and the active adapter +factory so we can pin the projection logic, filename encoding / +decoding, finalize path, and the cancel / clear lifecycle without +touching the network or filesystem (beyond ``tmp_path``). +""" + +from __future__ import annotations + +import asyncio +from pathlib import Path +from typing import Optional +from unittest.mock import MagicMock, patch + +import pytest + +from core.download_plugins.torrent import ( + TorrentDownloadPlugin, + _adapter_state_to_display, + _decode_filename, + _FILENAME_SEP, + _guess_quality_from_title, +) +from core.download_plugins.usenet import UsenetDownloadPlugin +from core.prowlarr_client import ProwlarrSearchResult +from core.torrent_clients.base import TorrentStatus +from core.usenet_clients.base import UsenetStatus + + +def _run(coro): + return asyncio.new_event_loop().run_until_complete(coro) + + +# --------------------------------------------------------------------------- +# Pure helpers +# --------------------------------------------------------------------------- + + +def test_decode_filename_splits_on_separator() -> None: + url, display = _decode_filename(f"https://x/y.torrent{_FILENAME_SEP}Album Name") + assert url == 'https://x/y.torrent' + assert display == 'Album Name' + + +def test_decode_filename_without_separator_returns_none_url() -> None: + url, display = _decode_filename('just a name') + assert url is None + assert display == 'just a name' + + +def test_decode_filename_handles_magnet_with_embedded_separators() -> None: + """Magnet URIs contain ``=`` and ``&`` but no ``||`` — so a + magnet must round-trip cleanly through the encoder.""" + magnet = 'magnet:?xt=urn:btih:abc123&dn=Album+Name' + encoded = f"{magnet}{_FILENAME_SEP}Display" + url, display = _decode_filename(encoded) + assert url == magnet + assert display == 'Display' + + +def test_guess_quality_from_title() -> None: + assert _guess_quality_from_title('Album [FLAC]') == 'flac' + assert _guess_quality_from_title('Album 24-bit Hi-Res') == 'flac' + assert _guess_quality_from_title('Album [MP3 320]') == 'mp3' + assert _guess_quality_from_title('Album [AAC 256]') == 'aac' + assert _guess_quality_from_title('Album [OGG]') == 'ogg' + # Default fallback so quality_score doesn't crash on bare titles. + assert _guess_quality_from_title('Just A Title') == 'mp3' + assert _guess_quality_from_title('') == 'mp3' + + +def test_adapter_state_mapping_covers_complete_states() -> None: + assert _adapter_state_to_display('downloading') == 'InProgress, Downloading' + assert _adapter_state_to_display('seeding') == 'Completed, Succeeded' + assert _adapter_state_to_display('completed') == 'Completed, Succeeded' + assert _adapter_state_to_display('error') == 'Completed, Errored' + assert _adapter_state_to_display('stalled') == 'InProgress, Stalled' + # Unknown state falls through with title-casing rather than crashing. + assert _adapter_state_to_display('weird') == 'Weird' + + +# --------------------------------------------------------------------------- +# Torrent plugin — search projection +# --------------------------------------------------------------------------- + + +def _make_torrent_result(**overrides) -> ProwlarrSearchResult: + base = dict( + guid='guid-1', title='Some Album FLAC', indexer_id=3, + indexer_name='Indexer', protocol='torrent', + download_url='https://x/y.torrent', magnet_uri=None, + info_url=None, size=500_000_000, seeders=12, leechers=3, + grabs=100, publish_date='2026-01-01', categories=[3040], + raw={}, + ) + base.update(overrides) + return ProwlarrSearchResult(**base) + + +def test_torrent_project_results_drops_non_torrent_protocol() -> None: + plugin = TorrentDownloadPlugin() + results = [ + _make_torrent_result(), + _make_torrent_result(protocol='usenet', title='Usenet Album'), + ] + tracks, albums = plugin._project_results(results) + assert len(tracks) == 1 + assert tracks[0].title == 'Some Album FLAC' + assert len(albums) == 1 + + +def test_torrent_project_results_drops_releases_without_download_url() -> None: + plugin = TorrentDownloadPlugin() + results = [_make_torrent_result(download_url=None, magnet_uri=None)] + tracks, albums = plugin._project_results(results) + assert tracks == [] + assert albums == [] + + +def test_torrent_project_results_prefers_magnet_when_available() -> None: + plugin = TorrentDownloadPlugin() + magnet = 'magnet:?xt=urn:btih:abc' + results = [_make_torrent_result(magnet_uri=magnet, download_url='https://x/y.torrent')] + tracks, _ = plugin._project_results(results) + url, _ = _decode_filename(tracks[0].filename) + assert url == magnet + + +def test_torrent_project_results_encodes_url_and_title_in_filename() -> None: + plugin = TorrentDownloadPlugin() + tracks, _ = plugin._project_results([_make_torrent_result()]) + url, display = _decode_filename(tracks[0].filename) + assert url == 'https://x/y.torrent' + assert display == 'Some Album FLAC' + + +def test_torrent_project_results_neutralizes_soulseek_specific_fields() -> None: + """TrackResult.quality_score punishes results with no upload + slots; torrent results don't have that concept so the + projection has to fill in non-punishing neutral values.""" + plugin = TorrentDownloadPlugin() + tracks, _ = plugin._project_results([_make_torrent_result(seeders=0)]) + # seeders=0 means we should still hand the picker something + # usable. free_upload_slots floors at 1 to avoid the 0-slot + # penalty applied to dead Soulseek peers. + assert tracks[0].free_upload_slots >= 1 + + +# --------------------------------------------------------------------------- +# Torrent plugin — is_configured / check_connection +# --------------------------------------------------------------------------- + + +def test_torrent_is_configured_requires_both_sides() -> None: + plugin = TorrentDownloadPlugin() + with patch.object(plugin._prowlarr, 'is_configured', return_value=False), \ + patch('core.download_plugins.torrent.get_active_torrent_adapter', return_value=None): + assert plugin.is_configured() is False + fake_adapter = MagicMock() + fake_adapter.is_configured.return_value = False + with patch.object(plugin._prowlarr, 'is_configured', return_value=True), \ + patch('core.download_plugins.torrent.get_active_torrent_adapter', return_value=fake_adapter): + assert plugin.is_configured() is False + fake_adapter.is_configured.return_value = True + with patch.object(plugin._prowlarr, 'is_configured', return_value=True), \ + patch('core.download_plugins.torrent.get_active_torrent_adapter', return_value=fake_adapter): + assert plugin.is_configured() is True + + +# --------------------------------------------------------------------------- +# Torrent plugin — finalize / cancel / clear +# --------------------------------------------------------------------------- + + +def test_torrent_finalize_picks_first_audio_file(tmp_path: Path) -> None: + plugin = TorrentDownloadPlugin() + # Seed an in-flight download row + plugin.active_downloads['dl-1'] = { + 'id': 'dl-1', 'filename': 'x', 'username': 'torrent', + 'display_name': 'X', 'state': 'InProgress, Downloading', + 'progress': 50.0, 'size': 0, 'transferred': 0, 'speed': 0, + 'file_path': None, 'torrent_hash': 'h1', 'error': None, + } + # Drop two audio files in the save dir + (tmp_path / 'b.flac').write_bytes(b'fLaC') + (tmp_path / 'a.mp3').write_bytes(b'ID3') + plugin._finalize_download('dl-1', str(tmp_path)) + row = plugin.active_downloads['dl-1'] + assert row['state'] == 'Completed, Succeeded' + assert row['progress'] == 100.0 + # Walker sorts → 'a.mp3' wins as first. + assert row['file_path'].endswith('a.mp3') + + +def test_torrent_finalize_marks_error_when_no_audio(tmp_path: Path) -> None: + plugin = TorrentDownloadPlugin() + plugin.active_downloads['dl-1'] = { + 'id': 'dl-1', 'filename': 'x', 'username': 'torrent', + 'display_name': 'X', 'state': 'InProgress, Downloading', + 'progress': 50.0, 'size': 0, 'transferred': 0, 'speed': 0, + 'file_path': None, 'torrent_hash': 'h1', 'error': None, + } + # tmp_path has no audio files + plugin._finalize_download('dl-1', str(tmp_path)) + assert plugin.active_downloads['dl-1']['state'] == 'Completed, Errored' + assert 'No audio files' in plugin.active_downloads['dl-1']['error'] + + +def test_torrent_finalize_marks_error_when_save_path_missing() -> None: + plugin = TorrentDownloadPlugin() + plugin.active_downloads['dl-1'] = { + 'id': 'dl-1', 'filename': 'x', 'username': 'torrent', + 'display_name': 'X', 'state': 'InProgress, Downloading', + 'progress': 50.0, 'size': 0, 'transferred': 0, 'speed': 0, + 'file_path': None, 'torrent_hash': 'h1', 'error': None, + } + plugin._finalize_download('dl-1', None) + assert plugin.active_downloads['dl-1']['state'] == 'Completed, Errored' + assert 'no save_path' in plugin.active_downloads['dl-1']['error'].lower() + + +def test_torrent_clear_completed_drops_only_done_rows() -> None: + plugin = TorrentDownloadPlugin() + plugin.active_downloads['a'] = {'id': 'a', 'state': 'InProgress, Downloading'} + plugin.active_downloads['b'] = {'id': 'b', 'state': 'Completed, Succeeded'} + plugin.active_downloads['c'] = {'id': 'c', 'state': 'Completed, Errored'} + plugin.active_downloads['d'] = {'id': 'd', 'state': 'Cancelled'} + _run(plugin.clear_all_completed_downloads()) + assert list(plugin.active_downloads.keys()) == ['a'] + + +def test_torrent_get_all_returns_status_objects() -> None: + plugin = TorrentDownloadPlugin() + plugin.active_downloads['a'] = { + 'id': 'a', 'filename': 'f', 'username': 'torrent', + 'state': 'InProgress, Downloading', 'progress': 50.0, + 'size': 100, 'transferred': 50, 'speed': 1000, + 'file_path': None, + } + statuses = _run(plugin.get_all_downloads()) + assert len(statuses) == 1 + assert statuses[0].id == 'a' + assert statuses[0].progress == 50.0 + + +# --------------------------------------------------------------------------- +# Usenet plugin — projection +# --------------------------------------------------------------------------- + + +def _make_usenet_result(**overrides) -> ProwlarrSearchResult: + base = dict( + guid='guid-u', title='Some Album', indexer_id=5, + indexer_name='UsenetIndexer', protocol='usenet', + download_url='https://x/y.nzb', magnet_uri=None, + info_url=None, size=400_000_000, seeders=None, leechers=None, + grabs=42, publish_date='2026-01-01', categories=[3010], + raw={}, + ) + base.update(overrides) + return ProwlarrSearchResult(**base) + + +def test_usenet_project_drops_torrent_protocol() -> None: + plugin = UsenetDownloadPlugin() + results = [_make_usenet_result(), _make_usenet_result(protocol='torrent', title='T')] + tracks, albums = plugin._project_results(results) + assert len(tracks) == 1 + assert tracks[0].username == 'usenet' + + +def test_usenet_project_drops_results_without_download_url() -> None: + """Usenet plugins reject magnet-only results entirely — NZBs + don't have a magnet equivalent.""" + plugin = UsenetDownloadPlugin() + results = [_make_usenet_result(download_url=None)] + tracks, _ = plugin._project_results(results) + assert tracks == [] + + +def test_usenet_project_encodes_url_in_filename() -> None: + plugin = UsenetDownloadPlugin() + tracks, _ = plugin._project_results([_make_usenet_result()]) + url, display = _decode_filename(tracks[0].filename) + assert url == 'https://x/y.nzb' + assert display == 'Some Album' + + +def test_usenet_finalize_picks_first_audio_file(tmp_path: Path) -> None: + """Same finalize contract as torrent — sanity check the shared + helper path works for usenet too.""" + plugin = UsenetDownloadPlugin() + plugin.active_downloads['u-1'] = { + 'id': 'u-1', 'filename': 'x', 'username': 'usenet', + 'display_name': 'X', 'state': 'InProgress, Downloading', + 'progress': 50.0, 'size': 0, 'transferred': 0, 'speed': 0, + 'file_path': None, 'job_id': 'j1', 'error': None, + } + (tmp_path / 'track1.flac').write_bytes(b'fLaC') + plugin._finalize_download('u-1', str(tmp_path)) + assert plugin.active_downloads['u-1']['state'] == 'Completed, Succeeded' + assert plugin.active_downloads['u-1']['file_path'].endswith('track1.flac') + + +def test_usenet_is_configured_requires_both_sides() -> None: + plugin = UsenetDownloadPlugin() + fake_adapter = MagicMock() + fake_adapter.is_configured.return_value = True + with patch.object(plugin._prowlarr, 'is_configured', return_value=False), \ + patch('core.download_plugins.usenet.get_active_usenet_adapter', return_value=fake_adapter): + assert plugin.is_configured() is False + with patch.object(plugin._prowlarr, 'is_configured', return_value=True), \ + patch('core.download_plugins.usenet.get_active_usenet_adapter', return_value=None): + assert plugin.is_configured() is False + with patch.object(plugin._prowlarr, 'is_configured', return_value=True), \ + patch('core.download_plugins.usenet.get_active_usenet_adapter', return_value=fake_adapter): + assert plugin.is_configured() is True + + +# --------------------------------------------------------------------------- +# Plugin conformance — both must satisfy the DownloadSourcePlugin Protocol +# --------------------------------------------------------------------------- + + +def test_plugins_conform_to_protocol() -> None: + from core.download_plugins.base import DownloadSourcePlugin + assert isinstance(TorrentDownloadPlugin(), DownloadSourcePlugin) + assert isinstance(UsenetDownloadPlugin(), DownloadSourcePlugin) + + +# --------------------------------------------------------------------------- +# Registry — both should register cleanly +# --------------------------------------------------------------------------- + + +def test_registry_includes_torrent_and_usenet() -> None: + """The registry decides what shows up in the orchestrator's + iteration helpers. If we forget to register a new plugin the + download source dropdown will silently no-op.""" + from core.download_plugins.registry import build_default_registry + registry = build_default_registry() + names = registry.names() + assert 'torrent' in names + assert 'usenet' in names diff --git a/webui/index.html b/webui/index.html index f6b0cb1b..35147361 100644 --- a/webui/index.html +++ b/webui/index.html @@ -4328,6 +4328,8 @@ + +
@@ -4735,6 +4737,15 @@
+ + +