diff --git a/.github/workflows/docker-publish.yml b/.github/workflows/docker-publish.yml index d11761e1..f049a16e 100644 --- a/.github/workflows/docker-publish.yml +++ b/.github/workflows/docker-publish.yml @@ -9,9 +9,9 @@ on: workflow_dispatch: inputs: version_tag: - description: 'Version tag (e.g. 2.5.7)' + description: 'Version tag (e.g. 2.5.8)' required: true - default: '2.5.7' + default: '2.5.8' jobs: build-and-push: diff --git a/.gitignore b/.gitignore index 8266b83b..8857f8d5 100644 --- a/.gitignore +++ b/.gitignore @@ -17,6 +17,7 @@ database/music_library.db-shm database/music_library.db-wal database/music_library.db.backup_* database/api_call_history.json +storage/image_cache/ logs/*.log logs/*.log.* diff --git a/README.md b/README.md index 579dc488..a79eeb7e 100644 --- a/README.md +++ b/README.md @@ -279,13 +279,34 @@ The template points at `boulderbadgedad/soulsync:latest` (stable) by default. To git clone https://github.com/Nezreka/SoulSync cd SoulSync python -m pip install -r requirements.txt + +# Build the React WebUI bundle used by the Python server. +# Docker does this automatically; Python installs must do it manually. +cd webui +npm ci +npm run build +cd .. + gunicorn -c gunicorn.conf.py wsgi:application # Open http://localhost:8008 ``` +When updating a Python/no-Docker install with `git pull`, rebuild the WebUI before restarting SoulSync: + +```bash +cd webui +npm ci +npm run build +cd .. +``` + +If `webui/static/dist/.vite/manifest.json` is missing or stale, React-owned routes and route handoffs may not load correctly. + ### Local Development -Use two terminals so the backend and Vite stay independent: +This is only for contributors working on the WebUI with hot reload. Normal Python/no-Docker installs should build once with `npm run build` as shown above, then run only Gunicorn. + +For active frontend development, use two terminals so the backend and Vite stay independent: 1. Backend ```bash diff --git a/config/settings.py b/config/settings.py index 2cfc3eeb..aa989f41 100644 --- a/config/settings.py +++ b/config/settings.py @@ -87,6 +87,10 @@ class ConfigManager: 'soulseek.api_key', 'deezer_download.arl', 'lidarr_download.api_key', + 'prowlarr.api_key', + 'torrent_client.password', + 'usenet_client.api_key', + 'usenet_client.password', # Enrichment services 'listenbrainz.token', 'acoustid.api_key', @@ -519,6 +523,37 @@ class ConfigManager: "quality_profile": "Any", "cleanup_after_import": True, }, + # Prowlarr — indexer aggregator. Feeds the torrent / usenet + # download plugins. Not a standalone source. + "prowlarr": { + "url": "", + "api_key": "", + # Comma-separated list of indexer IDs to limit searches to. + # Empty = search all enabled indexers. + "indexer_ids": "", + }, + # Torrent client — receives .torrent / magnet URIs from the + # torrent download plugin. ``type`` picks which adapter to + # instantiate (qbittorrent | transmission | deluge). + "torrent_client": { + "type": "qbittorrent", + "url": "", + "username": "", + "password": "", + "category": "soulsync", + "save_path": "", + }, + # Usenet client — receives .nzb URLs / payloads. ``type`` + # picks the adapter (sabnzbd | nzbget). SABnzbd uses an + # API key; NZBGet uses username + password. + "usenet_client": { + "type": "sabnzbd", + "url": "", + "api_key": "", + "username": "", + "password": "", + "category": "soulsync", + }, "soundcloud_download": { # Anonymous-only for now — SoundCloud Go+ OAuth tier could be # added later, with credentials living under a "session" subkey @@ -552,6 +587,13 @@ class ConfigManager: "path": os.environ.get('DATABASE_PATH', 'database/music_library.db'), "max_workers": 5 }, + "image_cache": { + "enabled": True, + "path": "storage/image_cache", + "ttl_seconds": 2592000, + "failed_ttl_seconds": 21600, + "max_download_mb": 15 + }, "metadata_enhancement": { "enabled": True, "embed_album_art": True, diff --git a/core/connection_test.py b/core/connection_test.py index 2019b4fe..c470cfec 100644 --- a/core/connection_test.py +++ b/core/connection_test.py @@ -305,6 +305,61 @@ def run_service_test(service, test_config): return False, "Invalid Genius access token." except Exception as e: return False, f"Genius connection error: {str(e)}" + elif service == "usenet_client": + client_type = (config_manager.get('usenet_client.type', '') or '').strip().lower() + url = config_manager.get('usenet_client.url', '') + if not url: + return False, "Usenet client URL is required." + if not client_type: + return False, "Pick a usenet client (SABnzbd or NZBGet)." + try: + from core.usenet_clients import adapter_for_type as _usenet_adapter_for_type + adapter = _usenet_adapter_for_type(client_type) + if adapter is None: + return False, f"Unknown usenet client type: {client_type}" + if not adapter.is_configured(): + if client_type == "sabnzbd": + return False, "SABnzbd needs both URL and API key." + return False, "NZBGet needs URL, username, and password." + if run_async(adapter.check_connection()): + return True, f"Connected to {client_type}" + return False, f"{client_type} probe failed — check URL, credentials, and that the client is running." + except Exception as e: + return False, f"Usenet client connection error: {str(e)}" + elif service == "torrent_client": + client_type = (config_manager.get('torrent_client.type', '') or '').strip().lower() + url = config_manager.get('torrent_client.url', '') + if not url: + return False, "Torrent client URL is required." + if not client_type: + return False, "Pick a torrent client (qBittorrent, Transmission, or Deluge)." + try: + from core.torrent_clients import adapter_for_type + adapter = adapter_for_type(client_type) + if adapter is None: + return False, f"Unknown torrent client type: {client_type}" + if not adapter.is_configured(): + return False, "Torrent client missing required credentials." + if run_async(adapter.check_connection()): + return True, f"Connected to {client_type}" + return False, f"{client_type} probe failed — check URL, credentials, and that the client is running." + except Exception as e: + return False, f"Torrent client connection error: {str(e)}" + elif service == "prowlarr": + url = config_manager.get('prowlarr.url', '') + api_key = config_manager.get('prowlarr.api_key', '') + if not url or not api_key: + return False, "Prowlarr URL and API key are required." + try: + import requests as _req + resp = _req.get(f"{url.rstrip('/')}/api/v1/system/status", + headers={'X-Api-Key': api_key}, timeout=10) + if resp.ok: + version = resp.json().get('version', '?') + return True, f"Connected to Prowlarr v{version}" + return False, f"Prowlarr returned HTTP {resp.status_code}" + except Exception as e: + return False, f"Prowlarr connection error: {str(e)}" elif service == "lidarr" or service == "lidarr_download": url = config_manager.get('lidarr_download.url', '') api_key = config_manager.get('lidarr_download.api_key', '') diff --git a/core/downloads/lifecycle.py b/core/downloads/lifecycle.py index 43e1e970..2a29feec 100644 --- a/core/downloads/lifecycle.py +++ b/core/downloads/lifecycle.py @@ -173,7 +173,7 @@ def on_download_completed(batch_id: str, task_id: str, success: bool, deps: Life # Guard against double-calling: track which tasks have already been completed # This prevents active_count from being decremented multiple times for the same task - # (e.g. monitor detects completion AND post-processing calls this again) + # (e.g. status polling and post-processing both observe the same terminal task) # NOTE: On duplicate calls, we skip decrement/tracking but STILL check batch completion. # This is critical because the first call may see the task in 'post_processing' (not finished), # and the second call (from post-processing worker) arrives after the task is truly 'completed'. diff --git a/core/downloads/monitor.py b/core/downloads/monitor.py index 79a90d8d..62d515e5 100644 --- a/core/downloads/monitor.py +++ b/core/downloads/monitor.py @@ -227,18 +227,28 @@ class WebUIDownloadMonitor: except Exception as e: logger.error(f"[Deferred] Error executing deferred operation {op[0]}: {e}") - # Handle completed downloads outside the lock to prevent deadlock - # (_on_download_completed acquires tasks_lock internally) + # Handle completed transfers outside the lock. The transfer engine's + # "complete" state only means the remote download finished; the + # post-processing worker still has to find, verify, tag, and move the + # file before it can report real batch success or failure. for batch_id, task_id in completed_tasks: try: # Submit post-processing worker (file move, tagging, AcoustID verification) # This makes batch downloads fully independent of browser polling. logger.info(f"[Monitor] Submitting post-processing worker for task {task_id}") missing_download_executor.submit(_run_post_processing_worker, task_id, batch_id) - # Chain to next download in the batch queue - _on_download_completed(batch_id, task_id, success=True) except Exception as e: logger.error(f"[Monitor] Error handling completed task {task_id}: {e}") + with tasks_lock: + if task_id in download_tasks: + download_tasks[task_id]['status'] = 'failed' + download_tasks[task_id]['error_message'] = f'Post-processing could not be scheduled: {e}' + try: + _on_download_completed(batch_id, task_id, success=False) + except Exception as completion_error: + logger.error( + f"[Monitor] Error marking failed post-processing submit for task {task_id}: {completion_error}" + ) # Handle exhausted retry tasks outside the lock to prevent deadlock for batch_id, task_id in exhausted_tasks: try: diff --git a/core/downloads/validation.py b/core/downloads/validation.py index 79131472..3ac38751 100644 --- a/core/downloads/validation.py +++ b/core/downloads/validation.py @@ -9,6 +9,7 @@ import logging import re from config.settings import config_manager +from core.imports.file_integrity import resolve_duration_tolerance logger = logging.getLogger(__name__) @@ -61,6 +62,24 @@ def filter_soundcloud_previews(results, expected_track): return [r for r in results if not _is_preview(r)] +def _duration_tolerance_seconds(expected_duration_ms): + override = resolve_duration_tolerance( + config_manager.get('post_processing.duration_tolerance_seconds', 0) + ) + if override is not None: + return override + expected_seconds = expected_duration_ms / 1000.0 + return 5.0 if expected_seconds > 600.0 else 3.0 + + +def _duration_mismatch_exceeds_integrity_tolerance(expected_duration_ms, candidate_duration_ms): + if not expected_duration_ms or not candidate_duration_ms: + return False + tolerance = _duration_tolerance_seconds(expected_duration_ms) + drift = abs((candidate_duration_ms / 1000.0) - (expected_duration_ms / 1000.0)) + return drift > tolerance + + def get_valid_candidates(results, spotify_track, query): """ This function is a direct port from sync.py. It scores and filters @@ -98,7 +117,21 @@ def get_valid_candidates(results, spotify_track, query): expected_is_version = any(kw in expected_title_lower for kw in _version_keywords) scored = [] + _strict_duration_sources = {'tidal', 'qobuz', 'hifi', 'deezer_dl', 'amazon'} for r in results: + if ( + r.username in _strict_duration_sources + and _duration_mismatch_exceeds_integrity_tolerance(expected_duration, r.duration or 0) + ): + logger.info( + "[%s] Rejecting candidate due to duration mismatch before download: " + "expected %.1fs, candidate %.1fs", + source_label, + expected_duration / 1000.0, + (r.duration or 0) / 1000.0, + ) + continue + # Score using matching engine's generic scorer (same weights as Soulseek) confidence, match_type = matching_engine.score_track_match( source_title=expected_title, diff --git a/core/image_cache.py b/core/image_cache.py new file mode 100644 index 00000000..040c1126 --- /dev/null +++ b/core/image_cache.py @@ -0,0 +1,342 @@ +"""Disk-backed image cache for browser-facing artwork URLs.""" + +from __future__ import annotations + +import hashlib +import mimetypes +import os +import sqlite3 +import threading +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Callable, Optional +from urllib.parse import urlparse + +import requests + +from config.settings import config_manager +from core.metadata.artwork import is_internal_image_host +from utils.logging_config import get_logger + +logger = get_logger("image_cache") + +DEFAULT_TTL_SECONDS = 30 * 24 * 60 * 60 +DEFAULT_FAILED_TTL_SECONDS = 6 * 60 * 60 +DEFAULT_MAX_DOWNLOAD_BYTES = 15 * 1024 * 1024 + + +class ImageCacheError(Exception): + """Raised when an image cannot be served from the cache.""" + + +@dataclass +class CachedImage: + key: str + path: Path + mime_type: str + size: int + status: str + + +class ImageCache: + def __init__( + self, + cache_dir: str | os.PathLike[str], + *, + ttl_seconds: int = DEFAULT_TTL_SECONDS, + failed_ttl_seconds: int = DEFAULT_FAILED_TTL_SECONDS, + max_download_bytes: int = DEFAULT_MAX_DOWNLOAD_BYTES, + fetcher: Optional[Callable[..., requests.Response]] = None, + ): + self.cache_dir = Path(cache_dir) + self.ttl_seconds = int(ttl_seconds) + self.failed_ttl_seconds = int(failed_ttl_seconds) + self.max_download_bytes = int(max_download_bytes) + self.fetcher = fetcher or requests.get + self.db_path = self.cache_dir / "image_cache.sqlite3" + self._db_lock = threading.RLock() + self._key_locks: dict[str, threading.Lock] = {} + self._key_locks_lock = threading.Lock() + self.cache_dir.mkdir(parents=True, exist_ok=True) + self._init_db() + + def cache_url_for(self, url: str | None) -> str | None: + """Register a URL and return its browser-facing cached path.""" + if not url: + return None + if str(url).startswith("/api/image-cache/"): + return str(url) + if not self.is_cacheable_url(str(url)): + return str(url) + + key = self.key_for_url(str(url)) + now = time.time() + with self._db_lock: + with self._connect() as conn: + conn.execute( + """ + INSERT INTO image_cache + (key, original_url, status, created_at, updated_at, last_accessed, + expires_at, size, mime_type, file_path, last_error) + VALUES (?, ?, 'pending', ?, ?, ?, 0, 0, '', '', '') + ON CONFLICT(key) DO UPDATE SET + original_url=excluded.original_url, + last_accessed=excluded.last_accessed + """, + (key, str(url), now, now, now), + ) + return f"/api/image-cache/{key}" + + def get(self, key: str) -> CachedImage: + row = self._get_row(key) + if not row: + raise ImageCacheError("Image cache key not found") + return self.get_url(row["original_url"]) + + def get_url(self, url: str) -> CachedImage: + if not self.is_cacheable_url(url): + raise ImageCacheError("URL is not cacheable") + + key = self.key_for_url(url) + lock = self._lock_for_key(key) + with lock: + row = self._get_row(key) + now = time.time() + if row and row["status"] == "ok" and row["file_path"]: + path = Path(row["file_path"]) + if path.exists(): + self._touch(key, now) + if float(row["expires_at"] or 0) > now: + return CachedImage(key, path, row["mime_type"] or "image/jpeg", int(row["size"] or 0), "hit") + + try: + return self._fetch_and_store(url, key, now) + except Exception as exc: + if row and row["status"] == "ok" and row["file_path"]: + stale_path = Path(row["file_path"]) + if stale_path.exists(): + logger.warning("Serving stale cached image for %s after refresh failed: %s", key, exc) + self._record_error(key, str(exc), now, keep_status=True) + return CachedImage( + key, + stale_path, + row["mime_type"] or "image/jpeg", + int(row["size"] or 0), + "stale", + ) + self._record_error(key, str(exc), now) + raise ImageCacheError(str(exc)) from exc + + @staticmethod + def key_for_url(url: str) -> str: + return hashlib.sha256(url.encode("utf-8")).hexdigest() + + @staticmethod + def is_cacheable_url(url: str) -> bool: + try: + parsed = urlparse(url) + if parsed.scheme not in {"http", "https"}: + return False + if parsed.username or parsed.password: + return False + if not parsed.hostname: + return False + return True + except Exception: + return False + + def _fetch_and_store(self, url: str, key: str, now: float) -> CachedImage: + if not self._is_fetch_allowed(url): + raise ImageCacheError("Image host is not allowed") + + response = self.fetcher( + url, + timeout=10, + stream=True, + headers={ + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" + ), + "Accept": "image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8", + "Referer": "https://www.deezer.com/", + }, + ) + try: + if response.status_code != 200: + raise ImageCacheError(f"Upstream image returned HTTP {response.status_code}") + + mime_type = (response.headers.get("Content-Type") or "image/jpeg").split(";", 1)[0].strip() + if not mime_type.startswith("image/"): + raise ImageCacheError(f"Upstream response is not an image: {mime_type}") + + declared_size = response.headers.get("Content-Length") + try: + if declared_size and int(declared_size) > self.max_download_bytes: + raise ImageCacheError("Image exceeds configured size limit") + except ValueError: + pass + + ext = mimetypes.guess_extension(mime_type) or ".img" + if ext == ".jpe": + ext = ".jpg" + path = self._path_for_key(key, ext) + path.parent.mkdir(parents=True, exist_ok=True) + tmp_path = path.with_suffix(path.suffix + ".tmp") + + total = 0 + try: + with open(tmp_path, "wb") as handle: + for chunk in response.iter_content(chunk_size=64 * 1024): + if not chunk: + continue + total += len(chunk) + if total > self.max_download_bytes: + raise ImageCacheError("Image exceeds configured size limit") + handle.write(chunk) + except Exception: + try: + tmp_path.unlink(missing_ok=True) + except Exception as cleanup_exc: + logger.debug("image_cache tmp cleanup failed: %s", cleanup_exc) + raise + + if total <= 0: + raise ImageCacheError("Image response was empty") + + os.replace(tmp_path, path) + expires_at = now + self.ttl_seconds + with self._db_lock: + with self._connect() as conn: + conn.execute( + """ + INSERT INTO image_cache + (key, original_url, status, created_at, updated_at, last_accessed, + expires_at, size, mime_type, file_path, last_error) + VALUES (?, ?, 'ok', ?, ?, ?, ?, ?, ?, ?, '') + ON CONFLICT(key) DO UPDATE SET + original_url=excluded.original_url, + status='ok', + updated_at=excluded.updated_at, + last_accessed=excluded.last_accessed, + expires_at=excluded.expires_at, + size=excluded.size, + mime_type=excluded.mime_type, + file_path=excluded.file_path, + last_error='' + """, + (key, url, now, now, now, expires_at, total, mime_type, str(path)), + ) + return CachedImage(key, path, mime_type, total, "miss") + finally: + response.close() + + def _path_for_key(self, key: str, extension: str) -> Path: + return self.cache_dir / key[:2] / key[2:4] / f"{key}{extension}" + + def _is_fetch_allowed(self, url: str) -> bool: + parsed = urlparse(url) + if parsed.scheme not in {"http", "https"}: + return False + if parsed.username or parsed.password: + return False + if not parsed.hostname: + return False + + # Internal hosts are explicitly supported because Plex/Jellyfin/Navidrome + # artwork often lives behind Docker/LAN-only URLs. Public hosts are allowed + # as image-only responses with size limits. + return bool(parsed.hostname) or is_internal_image_host(url) + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(self.db_path) + conn.row_factory = sqlite3.Row + return conn + + def _init_db(self) -> None: + with self._db_lock: + with self._connect() as conn: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS image_cache ( + key TEXT PRIMARY KEY, + original_url TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + last_accessed REAL NOT NULL, + expires_at REAL NOT NULL DEFAULT 0, + size INTEGER NOT NULL DEFAULT 0, + mime_type TEXT NOT NULL DEFAULT '', + file_path TEXT NOT NULL DEFAULT '', + last_error TEXT NOT NULL DEFAULT '' + ) + """ + ) + conn.execute("CREATE INDEX IF NOT EXISTS idx_image_cache_accessed ON image_cache(last_accessed)") + + def _get_row(self, key: str) -> Optional[sqlite3.Row]: + with self._db_lock: + with self._connect() as conn: + return conn.execute("SELECT * FROM image_cache WHERE key = ?", (key,)).fetchone() + + def _touch(self, key: str, now: float) -> None: + with self._db_lock: + with self._connect() as conn: + conn.execute("UPDATE image_cache SET last_accessed = ? WHERE key = ?", (now, key)) + + def _record_error(self, key: str, error: str, now: float, *, keep_status: bool = False) -> None: + status_sql = "status" if keep_status else "'failed'" + with self._db_lock: + with self._connect() as conn: + conn.execute( + f""" + UPDATE image_cache + SET status = {status_sql}, + updated_at = ?, + last_accessed = ?, + expires_at = ?, + last_error = ? + WHERE key = ? + """, + (now, now, now + self.failed_ttl_seconds, error[:500], key), + ) + + def _lock_for_key(self, key: str) -> threading.Lock: + with self._key_locks_lock: + lock = self._key_locks.get(key) + if lock is None: + lock = threading.Lock() + self._key_locks[key] = lock + return lock + + +_image_cache: Optional[ImageCache] = None +_image_cache_lock = threading.Lock() + + +def get_image_cache() -> ImageCache: + global _image_cache + with _image_cache_lock: + if _image_cache is None: + cache_dir = config_manager.get("image_cache.path", "storage/image_cache") + if not os.path.isabs(cache_dir): + cache_dir = str(config_manager.base_dir / cache_dir) + _image_cache = ImageCache( + cache_dir, + ttl_seconds=int(config_manager.get("image_cache.ttl_seconds", DEFAULT_TTL_SECONDS)), + failed_ttl_seconds=int(config_manager.get("image_cache.failed_ttl_seconds", DEFAULT_FAILED_TTL_SECONDS)), + max_download_bytes=int(config_manager.get("image_cache.max_download_mb", 15)) * 1024 * 1024, + ) + return _image_cache + + +def cached_image_url(url: str | None) -> str | None: + if not url or config_manager.get("image_cache.enabled", True) is False: + return url + try: + return get_image_cache().cache_url_for(url) + except Exception as exc: + logger.debug("image cache registration failed: %s", exc) + return url diff --git a/core/metadata/artwork.py b/core/metadata/artwork.py index 98af8431..694933d9 100644 --- a/core/metadata/artwork.py +++ b/core/metadata/artwork.py @@ -192,13 +192,13 @@ def normalize_image_url(thumb_url: str | None) -> str | None: def is_image_proxy_url(url: str) -> bool: - """Return True for SoulSync image-proxy URLs, absolute or relative.""" + """Return True for SoulSync image proxy/cache URLs, absolute or relative.""" if not url: return False try: parsed = urlparse(url) - return parsed.path == '/api/image-proxy' + return parsed.path == '/api/image-proxy' or parsed.path.startswith('/api/image-cache/') except Exception: return False @@ -235,10 +235,19 @@ def _browser_safe_image_url(url: str) -> str: if is_image_proxy_url(url): return url - if url.startswith('/api/image-proxy?url='): + if url.startswith('/api/image-proxy?url=') or url.startswith('/api/image-cache/'): return url if url.startswith('http://') or url.startswith('https://'): + try: + from core.image_cache import cached_image_url + + cached_url = cached_image_url(url) + if cached_url: + return cached_url + except Exception as exc: + logger.debug("image cache URL registration failed: %s", exc) + if is_internal_image_host(url): return f"/api/image-proxy?url={quote(url, safe='')}" return url diff --git a/core/prowlarr_client.py b/core/prowlarr_client.py new file mode 100644 index 00000000..4290f336 --- /dev/null +++ b/core/prowlarr_client.py @@ -0,0 +1,241 @@ +"""Prowlarr client — indexer aggregator. + +Prowlarr is the indexer manager component of the *arr stack. It exposes +configured Usenet / torrent indexers behind a single Newznab-style API +so downstream apps (Lidarr, Sonarr, Radarr, SoulSync) don't have to +implement an indexer integration per provider. + +This client is NOT a download source plugin. It does not implement +``DownloadSourcePlugin`` — Prowlarr only *searches*. The torrent / +usenet download plugins (built in subsequent commits) own the +add-to-client / poll-status / extract flow and call this client for +the search step. + +Surface: +- ``is_configured()`` — URL + API key present. +- ``check_connection()`` — hits ``/api/v1/system/status``. +- ``get_indexers()`` — list of configured indexers (id, name, protocol, + capabilities). +- ``search(query, categories, indexer_ids)`` — Newznab search across + selected indexers. Music categories default to the full audio tree. + +Auth: ``X-Api-Key`` header. Found in Prowlarr → Settings → General → +Security → API Key. +""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Sequence + +import requests as http_requests + +from config.settings import config_manager +from utils.logging_config import get_logger + +logger = get_logger("prowlarr_client") + + +# Newznab Music category tree. Prowlarr / Jackett / Newznab indexers +# all agree on these numeric IDs. 3000 is the parent — most indexers +# tag releases against the parent OR a leaf; searching the parent +# pulls everything. +MUSIC_CATEGORY_ALL = 3000 +MUSIC_CATEGORY_MP3 = 3010 +MUSIC_CATEGORY_VIDEO = 3020 +MUSIC_CATEGORY_AUDIOBOOK = 3030 +MUSIC_CATEGORY_LOSSLESS = 3040 +MUSIC_CATEGORY_OTHER = 3050 +MUSIC_CATEGORY_FOREIGN = 3060 + +DEFAULT_MUSIC_CATEGORIES: tuple = ( + MUSIC_CATEGORY_ALL, + MUSIC_CATEGORY_MP3, + MUSIC_CATEGORY_LOSSLESS, + MUSIC_CATEGORY_OTHER, +) + + +@dataclass +class ProwlarrIndexer: + """One configured indexer exposed by Prowlarr.""" + + id: int + name: str + protocol: str # "torrent" | "usenet" + enable: bool + privacy: str # "public" | "private" | "semiPrivate" + categories: List[int] = field(default_factory=list) + capabilities: Dict[str, Any] = field(default_factory=dict) + + +@dataclass +class ProwlarrSearchResult: + """One release returned by a Prowlarr search. + + ``download_url`` is the link the torrent / usenet client gets fed. + For torrent indexers it may be either a ``.torrent`` HTTP URL or + a magnet URI (sometimes both — ``magnet_uri`` is set when the + indexer exposes the magnet separately). + """ + + guid: str + title: str + indexer_id: int + indexer_name: str + protocol: str # "torrent" | "usenet" + download_url: Optional[str] = None + magnet_uri: Optional[str] = None + info_url: Optional[str] = None + size: int = 0 # bytes + seeders: Optional[int] = None + leechers: Optional[int] = None + grabs: Optional[int] = None + publish_date: Optional[str] = None + categories: List[int] = field(default_factory=list) + raw: Dict[str, Any] = field(default_factory=dict) + + +class ProwlarrClient: + """Thin sync-backed async wrapper around the Prowlarr v1 API.""" + + DEFAULT_TIMEOUT = 15 + + def __init__(self) -> None: + self._load_config() + + def _load_config(self) -> None: + self._url = (config_manager.get('prowlarr.url', '') or '').rstrip('/') + self._api_key = config_manager.get('prowlarr.api_key', '') or '' + + def reload_settings(self) -> None: + self._load_config() + logger.info("Prowlarr settings reloaded") + + def is_configured(self) -> bool: + return bool(self._url and self._api_key) + + async def check_connection(self) -> bool: + if not self.is_configured(): + return False + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._check_connection_sync) + + def _check_connection_sync(self) -> bool: + data = self._api_get('system/status') + return bool(data and 'version' in data) + + async def get_indexers(self) -> List[ProwlarrIndexer]: + if not self.is_configured(): + return [] + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._get_indexers_sync) + + def _get_indexers_sync(self) -> List[ProwlarrIndexer]: + data = self._api_get('indexer') + if not isinstance(data, list): + return [] + return [self._parse_indexer(entry) for entry in data if isinstance(entry, dict)] + + async def search( + self, + query: str, + categories: Sequence[int] = DEFAULT_MUSIC_CATEGORIES, + indexer_ids: Optional[Sequence[int]] = None, + limit: int = 100, + ) -> List[ProwlarrSearchResult]: + """Run a Newznab search across the selected indexers. + + ``indexer_ids`` is the list of Prowlarr internal indexer IDs to + query. ``None`` means all enabled indexers. + """ + if not self.is_configured() or not query.strip(): + return [] + loop = asyncio.get_event_loop() + return await loop.run_in_executor( + None, self._search_sync, query, list(categories), list(indexer_ids or []), limit + ) + + def _search_sync( + self, + query: str, + categories: List[int], + indexer_ids: List[int], + limit: int, + ) -> List[ProwlarrSearchResult]: + # Prowlarr's search endpoint accepts repeated params: ``categories=3000&categories=3010``. + # ``requests`` serializes lists in that exact form when passed as tuples of pairs. + params: List[tuple] = [('query', query), ('type', 'search'), ('limit', limit)] + for cat in categories: + params.append(('categories', cat)) + for indexer_id in indexer_ids: + params.append(('indexerIds', indexer_id)) + + data = self._api_get('search', params=params) + if not isinstance(data, list): + return [] + return [self._parse_result(entry) for entry in data if isinstance(entry, dict)] + + def _parse_indexer(self, entry: Dict[str, Any]) -> ProwlarrIndexer: + return ProwlarrIndexer( + id=int(entry.get('id') or 0), + name=entry.get('name') or '', + protocol=entry.get('protocol') or '', + enable=bool(entry.get('enable', True)), + privacy=entry.get('privacy') or '', + categories=[int(c.get('id') or 0) for c in entry.get('capabilities', {}).get('categories', []) if isinstance(c, dict)], + capabilities=entry.get('capabilities', {}) or {}, + ) + + def _parse_result(self, entry: Dict[str, Any]) -> ProwlarrSearchResult: + cats = entry.get('categories') or [] + category_ids: List[int] = [] + for cat in cats: + if isinstance(cat, dict) and cat.get('id') is not None: + try: + category_ids.append(int(cat['id'])) + except (TypeError, ValueError): + continue + elif isinstance(cat, int): + category_ids.append(cat) + + return ProwlarrSearchResult( + guid=str(entry.get('guid') or entry.get('infoUrl') or entry.get('downloadUrl') or ''), + title=entry.get('title') or '', + indexer_id=int(entry.get('indexerId') or 0), + indexer_name=entry.get('indexer') or '', + protocol=entry.get('protocol') or '', + download_url=entry.get('downloadUrl') or None, + magnet_uri=entry.get('magnetUrl') or None, + info_url=entry.get('infoUrl') or None, + size=int(entry.get('size') or 0), + seeders=entry.get('seeders'), + leechers=entry.get('leechers'), + grabs=entry.get('grabs'), + publish_date=entry.get('publishDate'), + categories=category_ids, + raw=entry, + ) + + def _api_get(self, path: str, params=None) -> Optional[Any]: + if not self.is_configured(): + return None + url = f"{self._url}/api/v1/{path.lstrip('/')}" + try: + resp = http_requests.get( + url, + headers={'X-Api-Key': self._api_key, 'Accept': 'application/json'}, + params=params, + timeout=self.DEFAULT_TIMEOUT, + ) + if not resp.ok: + logger.warning("Prowlarr %s returned HTTP %s", path, resp.status_code) + return None + return resp.json() + except http_requests.exceptions.RequestException as e: + logger.error("Prowlarr request to %s failed: %s", path, e) + return None + except ValueError as e: + logger.error("Prowlarr response to %s was not JSON: %s", path, e) + return None diff --git a/core/tidal_download_client.py b/core/tidal_download_client.py index a7efb40f..6ae2027e 100644 --- a/core/tidal_download_client.py +++ b/core/tidal_download_client.py @@ -95,6 +95,16 @@ HLS_MAP_TAG_RE = re.compile(r'#EXT-X-MAP:.*URI="([^"]+)"') from core.download_plugins.base import DownloadSourcePlugin +def _looks_like_json_decode_error(exc: Exception) -> bool: + name = exc.__class__.__name__.lower() + message = str(exc).lower() + return ( + "jsondecodeerror" in name + or "expecting value" in message + or "could not decode json" in message + ) + + class TidalDownloadClient(DownloadSourcePlugin): """ Tidal download client using tidalapi. @@ -230,6 +240,18 @@ class TidalDownloadClient(DownloadSourcePlugin): return {'status': 'error', 'message': 'Auth completed but session invalid'} except Exception as e: + if _looks_like_json_decode_error(e): + logger.error( + "Tidal device auth check received a non-JSON response from Tidal's token endpoint: %s", + e, + ) + return { + 'status': 'error', + 'message': ( + "Tidal returned an invalid auth response while SoulSync was finishing login. " + "Try again after disabling VPN/proxy/network filtering, then restart SoulSync if it repeats." + ), + } logger.error(f"Tidal device auth check error: {e}") return {'status': 'error', 'message': str(e)} diff --git a/core/torrent_clients/__init__.py b/core/torrent_clients/__init__.py new file mode 100644 index 00000000..ce23698c --- /dev/null +++ b/core/torrent_clients/__init__.py @@ -0,0 +1,56 @@ +"""Torrent client adapters. + +Each adapter wraps one BitTorrent client (qBittorrent, Transmission, +Deluge) behind the ``TorrentClientAdapter`` Protocol so the rest of +SoulSync can talk to whichever client the user picked through one +uniform surface. + +The active adapter is selected at runtime by the ``torrent_client.type`` +config key. See ``get_active_adapter()`` for the factory. +""" + +from __future__ import annotations + +from typing import Optional + +from config.settings import config_manager + +from core.torrent_clients.base import TorrentClientAdapter, TorrentStatus +from core.torrent_clients.deluge import DelugeAdapter +from core.torrent_clients.qbittorrent import QBittorrentAdapter +from core.torrent_clients.transmission import TransmissionAdapter + +__all__ = [ + "TorrentClientAdapter", + "TorrentStatus", + "QBittorrentAdapter", + "TransmissionAdapter", + "DelugeAdapter", + "get_active_adapter", + "adapter_for_type", +] + + +def adapter_for_type(client_type: str) -> Optional[TorrentClientAdapter]: + """Build a fresh adapter instance for the given client type string. + + ``None`` for unknown types so callers can present a helpful error + rather than crashing on a typo'd config value. + """ + if client_type == "qbittorrent": + return QBittorrentAdapter() + if client_type == "transmission": + return TransmissionAdapter() + if client_type == "deluge": + return DelugeAdapter() + return None + + +def get_active_adapter() -> Optional[TorrentClientAdapter]: + """Return an adapter for whichever torrent client the user has + selected in Settings. Reads ``torrent_client.type`` each call so a + settings change is picked up without restarting the process.""" + client_type = (config_manager.get('torrent_client.type', '') or '').strip().lower() + if not client_type: + return None + return adapter_for_type(client_type) diff --git a/core/torrent_clients/base.py b/core/torrent_clients/base.py new file mode 100644 index 00000000..4efe936e --- /dev/null +++ b/core/torrent_clients/base.py @@ -0,0 +1,110 @@ +"""Torrent client adapter contract. + +``TorrentClientAdapter`` is a structural Protocol — any class with +these method signatures is treated as a valid adapter. The download +plugin layer (built in a later commit) dispatches generically against +this surface so it doesn't have to know whether the user picked +qBittorrent, Transmission, or Deluge. + +The contract intentionally hides protocol-specific details: +- qBittorrent uses cookie auth + multipart form uploads. +- Transmission uses an X-Transmission-Session-Id header + JSON RPC. +- Deluge 2.x uses /json with a session cookie. + +All three converge on the same eight verbs below. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import List, Optional, Protocol, runtime_checkable + + +@dataclass +class TorrentStatus: + """Adapter-uniform view of one torrent's live state. + + Field semantics: + - ``state`` is one of: ``queued`` | ``downloading`` | ``seeding`` | + ``paused`` | ``stalled`` | ``error`` | ``completed``. Each + adapter maps its native state names to this set. + - ``progress`` is 0.0–1.0. + - ``save_path`` is where files land on the torrent client's host. + For remote clients this is a path on the *remote* machine. + - ``files`` is the list of relative paths inside the torrent. Empty + until the client has finished fetching the metadata. + """ + + id: str # torrent hash (qBit, Deluge) or numeric id (Transmission) + name: str + state: str + progress: float + size: int # total size in bytes + downloaded: int # bytes downloaded so far + download_speed: int # bytes/sec + upload_speed: int # bytes/sec + seeders: int = 0 + peers: int = 0 + eta: Optional[int] = None # seconds, None if unknown + save_path: Optional[str] = None + files: Optional[List[str]] = None + error: Optional[str] = None + + +@runtime_checkable +class TorrentClientAdapter(Protocol): + """Structural contract every torrent-client adapter implements.""" + + def is_configured(self) -> bool: + """True when the adapter has a URL and any credentials it + needs. Reads from config_manager — never raises on missing + config, just returns False so the orchestrator can dim the + torrent download source in the UI.""" + ... + + async def check_connection(self) -> bool: + """Probe the client over the network. Logs in if required.""" + ... + + async def add_torrent( + self, + url_or_magnet: str, + category: str = "soulsync", + save_path: Optional[str] = None, + ) -> Optional[str]: + """Hand the torrent client a HTTP/HTTPS URL pointing to a + ``.torrent`` file or a ``magnet:`` URI. Returns the torrent's + client-side identifier (info-hash for qBit / Deluge, numeric + id for Transmission) or ``None`` on failure.""" + ... + + async def add_torrent_file( + self, + file_bytes: bytes, + category: str = "soulsync", + save_path: Optional[str] = None, + ) -> Optional[str]: + """Upload a raw ``.torrent`` payload. Same return as + ``add_torrent``. Used when the indexer doesn't expose a + direct download URL and SoulSync had to fetch the file + itself first.""" + ... + + async def get_status(self, torrent_id: str) -> Optional[TorrentStatus]: + """Return live status for one torrent, or ``None`` if the + client doesn't know about it.""" + ... + + async def get_all(self) -> List[TorrentStatus]: + """Return live status for every torrent the client currently + tracks. Used by the global download list.""" + ... + + async def remove(self, torrent_id: str, delete_files: bool = False) -> bool: + """Remove the torrent from the client. ``delete_files=True`` + also deletes the downloaded data on disk.""" + ... + + async def pause(self, torrent_id: str) -> bool: ... + + async def resume(self, torrent_id: str) -> bool: ... diff --git a/core/torrent_clients/deluge.py b/core/torrent_clients/deluge.py new file mode 100644 index 00000000..7c7e6ae9 --- /dev/null +++ b/core/torrent_clients/deluge.py @@ -0,0 +1,307 @@ +"""Deluge 2.x JSON-RPC adapter. + +Auth model: POST ``/json`` with method ``auth.login`` returns a +``_session_id`` cookie. The session cookie + every subsequent JSON-RPC +call must include a monotonically-incrementing ``id`` field. ``params`` +is a list (positional), not an object. + +Reference: https://deluge.readthedocs.io/en/latest/reference/api.html +and https://deluge.readthedocs.io/en/latest/reference/webapi.html +""" + +from __future__ import annotations + +import asyncio +import base64 +import threading +from itertools import count +from typing import Any, List, Optional + +import requests as http_requests + +from config.settings import config_manager +from core.torrent_clients.base import TorrentStatus +from utils.logging_config import get_logger + +logger = get_logger("torrent.deluge") + + +# Deluge native state strings → adapter-uniform set. +_DELUGE_STATE_MAP = { + "Allocating": "queued", + "Checking": "queued", + "Downloading": "downloading", + "Seeding": "seeding", + "Paused": "paused", + "Queued": "queued", + "Error": "error", + "Moving": "queued", +} + + +def _map_state(deluge_state: str, progress: float) -> str: + mapped = _DELUGE_STATE_MAP.get(deluge_state or '', "error") + if mapped == "paused" and progress >= 1.0: + return "completed" + return mapped + + +class DelugeAdapter: + """Deluge 2.x WebUI JSON-RPC adapter.""" + + DEFAULT_TIMEOUT = 15 + + # Fields we ask ``core.get_torrents_status`` to return — explicit + # to keep payload size predictable across Deluge versions that + # add fields over time. + _STATUS_FIELDS = [ + 'hash', 'name', 'state', 'progress', 'total_size', + 'total_done', 'download_payload_rate', 'upload_payload_rate', + 'num_seeds', 'num_peers', 'eta', 'save_path', 'tracker_status', + ] + + def __init__(self) -> None: + self._session: Optional[http_requests.Session] = None + self._session_lock = threading.Lock() + self._id_counter = count(1) + self._load_config() + + def _load_config(self) -> None: + self._url = (config_manager.get('torrent_client.url', '') or '').rstrip('/') + # Deluge's WebUI auth uses a single password, not username+password. + # We accept whichever field the user filled in — keeps the UI uniform. + self._password = ( + config_manager.get('torrent_client.password', '') + or config_manager.get('torrent_client.username', '') + or '' + ) + self._category = config_manager.get('torrent_client.category', 'soulsync') or 'soulsync' + self._save_path = config_manager.get('torrent_client.save_path', '') or '' + with self._session_lock: + self._session = None + + def reload_settings(self) -> None: + self._load_config() + + def is_configured(self) -> bool: + # Deluge WebUI requires a password; without it auth.login fails + # outright. Refuse the configuration up front rather than letting + # every call 401. + return bool(self._url and self._password) + + async def check_connection(self) -> bool: + if not self.is_configured(): + return False + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._check_connection_sync) + + def _check_connection_sync(self) -> bool: + sess = self._ensure_session_sync() + if sess is None: + return False + # web.connected() returns True iff the WebUI's daemon link is up. + # If daemons aren't auto-connected, fall back to a cheap call + # that just confirms we're authenticated. + result = self._rpc_sync('web.connected', []) + if result is True: + return True + # Fall back to a generic auth probe. + return self._rpc_sync('auth.check_session', []) is not None + + def _ensure_session_sync(self) -> Optional[http_requests.Session]: + with self._session_lock: + if self._session is not None: + return self._session + sess = http_requests.Session() + self._session = sess + # Log in. auth.login takes the password as a single positional arg. + result = self._rpc_sync('auth.login', [self._password]) + if result is not True: + logger.error("Deluge auth.login returned %r", result) + with self._session_lock: + self._session = None + return None + with self._session_lock: + return self._session + + def _rpc_sync(self, method: str, params: list) -> Any: + if not self._url: + return None + with self._session_lock: + sess = self._session + if sess is None: + # Bootstrap a session container; auth.login will populate it. + with self._session_lock: + if self._session is None: + self._session = http_requests.Session() + sess = self._session + payload = { + 'id': next(self._id_counter), + 'method': method, + 'params': params, + } + try: + resp = sess.post( + f"{self._url}/json", json=payload, + headers={'Content-Type': 'application/json'}, + timeout=self.DEFAULT_TIMEOUT, + ) + if not resp.ok: + logger.warning("Deluge %s returned HTTP %s", method, resp.status_code) + return None + data = resp.json() + err = data.get('error') + if err: + # Code 1 = unknown method, 2 = bad params, etc. + # Code 'No Auth' surfaces as a string in some versions. + logger.warning("Deluge %s error: %r", method, err) + return None + return data.get('result') + except Exception as e: + logger.error("Deluge %s call failed: %s", method, e) + return None + + async def add_torrent( + self, + url_or_magnet: str, + category: str = "soulsync", + save_path: Optional[str] = None, + ) -> Optional[str]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor( + None, self._add_torrent_sync, url_or_magnet, category, save_path + ) + + def _add_torrent_sync( + self, + url_or_magnet: str, + category: str, + save_path: Optional[str], + ) -> Optional[str]: + if self._ensure_session_sync() is None: + return None + options: dict = {} + if save_path or self._save_path: + options['download_location'] = save_path or self._save_path + # Deluge distinguishes magnet URIs from HTTP .torrent URLs at + # the API layer — different method names. + if url_or_magnet.startswith('magnet:'): + method = 'core.add_torrent_magnet' + else: + method = 'core.add_torrent_url' + torrent_hash = self._rpc_sync(method, [url_or_magnet, options]) + if not torrent_hash: + return None + self._apply_label(str(torrent_hash), category or self._category) + return str(torrent_hash) + + async def add_torrent_file( + self, + file_bytes: bytes, + category: str = "soulsync", + save_path: Optional[str] = None, + ) -> Optional[str]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor( + None, self._add_torrent_file_sync, file_bytes, category, save_path + ) + + def _add_torrent_file_sync( + self, + file_bytes: bytes, + category: str, + save_path: Optional[str], + ) -> Optional[str]: + if self._ensure_session_sync() is None: + return None + options: dict = {} + if save_path or self._save_path: + options['download_location'] = save_path or self._save_path + encoded = base64.b64encode(file_bytes).decode('ascii') + torrent_hash = self._rpc_sync('core.add_torrent_file', ['soulsync.torrent', encoded, options]) + if not torrent_hash: + return None + self._apply_label(str(torrent_hash), category or self._category) + return str(torrent_hash) + + def _apply_label(self, torrent_hash: str, label: str) -> None: + """Best-effort label assignment. The Label plugin is optional + in Deluge — if it isn't installed the RPC call fails silently + and we don't propagate the error: the torrent is still added.""" + if not label: + return + # Ensure the label exists before assigning. + self._rpc_sync('label.add', [label]) + self._rpc_sync('label.set_torrent', [torrent_hash, label]) + + async def get_status(self, torrent_id: str) -> Optional[TorrentStatus]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._get_status_sync, torrent_id) + + def _get_status_sync(self, torrent_id: str) -> Optional[TorrentStatus]: + result = self._rpc_sync('core.get_torrent_status', [torrent_id, self._STATUS_FIELDS]) + if not isinstance(result, dict) or not result: + return None + # core.get_torrent_status doesn't echo back the hash — patch it in. + result.setdefault('hash', torrent_id) + return self._parse_status(result) + + async def get_all(self) -> List[TorrentStatus]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._get_all_sync) + + def _get_all_sync(self) -> List[TorrentStatus]: + result = self._rpc_sync('core.get_torrents_status', [{}, self._STATUS_FIELDS]) + if not isinstance(result, dict): + return [] + out = [] + for hash_id, item in result.items(): + if not isinstance(item, dict): + continue + item.setdefault('hash', hash_id) + out.append(self._parse_status(item)) + return out + + def _parse_status(self, item: dict) -> TorrentStatus: + progress = float(item.get('progress') or 0.0) + # Deluge expresses progress as 0-100; normalize to 0-1. + if progress > 1.0: + progress = progress / 100.0 + return TorrentStatus( + id=str(item.get('hash') or ''), + name=item.get('name') or '', + state=_map_state(item.get('state') or '', progress), + progress=progress, + size=int(item.get('total_size') or 0), + downloaded=int(item.get('total_done') or 0), + download_speed=int(item.get('download_payload_rate') or 0), + upload_speed=int(item.get('upload_payload_rate') or 0), + seeders=int(item.get('num_seeds') or 0), + peers=int(item.get('num_peers') or 0), + eta=int(item['eta']) if isinstance(item.get('eta'), (int, float)) and item.get('eta', 0) > 0 else None, + save_path=item.get('save_path'), + error=item.get('tracker_status') if 'Error' in (item.get('state') or '') else None, + ) + + async def remove(self, torrent_id: str, delete_files: bool = False) -> bool: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._remove_sync, torrent_id, delete_files) + + def _remove_sync(self, torrent_id: str, delete_files: bool) -> bool: + result = self._rpc_sync('core.remove_torrent', [torrent_id, bool(delete_files)]) + return result is True + + async def pause(self, torrent_id: str) -> bool: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._pause_sync, torrent_id) + + def _pause_sync(self, torrent_id: str) -> bool: + # Deluge 2.x core.pause_torrent takes a list of hashes. + return self._rpc_sync('core.pause_torrent', [[torrent_id]]) is not None + + async def resume(self, torrent_id: str) -> bool: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._resume_sync, torrent_id) + + def _resume_sync(self, torrent_id: str) -> bool: + return self._rpc_sync('core.resume_torrent', [[torrent_id]]) is not None diff --git a/core/torrent_clients/qbittorrent.py b/core/torrent_clients/qbittorrent.py new file mode 100644 index 00000000..3373086e --- /dev/null +++ b/core/torrent_clients/qbittorrent.py @@ -0,0 +1,289 @@ +"""qBittorrent WebUI v2 adapter. + +Auth model: POST ``/api/v2/auth/login`` with form-encoded +``username`` + ``password`` returns a ``SID`` cookie that's required +on every subsequent call. The cookie lives on a ``requests.Session`` +maintained by this adapter; we lazily re-login on 403 in case the +server expired the session. + +Reference: https://github.com/qbittorrent/qBittorrent/wiki/WebUI-API-(qBittorrent-4.1) +""" + +from __future__ import annotations + +import asyncio +import threading +from typing import List, Optional + +import requests as http_requests + +from config.settings import config_manager +from core.torrent_clients.base import TorrentStatus +from utils.logging_config import get_logger + +logger = get_logger("torrent.qbittorrent") + + +# qBittorrent's native state strings. Mapped onto the adapter-uniform +# set in ``_map_state``. See https://github.com/qbittorrent/qBittorrent/wiki/WebUI-API-(qBittorrent-4.1)#torrent-list +_QBIT_STATE_MAP = { + "allocating": "queued", + "checkingDL": "queued", + "checkingUP": "seeding", + "checkingResumeData": "queued", + "downloading": "downloading", + "error": "error", + "forcedDL": "downloading", + "forcedUP": "seeding", + "metaDL": "downloading", + "missingFiles": "error", + "moving": "queued", + "pausedDL": "paused", + "pausedUP": "completed", + "queuedDL": "queued", + "queuedUP": "queued", + "stalledDL": "stalled", + "stalledUP": "seeding", + "uploading": "seeding", + "unknown": "error", +} + + +def _map_state(qbit_state: str) -> str: + return _QBIT_STATE_MAP.get(qbit_state, "error") + + +class QBittorrentAdapter: + """qBittorrent WebUI v2 adapter.""" + + DEFAULT_TIMEOUT = 15 + + def __init__(self) -> None: + self._session: Optional[http_requests.Session] = None + self._session_lock = threading.Lock() + self._load_config() + + def _load_config(self) -> None: + self._url = (config_manager.get('torrent_client.url', '') or '').rstrip('/') + self._username = config_manager.get('torrent_client.username', '') or '' + self._password = config_manager.get('torrent_client.password', '') or '' + self._category = config_manager.get('torrent_client.category', 'soulsync') or 'soulsync' + self._save_path = config_manager.get('torrent_client.save_path', '') or '' + # Drop any existing session — credentials may have changed. + with self._session_lock: + self._session = None + + def reload_settings(self) -> None: + self._load_config() + + def is_configured(self) -> bool: + # qBittorrent allows no-auth setups (LAN), so credentials are + # optional — URL is the only hard requirement. + return bool(self._url) + + async def check_connection(self) -> bool: + if not self.is_configured(): + return False + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._check_connection_sync) + + def _check_connection_sync(self) -> bool: + try: + sess = self._ensure_session_sync() + if sess is None: + return False + resp = sess.get(f"{self._url}/api/v2/app/version", timeout=self.DEFAULT_TIMEOUT) + return resp.ok + except Exception as e: + logger.error("qBittorrent connection probe failed: %s", e) + return False + + def _ensure_session_sync(self) -> Optional[http_requests.Session]: + with self._session_lock: + if self._session is not None: + return self._session + sess = http_requests.Session() + # No-auth setup — skip login. + if not self._username and not self._password: + self._session = sess + return sess + try: + resp = sess.post( + f"{self._url}/api/v2/auth/login", + data={'username': self._username, 'password': self._password}, + # qBittorrent rejects login attempts that arrive without a + # Referer matching its configured host (CSRF guard). Sending + # the WebUI's own URL satisfies the check. + headers={'Referer': self._url}, + timeout=self.DEFAULT_TIMEOUT, + ) + if not resp.ok or resp.text.strip() != 'Ok.': + logger.error("qBittorrent login failed: HTTP %s body=%r", resp.status_code, resp.text[:200]) + return None + self._session = sess + return sess + except Exception as e: + logger.error("qBittorrent login error: %s", e) + return None + + def _call(self, method: str, path: str, **kwargs) -> Optional[http_requests.Response]: + sess = self._ensure_session_sync() + if sess is None: + return None + try: + kwargs.setdefault('timeout', self.DEFAULT_TIMEOUT) + kwargs.setdefault('headers', {}).setdefault('Referer', self._url) + resp = sess.request(method, f"{self._url}{path}", **kwargs) + # Session expired — try one re-login and retry. + if resp.status_code == 403: + with self._session_lock: + self._session = None + sess = self._ensure_session_sync() + if sess is None: + return None + resp = sess.request(method, f"{self._url}{path}", **kwargs) + return resp + except Exception as e: + logger.error("qBittorrent %s %s failed: %s", method, path, e) + return None + + async def add_torrent( + self, + url_or_magnet: str, + category: str = "soulsync", + save_path: Optional[str] = None, + ) -> Optional[str]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor( + None, self._add_torrent_sync, url_or_magnet, category, save_path + ) + + def _add_torrent_sync( + self, + url_or_magnet: str, + category: str, + save_path: Optional[str], + ) -> Optional[str]: + data = {'urls': url_or_magnet, 'category': category or self._category} + if save_path or self._save_path: + data['savepath'] = save_path or self._save_path + resp = self._call('POST', '/api/v2/torrents/add', data=data) + if not resp or not resp.ok: + return None + # qBittorrent's /add endpoint returns 200 "Ok." on success but + # NOT the resulting info-hash. We have to look it up via + # /torrents/info filtered by category — the just-added torrent + # will be the newest entry in that category. + return self._lookup_latest_hash(category or self._category) + + def _lookup_latest_hash(self, category: str) -> Optional[str]: + resp = self._call('GET', '/api/v2/torrents/info', params={'category': category, 'sort': 'added_on', 'reverse': 'true', 'limit': 1}) + if not resp or not resp.ok: + return None + try: + items = resp.json() + if items: + return items[0].get('hash') + except Exception as e: + logger.error("qBittorrent /torrents/info parse failed: %s", e) + return None + + async def add_torrent_file( + self, + file_bytes: bytes, + category: str = "soulsync", + save_path: Optional[str] = None, + ) -> Optional[str]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor( + None, self._add_torrent_file_sync, file_bytes, category, save_path + ) + + def _add_torrent_file_sync( + self, + file_bytes: bytes, + category: str, + save_path: Optional[str], + ) -> Optional[str]: + data = {'category': category or self._category} + if save_path or self._save_path: + data['savepath'] = save_path or self._save_path + files = {'torrents': ('soulsync.torrent', file_bytes, 'application/x-bittorrent')} + resp = self._call('POST', '/api/v2/torrents/add', data=data, files=files) + if not resp or not resp.ok: + return None + return self._lookup_latest_hash(category or self._category) + + async def get_status(self, torrent_id: str) -> Optional[TorrentStatus]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._get_status_sync, torrent_id) + + def _get_status_sync(self, torrent_id: str) -> Optional[TorrentStatus]: + resp = self._call('GET', '/api/v2/torrents/info', params={'hashes': torrent_id}) + if not resp or not resp.ok: + return None + try: + items = resp.json() + except Exception as e: + logger.error("qBittorrent get_status parse failed: %s", e) + return None + if not items: + return None + return self._parse_status(items[0]) + + async def get_all(self) -> List[TorrentStatus]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._get_all_sync) + + def _get_all_sync(self) -> List[TorrentStatus]: + resp = self._call('GET', '/api/v2/torrents/info') + if not resp or not resp.ok: + return [] + try: + return [self._parse_status(item) for item in resp.json()] + except Exception as e: + logger.error("qBittorrent get_all parse failed: %s", e) + return [] + + def _parse_status(self, item: dict) -> TorrentStatus: + return TorrentStatus( + id=str(item.get('hash') or ''), + name=item.get('name') or '', + state=_map_state(item.get('state') or 'unknown'), + progress=float(item.get('progress') or 0.0), + size=int(item.get('size') or 0), + downloaded=int(item.get('downloaded') or 0), + download_speed=int(item.get('dlspeed') or 0), + upload_speed=int(item.get('upspeed') or 0), + seeders=int(item.get('num_seeds') or 0), + peers=int(item.get('num_leechs') or 0), + eta=item.get('eta') if isinstance(item.get('eta'), int) and item.get('eta', 0) > 0 else None, + save_path=item.get('save_path'), + ) + + async def remove(self, torrent_id: str, delete_files: bool = False) -> bool: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._remove_sync, torrent_id, delete_files) + + def _remove_sync(self, torrent_id: str, delete_files: bool) -> bool: + resp = self._call('POST', '/api/v2/torrents/delete', data={ + 'hashes': torrent_id, + 'deleteFiles': 'true' if delete_files else 'false', + }) + return bool(resp and resp.ok) + + async def pause(self, torrent_id: str) -> bool: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._pause_sync, torrent_id) + + def _pause_sync(self, torrent_id: str) -> bool: + resp = self._call('POST', '/api/v2/torrents/pause', data={'hashes': torrent_id}) + return bool(resp and resp.ok) + + async def resume(self, torrent_id: str) -> bool: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._resume_sync, torrent_id) + + def _resume_sync(self, torrent_id: str) -> bool: + resp = self._call('POST', '/api/v2/torrents/resume', data={'hashes': torrent_id}) + return bool(resp and resp.ok) diff --git a/core/torrent_clients/transmission.py b/core/torrent_clients/transmission.py new file mode 100644 index 00000000..266bda79 --- /dev/null +++ b/core/torrent_clients/transmission.py @@ -0,0 +1,266 @@ +"""Transmission RPC adapter. + +Auth model: Transmission uses HTTP Basic auth + an +``X-Transmission-Session-Id`` header. The session ID rotates and is +returned on every 409 response — the adapter caches the latest value +and replays the original request transparently. + +Reference: https://github.com/transmission/transmission/blob/main/docs/rpc-spec.md +""" + +from __future__ import annotations + +import asyncio +import base64 +import threading +from typing import List, Optional + +import requests as http_requests + +from config.settings import config_manager +from core.torrent_clients.base import TorrentStatus +from utils.logging_config import get_logger + +logger = get_logger("torrent.transmission") + + +# Transmission RPC status codes. Defined as numeric constants in the +# RPC spec — converted to the adapter-uniform string set here. +_TRANSMISSION_STATUS = { + 0: "paused", + 1: "queued", # queued to check files + 2: "queued", # checking files + 3: "queued", # queued to download + 4: "downloading", + 5: "queued", # queued to seed + 6: "seeding", +} + + +def _map_state(status_code: int, percent_done: float) -> str: + base = _TRANSMISSION_STATUS.get(status_code, "error") + # Transmission reports 'paused' (0) for both never-started and + # fully-downloaded-but-not-seeding. Differentiate by progress. + if status_code == 0 and percent_done >= 1.0: + return "completed" + return base + + +class TransmissionAdapter: + """Transmission RPC adapter (transmission-rpc protocol v17+).""" + + DEFAULT_TIMEOUT = 15 + + def __init__(self) -> None: + self._session_id: Optional[str] = None + self._session_id_lock = threading.Lock() + self._load_config() + + def _load_config(self) -> None: + url = (config_manager.get('torrent_client.url', '') or '').rstrip('/') + # Transmission's RPC endpoint is always /transmission/rpc — if the + # user pasted a bare host URL, append it. If they pasted the full + # /transmission/rpc URL, leave it. + if url and not url.endswith('/transmission/rpc'): + if '/transmission' not in url: + url = f"{url}/transmission/rpc" + self._url = url + self._username = config_manager.get('torrent_client.username', '') or '' + self._password = config_manager.get('torrent_client.password', '') or '' + self._category = config_manager.get('torrent_client.category', 'soulsync') or 'soulsync' + self._save_path = config_manager.get('torrent_client.save_path', '') or '' + with self._session_id_lock: + self._session_id = None + + def reload_settings(self) -> None: + self._load_config() + + def is_configured(self) -> bool: + return bool(self._url) + + async def check_connection(self) -> bool: + if not self.is_configured(): + return False + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._check_connection_sync) + + def _check_connection_sync(self) -> bool: + resp = self._rpc('session-get', {}) + return resp is not None + + def _rpc(self, method: str, arguments: dict) -> Optional[dict]: + """One Transmission RPC round-trip. Handles the 409 + session-id renegotiation transparently — Transmission rejects + the first call with HTTP 409 and a fresh + ``X-Transmission-Session-Id`` header, which subsequent calls + must echo back.""" + if not self._url: + return None + auth = (self._username, self._password) if self._username else None + payload = {'method': method, 'arguments': arguments} + for _attempt in range(2): + try: + with self._session_id_lock: + sid = self._session_id + headers = {'Content-Type': 'application/json'} + if sid: + headers['X-Transmission-Session-Id'] = sid + resp = http_requests.post( + self._url, json=payload, headers=headers, auth=auth, + timeout=self.DEFAULT_TIMEOUT, + ) + if resp.status_code == 409: + # Pick up the new session id and retry. + new_sid = resp.headers.get('X-Transmission-Session-Id') + if not new_sid: + logger.error("Transmission 409 with no X-Transmission-Session-Id header") + return None + with self._session_id_lock: + self._session_id = new_sid + continue + if not resp.ok: + logger.warning("Transmission RPC %s returned HTTP %s", method, resp.status_code) + return None + data = resp.json() + if data.get('result') != 'success': + logger.warning("Transmission RPC %s result=%s", method, data.get('result')) + return None + return data.get('arguments', {}) + except Exception as e: + logger.error("Transmission RPC %s failed: %s", method, e) + return None + return None + + async def add_torrent( + self, + url_or_magnet: str, + category: str = "soulsync", + save_path: Optional[str] = None, + ) -> Optional[str]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor( + None, self._add_torrent_sync, url_or_magnet, category, save_path + ) + + def _add_torrent_sync( + self, + url_or_magnet: str, + category: str, + save_path: Optional[str], + ) -> Optional[str]: + args: dict = {'filename': url_or_magnet, 'labels': [category or self._category]} + if save_path or self._save_path: + args['download-dir'] = save_path or self._save_path + return self._add_torrent_finish(args) + + async def add_torrent_file( + self, + file_bytes: bytes, + category: str = "soulsync", + save_path: Optional[str] = None, + ) -> Optional[str]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor( + None, self._add_torrent_file_sync, file_bytes, category, save_path + ) + + def _add_torrent_file_sync( + self, + file_bytes: bytes, + category: str, + save_path: Optional[str], + ) -> Optional[str]: + args: dict = { + 'metainfo': base64.b64encode(file_bytes).decode('ascii'), + 'labels': [category or self._category], + } + if save_path or self._save_path: + args['download-dir'] = save_path or self._save_path + return self._add_torrent_finish(args) + + def _add_torrent_finish(self, args: dict) -> Optional[str]: + data = self._rpc('torrent-add', args) + if not data: + return None + # torrent-add returns either ``torrent-added`` (new torrent) or + # ``torrent-duplicate`` (already exists). Both carry the hash. + torrent = data.get('torrent-added') or data.get('torrent-duplicate') + if not torrent: + return None + return str(torrent.get('hashString') or '') or None + + async def get_status(self, torrent_id: str) -> Optional[TorrentStatus]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._get_status_sync, torrent_id) + + def _get_status_sync(self, torrent_id: str) -> Optional[TorrentStatus]: + data = self._rpc('torrent-get', { + 'ids': [torrent_id], + 'fields': self._STATUS_FIELDS, + }) + if not data or not data.get('torrents'): + return None + return self._parse_status(data['torrents'][0]) + + async def get_all(self) -> List[TorrentStatus]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._get_all_sync) + + def _get_all_sync(self) -> List[TorrentStatus]: + data = self._rpc('torrent-get', {'fields': self._STATUS_FIELDS}) + if not data: + return [] + return [self._parse_status(t) for t in data.get('torrents', [])] + + _STATUS_FIELDS = [ + 'hashString', 'name', 'status', 'percentDone', 'totalSize', + 'downloadedEver', 'rateDownload', 'rateUpload', 'peersSendingToUs', + 'peersGettingFromUs', 'eta', 'downloadDir', 'errorString', + ] + + def _parse_status(self, item: dict) -> TorrentStatus: + progress = float(item.get('percentDone') or 0.0) + status_code = int(item.get('status') or 0) + eta_raw = item.get('eta') + # Transmission returns -1 / -2 for "unknown" — surface as None. + eta = eta_raw if isinstance(eta_raw, int) and eta_raw > 0 else None + return TorrentStatus( + id=str(item.get('hashString') or ''), + name=item.get('name') or '', + state=_map_state(status_code, progress), + progress=progress, + size=int(item.get('totalSize') or 0), + downloaded=int(item.get('downloadedEver') or 0), + download_speed=int(item.get('rateDownload') or 0), + upload_speed=int(item.get('rateUpload') or 0), + seeders=int(item.get('peersSendingToUs') or 0), + peers=int(item.get('peersGettingFromUs') or 0), + eta=eta, + save_path=item.get('downloadDir'), + error=item.get('errorString') or None, + ) + + async def remove(self, torrent_id: str, delete_files: bool = False) -> bool: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._remove_sync, torrent_id, delete_files) + + def _remove_sync(self, torrent_id: str, delete_files: bool) -> bool: + data = self._rpc('torrent-remove', { + 'ids': [torrent_id], + 'delete-local-data': bool(delete_files), + }) + return data is not None + + async def pause(self, torrent_id: str) -> bool: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._pause_sync, torrent_id) + + def _pause_sync(self, torrent_id: str) -> bool: + return self._rpc('torrent-stop', {'ids': [torrent_id]}) is not None + + async def resume(self, torrent_id: str) -> bool: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._resume_sync, torrent_id) + + def _resume_sync(self, torrent_id: str) -> bool: + return self._rpc('torrent-start', {'ids': [torrent_id]}) is not None diff --git a/core/usenet_clients/__init__.py b/core/usenet_clients/__init__.py new file mode 100644 index 00000000..640b3ef3 --- /dev/null +++ b/core/usenet_clients/__init__.py @@ -0,0 +1,49 @@ +"""Usenet client adapters. + +Each adapter wraps one Usenet downloader (SABnzbd, NZBGet) behind +the ``UsenetClientAdapter`` Protocol so the rest of SoulSync can +talk to whichever client the user picked through one uniform +surface. + +The active adapter is selected at runtime by the +``usenet_client.type`` config key. See ``get_active_adapter()`` +for the factory. +""" + +from __future__ import annotations + +from typing import Optional + +from config.settings import config_manager + +from core.usenet_clients.base import UsenetClientAdapter, UsenetStatus +from core.usenet_clients.nzbget import NZBGetAdapter +from core.usenet_clients.sabnzbd import SABnzbdAdapter + +__all__ = [ + "UsenetClientAdapter", + "UsenetStatus", + "SABnzbdAdapter", + "NZBGetAdapter", + "get_active_adapter", + "adapter_for_type", +] + + +def adapter_for_type(client_type: str) -> Optional[UsenetClientAdapter]: + """Build a fresh adapter instance for the given client type string. + ``None`` for unknown types.""" + if client_type == "sabnzbd": + return SABnzbdAdapter() + if client_type == "nzbget": + return NZBGetAdapter() + return None + + +def get_active_adapter() -> Optional[UsenetClientAdapter]: + """Return an adapter for whichever usenet client the user has + selected in Settings. Reads ``usenet_client.type`` each call.""" + client_type = (config_manager.get('usenet_client.type', '') or '').strip().lower() + if not client_type: + return None + return adapter_for_type(client_type) diff --git a/core/usenet_clients/base.py b/core/usenet_clients/base.py new file mode 100644 index 00000000..f77b3d78 --- /dev/null +++ b/core/usenet_clients/base.py @@ -0,0 +1,74 @@ +"""Usenet client adapter contract. + +``UsenetClientAdapter`` mirrors ``TorrentClientAdapter`` in shape so +the download plugin layer can reuse the same dispatch pattern. +Differences from the torrent side: + +- No magnet URI equivalent — usenet jobs are always ``.nzb`` files + or URLs that resolve to one. +- No seed/peer counts — usenet is a download-only protocol. +- Status values reflect usenet semantics: ``downloading`` / + ``extracting`` / ``verifying`` / ``repairing`` / ``completed`` / + ``failed`` / ``paused``. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import List, Optional, Protocol, runtime_checkable + + +@dataclass +class UsenetStatus: + """Adapter-uniform view of one usenet job. + + Field semantics: + - ``state`` is one of: ``queued`` | ``downloading`` | ``extracting`` + | ``verifying`` | ``repairing`` | ``completed`` | ``failed`` | + ``paused``. Each adapter maps its native names to this set. + - ``progress`` is 0.0–1.0 across the entire job (download + par2 + + unpack), so a job stalled at the verify step still shows < 1.0. + """ + + id: str # SAB nzo_id / NZBGet NZBID + name: str + state: str + progress: float + size: int # total size in bytes + downloaded: int # bytes downloaded so far + download_speed: int # bytes/sec + eta: Optional[int] = None # seconds, None if unknown + save_path: Optional[str] = None + category: Optional[str] = None + files: Optional[List[str]] = None + error: Optional[str] = None + + +@runtime_checkable +class UsenetClientAdapter(Protocol): + """Structural contract every usenet-client adapter implements.""" + + def is_configured(self) -> bool: ... + + async def check_connection(self) -> bool: ... + + async def add_nzb( + self, + url_or_bytes, + category: str = "soulsync", + save_path: Optional[str] = None, + ) -> Optional[str]: + """Hand the usenet client either a ``.nzb`` HTTP URL (``str``) + or the raw payload (``bytes``). Returns the client-side job id + on success, ``None`` on failure.""" + ... + + async def get_status(self, job_id: str) -> Optional[UsenetStatus]: ... + + async def get_all(self) -> List[UsenetStatus]: ... + + async def remove(self, job_id: str, delete_files: bool = False) -> bool: ... + + async def pause(self, job_id: str) -> bool: ... + + async def resume(self, job_id: str) -> bool: ... diff --git a/core/usenet_clients/nzbget.py b/core/usenet_clients/nzbget.py new file mode 100644 index 00000000..d58074ff --- /dev/null +++ b/core/usenet_clients/nzbget.py @@ -0,0 +1,277 @@ +"""NZBGet adapter. + +Auth model: HTTP Basic auth on the JSON-RPC endpoint ``/jsonrpc``. +Every method takes positional ``params``. Identical pattern to +Deluge but with different method names. + +Reference: https://nzbget.com/documentation/api/ +""" + +from __future__ import annotations + +import asyncio +import base64 +from itertools import count +from typing import Any, List, Optional, Union + +import requests as http_requests + +from config.settings import config_manager +from core.usenet_clients.base import UsenetStatus +from utils.logging_config import get_logger + +logger = get_logger("usenet.nzbget") + + +# NZBGet's ``Status`` field on ListGroups → adapter-uniform set. +# NZBGet states (group): QUEUED, PAUSED, DOWNLOADING, FETCHING, PP_QUEUED, +# LOADING_PARS, VERIFYING_SOURCES, REPAIRING, VERIFYING_REPAIRED, RENAMING, +# UNPACKING, MOVING, EXECUTING_SCRIPT, PP_FINISHED. +_NZBGET_STATE_MAP = { + "QUEUED": "queued", + "PAUSED": "paused", + "DOWNLOADING": "downloading", + "FETCHING": "downloading", + "PP_QUEUED": "queued", + "LOADING_PARS": "verifying", + "VERIFYING_SOURCES": "verifying", + "REPAIRING": "repairing", + "VERIFYING_REPAIRED": "verifying", + "RENAMING": "extracting", + "UNPACKING": "extracting", + "MOVING": "extracting", + "EXECUTING_SCRIPT": "extracting", + "PP_FINISHED": "completed", +} + + +def _map_state(nzbget_state: str) -> str: + return _NZBGET_STATE_MAP.get(nzbget_state or '', "error") + + +class NZBGetAdapter: + """NZBGet JSON-RPC adapter.""" + + DEFAULT_TIMEOUT = 15 + + def __init__(self) -> None: + self._id_counter = count(1) + self._load_config() + + def _load_config(self) -> None: + self._url = (config_manager.get('usenet_client.url', '') or '').rstrip('/') + self._username = config_manager.get('usenet_client.username', '') or '' + self._password = config_manager.get('usenet_client.password', '') or '' + self._category = config_manager.get('usenet_client.category', 'soulsync') or 'soulsync' + + def reload_settings(self) -> None: + self._load_config() + + def is_configured(self) -> bool: + return bool(self._url and self._username and self._password) + + async def check_connection(self) -> bool: + if not self.is_configured(): + return False + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._check_connection_sync) + + def _check_connection_sync(self) -> bool: + return self._rpc_sync('version', []) is not None + + def _rpc_sync(self, method: str, params: list) -> Any: + if not self._url: + return None + try: + resp = http_requests.post( + f"{self._url}/jsonrpc", + json={'method': method, 'params': params, 'id': next(self._id_counter)}, + auth=(self._username, self._password) if self._username else None, + headers={'Content-Type': 'application/json'}, + timeout=self.DEFAULT_TIMEOUT, + ) + if not resp.ok: + logger.warning("NZBGet %s returned HTTP %s", method, resp.status_code) + return None + data = resp.json() + if data.get('error'): + logger.warning("NZBGet %s error: %r", method, data.get('error')) + return None + return data.get('result') + except http_requests.exceptions.RequestException as e: + logger.error("NZBGet %s call failed: %s", method, e) + return None + except ValueError as e: + logger.error("NZBGet %s response not JSON: %s", method, e) + return None + + async def add_nzb( + self, + url_or_bytes: Union[str, bytes], + category: str = "soulsync", + save_path: Optional[str] = None, + ) -> Optional[str]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor( + None, self._add_nzb_sync, url_or_bytes, category, save_path + ) + + def _add_nzb_sync( + self, + url_or_bytes: Union[str, bytes], + category: str, + save_path: Optional[str], + ) -> Optional[str]: + cat = category or self._category + # NZBGet's ``append`` takes: NZBFilename, Content, Category, + # Priority, AddToTop, AddPaused, DupeKey, DupeScore, DupeMode, + # PPParameters. We pass the minimum required for an unpause-on-add. + # Content is either base64 of the raw .nzb or a URL — NZBGet + # auto-detects which based on whether it looks like a URL. + if isinstance(url_or_bytes, bytes): + content = base64.b64encode(url_or_bytes).decode('ascii') + nzb_filename = 'soulsync.nzb' + else: + content = url_or_bytes + nzb_filename = '' + params = [ + nzb_filename, # NZBFilename + content, # Content (URL or base64 NZB) + cat, # Category + 0, # Priority + False, # AddToTop + False, # AddPaused + '', # DupeKey + 0, # DupeScore + 'SCORE', # DupeMode + [], # PPParameters + ] + result = self._rpc_sync('append', params) + if isinstance(result, int) and result > 0: + return str(result) + return None + + async def get_status(self, job_id: str) -> Optional[UsenetStatus]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._get_status_sync, job_id) + + def _get_status_sync(self, job_id: str) -> Optional[UsenetStatus]: + for status in self._get_all_sync(): + if status.id == job_id: + return status + return None + + async def get_all(self) -> List[UsenetStatus]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._get_all_sync) + + def _get_all_sync(self) -> List[UsenetStatus]: + out: List[UsenetStatus] = [] + groups = self._rpc_sync('listgroups', [0]) + if isinstance(groups, list): + for group in groups: + out.append(self._parse_group(group)) + history = self._rpc_sync('history', [False]) + if isinstance(history, list): + for entry in history: + out.append(self._parse_history(entry)) + return out + + def _parse_group(self, group: dict) -> UsenetStatus: + # NZBGet reports sizes split into ``FileSizeLo`` (low 32 bits) + + # ``FileSizeHi`` (high 32 bits) for compat with old clients — + # ``FileSizeMB`` is the human-friendly aggregate. + size_mb = self._mb_value(group, 'FileSize') + remaining_mb = self._mb_value(group, 'RemainingSize') + size_bytes = int(size_mb * 1024 * 1024) if size_mb else 0 + downloaded_bytes = int((size_mb - remaining_mb) * 1024 * 1024) if size_mb and remaining_mb is not None else 0 + progress = 0.0 + if size_bytes > 0: + progress = max(0.0, min(downloaded_bytes / size_bytes, 1.0)) + # NZBGet's per-group ``DownloadRate`` field is in bytes/sec. + speed = int(group.get('DownloadRate') or 0) + return UsenetStatus( + id=str(group.get('NZBID') or ''), + name=group.get('NZBName') or '', + state=_map_state(group.get('Status') or ''), + progress=progress, + size=size_bytes, + downloaded=downloaded_bytes, + download_speed=speed, + save_path=group.get('DestDir'), + category=group.get('Category'), + ) + + def _parse_history(self, entry: dict) -> UsenetStatus: + # History entries have ``Status`` like ``SUCCESS/HEALTH``, + # ``SUCCESS/UNPACK``, ``FAILURE/PAR``, etc. + status_field = entry.get('Status') or '' + is_failed = status_field.startswith('FAILURE') + size_mb = self._mb_value(entry, 'FileSize') + size_bytes = int(size_mb * 1024 * 1024) if size_mb else 0 + return UsenetStatus( + id=str(entry.get('NZBID') or ''), + name=entry.get('Name') or entry.get('NZBName') or '', + state='failed' if is_failed else 'completed', + progress=0.0 if is_failed else 1.0, + size=size_bytes, + downloaded=size_bytes if not is_failed else 0, + download_speed=0, + save_path=entry.get('DestDir'), + category=entry.get('Category'), + error=status_field if is_failed else None, + ) + + @staticmethod + def _mb_value(entry: dict, prefix: str) -> Optional[float]: + """Read an NZBGet size field. Prefers the high+low 32-bit split + when available (most accurate); falls back to the ``MB`` + aggregate for older NZBGet versions.""" + lo = entry.get(f'{prefix}Lo') + hi = entry.get(f'{prefix}Hi') + if isinstance(lo, int) and isinstance(hi, int): + total_bytes = (hi << 32) | lo + return total_bytes / (1024 * 1024) + mb = entry.get(f'{prefix}MB') + if isinstance(mb, (int, float)): + return float(mb) + return None + + async def remove(self, job_id: str, delete_files: bool = False) -> bool: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._remove_sync, job_id, delete_files) + + def _remove_sync(self, job_id: str, delete_files: bool) -> bool: + # editqueue commands take a list of NZBIDs. ``GroupFinalDelete`` + # both removes and deletes downloaded data; ``GroupDelete`` just + # removes the queue entry. + try: + id_int = int(job_id) + except (TypeError, ValueError): + return False + command = 'GroupFinalDelete' if delete_files else 'GroupDelete' + # editqueue(Command, Offset, EditText, IDs) + result = self._rpc_sync('editqueue', [command, 0, '', [id_int]]) + return bool(result) + + async def pause(self, job_id: str) -> bool: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._pause_sync, job_id) + + def _pause_sync(self, job_id: str) -> bool: + try: + id_int = int(job_id) + except (TypeError, ValueError): + return False + return bool(self._rpc_sync('editqueue', ['GroupPause', 0, '', [id_int]])) + + async def resume(self, job_id: str) -> bool: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._resume_sync, job_id) + + def _resume_sync(self, job_id: str) -> bool: + try: + id_int = int(job_id) + except (TypeError, ValueError): + return False + return bool(self._rpc_sync('editqueue', ['GroupResume', 0, '', [id_int]])) diff --git a/core/usenet_clients/sabnzbd.py b/core/usenet_clients/sabnzbd.py new file mode 100644 index 00000000..0cf3428c --- /dev/null +++ b/core/usenet_clients/sabnzbd.py @@ -0,0 +1,284 @@ +"""SABnzbd adapter. + +Auth model: a single API key passed as ``?apikey=...`` on every +request. No login flow. Every endpoint is the same path ``/api`` with +a ``mode=`` query param. + +Reference: https://sabnzbd.org/wiki/configuration/4.3/api +""" + +from __future__ import annotations + +import asyncio +from typing import List, Optional, Union + +import requests as http_requests + +from config.settings import config_manager +from core.usenet_clients.base import UsenetStatus +from utils.logging_config import get_logger + +logger = get_logger("usenet.sabnzbd") + + +# SAB queue states + history states → adapter-uniform set. +# Queue: Idle, Paused, Downloading, Grabbing, Queued, Checking, +# QuickCheck, Verifying, Repairing, Fetching, Extracting, Moving, +# Running, Completed, Failed. +_SAB_QUEUE_STATE_MAP = { + "idle": "queued", + "queued": "queued", + "grabbing": "queued", + "fetching": "downloading", + "downloading": "downloading", + "paused": "paused", + "checking": "verifying", + "quickcheck": "verifying", + "verifying": "verifying", + "repairing": "repairing", + "extracting": "extracting", + "moving": "extracting", + "running": "extracting", + "completed": "completed", + "failed": "failed", +} + + +def _map_state(sab_state: str) -> str: + return _SAB_QUEUE_STATE_MAP.get((sab_state or "").lower(), "error") + + +class SABnzbdAdapter: + """SABnzbd REST API adapter (v2+).""" + + DEFAULT_TIMEOUT = 15 + + def __init__(self) -> None: + self._load_config() + + def _load_config(self) -> None: + self._url = (config_manager.get('usenet_client.url', '') or '').rstrip('/') + self._api_key = config_manager.get('usenet_client.api_key', '') or '' + self._category = config_manager.get('usenet_client.category', 'soulsync') or 'soulsync' + + def reload_settings(self) -> None: + self._load_config() + + def is_configured(self) -> bool: + return bool(self._url and self._api_key) + + async def check_connection(self) -> bool: + if not self.is_configured(): + return False + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._check_connection_sync) + + def _check_connection_sync(self) -> bool: + # ``mode=version`` is the cheapest authenticated probe SAB exposes. + data = self._call_sync('version') + return bool(data and data.get('version')) + + def _call_sync(self, mode: str, **extra) -> Optional[dict]: + if not self.is_configured(): + return None + params = { + 'mode': mode, + 'output': 'json', + 'apikey': self._api_key, + } + params.update(extra) + try: + resp = http_requests.get(f"{self._url}/api", params=params, timeout=self.DEFAULT_TIMEOUT) + if not resp.ok: + logger.warning("SABnzbd mode=%s returned HTTP %s", mode, resp.status_code) + return None + return resp.json() + except http_requests.exceptions.RequestException as e: + logger.error("SABnzbd mode=%s request failed: %s", mode, e) + return None + except ValueError as e: + logger.error("SABnzbd mode=%s response was not JSON: %s", mode, e) + return None + + def _post_sync(self, mode: str, files=None, **extra) -> Optional[dict]: + if not self.is_configured(): + return None + params = { + 'mode': mode, + 'output': 'json', + 'apikey': self._api_key, + } + params.update(extra) + try: + resp = http_requests.post(f"{self._url}/api", params=params, files=files, + timeout=self.DEFAULT_TIMEOUT) + if not resp.ok: + logger.warning("SABnzbd POST mode=%s returned HTTP %s", mode, resp.status_code) + return None + return resp.json() + except http_requests.exceptions.RequestException as e: + logger.error("SABnzbd POST mode=%s failed: %s", mode, e) + return None + except ValueError as e: + logger.error("SABnzbd POST mode=%s response was not JSON: %s", mode, e) + return None + + async def add_nzb( + self, + url_or_bytes: Union[str, bytes], + category: str = "soulsync", + save_path: Optional[str] = None, + ) -> Optional[str]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor( + None, self._add_nzb_sync, url_or_bytes, category, save_path + ) + + def _add_nzb_sync( + self, + url_or_bytes: Union[str, bytes], + category: str, + save_path: Optional[str], + ) -> Optional[str]: + cat = category or self._category + if isinstance(url_or_bytes, bytes): + files = {'name': ('soulsync.nzb', url_or_bytes, 'application/x-nzb')} + data = self._post_sync('addfile', files=files, cat=cat) + else: + data = self._call_sync('addurl', name=url_or_bytes, cat=cat) + if not data or not data.get('status'): + return None + ids = data.get('nzo_ids') or [] + return ids[0] if ids else None + + async def get_status(self, job_id: str) -> Optional[UsenetStatus]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._get_status_sync, job_id) + + def _get_status_sync(self, job_id: str) -> Optional[UsenetStatus]: + # Check active queue first; if not found, fall back to history. + for status in self._get_all_sync(): + if status.id == job_id: + return status + return None + + async def get_all(self) -> List[UsenetStatus]: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._get_all_sync) + + def _get_all_sync(self) -> List[UsenetStatus]: + out: List[UsenetStatus] = [] + # Active queue + queue = self._call_sync('queue') + if queue and isinstance(queue.get('queue'), dict): + for slot in queue['queue'].get('slots', []) or []: + out.append(self._parse_queue_slot(slot)) + # History — completed / failed jobs SAB still tracks + history = self._call_sync('history', limit=50) + if history and isinstance(history.get('history'), dict): + for slot in history['history'].get('slots', []) or []: + out.append(self._parse_history_slot(slot)) + return out + + def _parse_queue_slot(self, slot: dict) -> UsenetStatus: + try: + percentage = float(slot.get('percentage') or 0.0) + except (TypeError, ValueError): + percentage = 0.0 + progress = percentage / 100.0 + # mb / mbleft are strings of MB values in SAB's queue API. + size_mb = self._safe_float(slot.get('mb')) + left_mb = self._safe_float(slot.get('mbleft')) + size_bytes = int(size_mb * 1024 * 1024) if size_mb else 0 + downloaded_bytes = int((size_mb - left_mb) * 1024 * 1024) if size_mb and left_mb is not None else 0 + # ``timeleft`` is HH:MM:SS — convert to seconds. + eta = self._parse_timeleft(slot.get('timeleft')) + return UsenetStatus( + id=str(slot.get('nzo_id') or ''), + name=slot.get('filename') or slot.get('name') or '', + state=_map_state(slot.get('status') or ''), + progress=max(0.0, min(progress, 1.0)), + size=size_bytes, + downloaded=max(0, downloaded_bytes), + download_speed=0, # queue endpoint doesn't include per-slot speed + eta=eta, + category=slot.get('cat'), + ) + + def _parse_history_slot(self, slot: dict) -> UsenetStatus: + # History entries are post-download — progress is 1.0 unless failed. + sab_state = (slot.get('status') or '').lower() + is_failed = sab_state == 'failed' + return UsenetStatus( + id=str(slot.get('nzo_id') or ''), + name=slot.get('name') or '', + state='failed' if is_failed else 'completed', + progress=0.0 if is_failed else 1.0, + size=int(slot.get('bytes') or 0), + downloaded=int(slot.get('bytes') or 0) if not is_failed else 0, + download_speed=0, + save_path=slot.get('storage') or slot.get('path'), + category=slot.get('category'), + error=slot.get('fail_message') if is_failed else None, + ) + + @staticmethod + def _safe_float(value) -> Optional[float]: + if value is None or value == '': + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + @staticmethod + def _parse_timeleft(value) -> Optional[int]: + if not value or not isinstance(value, str): + return None + parts = value.split(':') + try: + if len(parts) == 3: + h, m, s = parts + return int(h) * 3600 + int(m) * 60 + int(s) + if len(parts) == 2: + m, s = parts + return int(m) * 60 + int(s) + except ValueError: + return None + return None + + async def remove(self, job_id: str, delete_files: bool = False) -> bool: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._remove_sync, job_id, delete_files) + + def _remove_sync(self, job_id: str, delete_files: bool) -> bool: + # SAB deletes from queue or history depending on where the job is. + # We try queue first; if SAB reports no-op, fall through to history. + params = {'name': 'delete', 'value': job_id} + if delete_files: + params['del_files'] = 1 + data = self._call_sync('queue', **params) + if data and data.get('status'): + return True + # History delete + history_params = {'name': 'delete', 'value': job_id} + if delete_files: + history_params['del_files'] = 1 + data = self._call_sync('history', **history_params) + return bool(data and data.get('status')) + + async def pause(self, job_id: str) -> bool: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._pause_sync, job_id) + + def _pause_sync(self, job_id: str) -> bool: + data = self._call_sync('queue', name='pause', value=job_id) + return bool(data and data.get('status')) + + async def resume(self, job_id: str) -> bool: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self._resume_sync, job_id) + + def _resume_sync(self, job_id: str) -> bool: + data = self._call_sync('queue', name='resume', value=job_id) + return bool(data and data.get('status')) diff --git a/tests/downloads/test_downloads_validation.py b/tests/downloads/test_downloads_validation.py index 62798523..ddb5346a 100644 --- a/tests/downloads/test_downloads_validation.py +++ b/tests/downloads/test_downloads_validation.py @@ -11,19 +11,32 @@ from __future__ import annotations from dataclasses import dataclass from typing import Optional -from core.downloads.validation import filter_soundcloud_previews +from core.downloads import validation +from core.downloads.validation import filter_soundcloud_previews, get_valid_candidates @dataclass class _Track: duration_ms: int + name: str = 'Song' + artists: tuple[str, ...] = ('Artist',) @dataclass class _Candidate: username: str duration: Optional[int] # milliseconds - title: str = '' + title: str = 'Song' + artist: str = 'Artist' + filename: str = 'candidate' + + +class _MatchingEngine: + def score_track_match(self, **kwargs): + return 0.99, 'core_title_match' + + def normalize_string(self, text): + return (text or '').lower() def test_drops_soundcloud_30s_preview_when_expected_long(): @@ -90,3 +103,23 @@ def test_keeps_soundcloud_candidate_at_threshold(): # 110s passes both checks: > 35s AND > 100s (half of 200s) cand = _Candidate(username='soundcloud', duration=110_000) assert filter_soundcloud_previews([cand], expected) == [cand] + + +def test_rejects_tidal_candidate_that_would_fail_integrity_duration(monkeypatch): + """Structured sources should not download candidates that post-processing + will immediately quarantine for the same duration mismatch.""" + monkeypatch.setattr(validation, 'matching_engine', _MatchingEngine()) + expected = _Track(duration_ms=338_000) + wrong_tidal = _Candidate(username='tidal', duration=30_000) + + assert get_valid_candidates([wrong_tidal], expected, 'Artist Song') == [] + + +def test_keeps_tidal_candidate_inside_integrity_duration_tolerance(monkeypatch): + monkeypatch.setattr(validation, 'matching_engine', _MatchingEngine()) + expected = _Track(duration_ms=338_000) + tidal = _Candidate(username='tidal', duration=340_000) + + result = get_valid_candidates([tidal], expected, 'Artist Song') + + assert result == [tidal] diff --git a/tests/downloads/test_tidal_pinning.py b/tests/downloads/test_tidal_pinning.py index 74327f37..bb4db8e9 100644 --- a/tests/downloads/test_tidal_pinning.py +++ b/tests/downloads/test_tidal_pinning.py @@ -61,6 +61,26 @@ def test_is_authenticated_false_when_session_check_login_raises(tidal_client_wit assert client.is_authenticated() is False +def test_check_device_auth_returns_helpful_message_for_non_json_tidal_response(tidal_client_with_engine): + client, _ = tidal_client_with_engine + + class FakeFuture: + def running(self): + return False + + def result(self, timeout=0): + raise ValueError("Expecting value: line 1 column 1 (char 0)") + + client._device_auth_future = FakeFuture() + client._device_auth_link = {'verification_uri': 'https://link.tidal.com', 'user_code': 'ABCD'} + + result = client.check_device_auth() + + assert result['status'] == 'error' + assert 'invalid auth response' in result['message'] + assert 'Expecting value' not in result['message'] + + # --------------------------------------------------------------------------- # download() — filename parsing + id contract # --------------------------------------------------------------------------- diff --git a/tests/metadata/test_image_url_normalization.py b/tests/metadata/test_image_url_normalization.py index 2832ba4c..01ba4933 100644 --- a/tests/metadata/test_image_url_normalization.py +++ b/tests/metadata/test_image_url_normalization.py @@ -10,6 +10,7 @@ from urllib.parse import quote "thumb_url", [ "/api/image-proxy?url=https%3A%2F%2Fexample.com%2Fcover.jpg", + "/api/image-cache/" + ("a" * 64), "http://host.docker.internal:4533/api/image-proxy?u=ketiska&t=abc&s=def&v=1.16.1&c=SoulSync&f=json", ], ) @@ -20,10 +21,11 @@ def test_normalize_image_url_leaves_existing_image_proxy_urls_alone(thumb_url): assert normalize_image_url(thumb_url) == thumb_url -def test_normalize_image_url_proxies_internal_http_urls(monkeypatch): - """Raw internal image URLs should still be routed through SoulSync's proxy.""" +def test_normalize_image_url_registers_internal_http_urls_with_image_cache(monkeypatch): + """Raw internal image URLs should be routed through SoulSync's hashed cache URL.""" from core.metadata import normalize_image_url from core.metadata import artwork + import core.image_cache as image_cache class _FakeConfig: def get_active_media_server(self): @@ -39,6 +41,33 @@ def test_normalize_image_url_proxies_internal_http_urls(monkeypatch): return {} monkeypatch.setattr(artwork, "get_config_manager", lambda: _FakeConfig()) + monkeypatch.setattr(image_cache, "cached_image_url", lambda u: "/api/image-cache/" + ("b" * 64)) + + url = "http://localhost:4533/cover.jpg" + assert normalize_image_url(url) == "/api/image-cache/" + ("b" * 64) + + +def test_normalize_image_url_falls_back_to_proxy_when_cache_registration_fails(monkeypatch): + """If cache registration breaks, internal URLs keep the old image-proxy behavior.""" + from core.metadata import normalize_image_url + from core.metadata import artwork + import core.image_cache as image_cache + + class _FakeConfig: + def get_active_media_server(self): + return "spotify" + + def get_plex_config(self): + return {} + + def get_jellyfin_config(self): + return {} + + def get_navidrome_config(self): + return {} + + monkeypatch.setattr(artwork, "get_config_manager", lambda: _FakeConfig()) + monkeypatch.setattr(image_cache, "cached_image_url", lambda u: (_ for _ in ()).throw(RuntimeError("boom"))) url = "http://localhost:4533/cover.jpg" assert normalize_image_url(url) == f"/api/image-proxy?url={quote(url, safe='')}" diff --git a/tests/test_image_cache.py b/tests/test_image_cache.py new file mode 100644 index 00000000..f4556757 --- /dev/null +++ b/tests/test_image_cache.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +from core.image_cache import ImageCache + + +class FakeResponse: + def __init__(self, body: bytes, *, status_code: int = 200, content_type: str = "image/jpeg"): + self.body = body + self.status_code = status_code + self.headers = { + "Content-Type": content_type, + "Content-Length": str(len(body)), + } + self.closed = False + + def iter_content(self, chunk_size=65536): + yield self.body + + def close(self): + self.closed = True + + +def test_cache_url_for_registers_hashed_browser_path(tmp_path): + cache = ImageCache(tmp_path) + url = "https://images.example.test/cover.jpg?token=secret" + + cached_url = cache.cache_url_for(url) + + assert cached_url == f"/api/image-cache/{ImageCache.key_for_url(url)}" + assert "secret" not in cached_url + + +def test_get_url_fetches_once_then_serves_cached_file(tmp_path): + calls = [] + + def fetcher(url, **kwargs): + calls.append((url, kwargs)) + return FakeResponse(b"fake-jpeg-bytes") + + cache = ImageCache(tmp_path, fetcher=fetcher) + url = "https://images.example.test/cover.jpg" + + first = cache.get_url(url) + second = cache.get_url(url) + + assert first.status == "miss" + assert second.status == "hit" + assert first.path == second.path + assert first.path.read_bytes() == b"fake-jpeg-bytes" + assert first.mime_type == "image/jpeg" + assert len(calls) == 1 + + +def test_get_url_rejects_non_image_responses(tmp_path): + cache = ImageCache( + tmp_path, + fetcher=lambda url, **kwargs: FakeResponse(b"", content_type="text/html"), + ) + + try: + cache.get_url("https://images.example.test/not-image") + except Exception as exc: + assert "not an image" in str(exc) + else: + raise AssertionError("Expected non-image response to fail") + diff --git a/tests/test_manual_pick_no_auto_retry.py b/tests/test_manual_pick_no_auto_retry.py index c5b23cfe..efa88359 100644 --- a/tests/test_manual_pick_no_auto_retry.py +++ b/tests/test_manual_pick_no_auto_retry.py @@ -93,3 +93,75 @@ def test_manual_pick_skips_retry_on_errored_state(fake_monitor): assert task['status'] == 'downloading' +def test_monitor_waits_for_post_processing_before_batch_success(monkeypatch): + """Engine completion is not the same as a successful import. + + The monitor should start post-processing when slskd reports a completed + transfer, but the post-processing worker must be the only code path that + reports final success/failure to the batch lifecycle. + """ + monkeypatch.setattr(dm, '_make_context_key', lambda u, f: f"{u}::{f}") + monkeypatch.setattr(dm.WebUIDownloadMonitor, '_validate_worker_counts', lambda self: None) + + submitted = [] + completions = [] + + class FakeExecutor: + def submit(self, func, task_id, batch_id): + submitted.append((func, task_id, batch_id)) + + def fake_post_processing_worker(task_id, batch_id): + return None + + monkeypatch.setattr(dm, 'missing_download_executor', FakeExecutor()) + monkeypatch.setattr(dm, '_run_post_processing_worker', fake_post_processing_worker) + monkeypatch.setattr( + dm, + '_on_download_completed', + lambda batch_id, task_id, success: completions.append((batch_id, task_id, success)), + ) + + with dm.tasks_lock: + previous_tasks = dict(dm.download_tasks) + previous_batches = dict(dm.download_batches) + dm.download_tasks.clear() + dm.download_batches.clear() + dm.download_tasks['task-1'] = { + 'track_info': {'name': 'Test Track'}, + 'username': 'Pinasound', + 'filename': r'@@tmllb\Music\Album\01. Track.flac', + 'status': 'downloading', + 'download_id': 'download-1', + 'status_change_time': time.time(), + } + dm.download_batches['batch-1'] = {'queue': ['task-1']} + + try: + monitor = dm.WebUIDownloadMonitor() + monitor.monitoring = True + monitor.monitored_batches.add('batch-1') + monkeypatch.setattr( + monitor, + '_get_live_transfers', + lambda: { + r'Pinasound::@@tmllb\Music\Album\01. Track.flac': { + 'state': 'Completed, Succeeded', + 'size': 100, + 'bytesTransferred': 100, + } + }, + ) + + monitor._check_all_downloads() + + assert submitted == [(fake_post_processing_worker, 'task-1', 'batch-1')] + assert completions == [] + assert dm.download_tasks['task-1']['status'] == 'post_processing' + finally: + with dm.tasks_lock: + dm.download_tasks.clear() + dm.download_tasks.update(previous_tasks) + dm.download_batches.clear() + dm.download_batches.update(previous_batches) + + diff --git a/tests/test_prowlarr_client.py b/tests/test_prowlarr_client.py new file mode 100644 index 00000000..eb3fe6c0 --- /dev/null +++ b/tests/test_prowlarr_client.py @@ -0,0 +1,223 @@ +"""Tests for ``core/prowlarr_client.py``. + +Pins the parse + dispatch behavior so a future Prowlarr API tweak +that drops a field doesn't silently lose data, and the search +endpoint keeps building the repeated-key query Prowlarr expects. +""" + +from __future__ import annotations + +import asyncio +from unittest.mock import patch, MagicMock + +import pytest + +from core.prowlarr_client import ( + DEFAULT_MUSIC_CATEGORIES, + ProwlarrClient, + ProwlarrIndexer, + ProwlarrSearchResult, +) + + +def _run(coro): + return asyncio.new_event_loop().run_until_complete(coro) + + +def _client_with_config(url="http://prowlarr:9696", api_key="secret"): + """Build a client whose ``_load_config`` already ran with the + given URL + key, sidestepping the real config_manager.""" + client = ProwlarrClient.__new__(ProwlarrClient) + client._url = url.rstrip('/') + client._api_key = api_key + return client + + +# --------------------------------------------------------------------------- +# Pure parsers +# --------------------------------------------------------------------------- + + +def test_parse_indexer_extracts_core_fields() -> None: + client = _client_with_config() + entry = { + 'id': 7, + 'name': 'Public Tracker', + 'protocol': 'torrent', + 'enable': True, + 'privacy': 'public', + 'capabilities': { + 'categories': [ + {'id': 3000, 'name': 'Audio'}, + {'id': 3040, 'name': 'Audio/Lossless'}, + ], + }, + } + indexer = client._parse_indexer(entry) + assert indexer == ProwlarrIndexer( + id=7, + name='Public Tracker', + protocol='torrent', + enable=True, + privacy='public', + categories=[3000, 3040], + capabilities=entry['capabilities'], + ) + + +def test_parse_indexer_tolerates_missing_capabilities() -> None: + """Some indexers (the ones in error state) come back with no + ``capabilities`` block — must not crash.""" + client = _client_with_config() + indexer = client._parse_indexer({'id': 1, 'name': 'X', 'protocol': 'usenet'}) + assert indexer.id == 1 + assert indexer.protocol == 'usenet' + assert indexer.categories == [] + + +def test_parse_result_extracts_torrent_fields() -> None: + client = _client_with_config() + entry = { + 'guid': 'guid-1', + 'title': 'Some Album FLAC', + 'indexerId': 3, + 'indexer': 'Tracker', + 'protocol': 'torrent', + 'downloadUrl': 'https://example.com/x.torrent', + 'magnetUrl': 'magnet:?xt=urn:btih:abc', + 'infoUrl': 'https://example.com/details/1', + 'size': 524288000, + 'seeders': 12, + 'leechers': 3, + 'grabs': 100, + 'publishDate': '2026-05-10T00:00:00Z', + 'categories': [{'id': 3040, 'name': 'Audio/Lossless'}], + } + result = client._parse_result(entry) + assert result.title == 'Some Album FLAC' + assert result.indexer_id == 3 + assert result.download_url == 'https://example.com/x.torrent' + assert result.magnet_uri == 'magnet:?xt=urn:btih:abc' + assert result.size == 524288000 + assert result.seeders == 12 + assert result.categories == [3040] + + +def test_parse_result_accepts_int_categories() -> None: + """Some indexers return categories as bare ints instead of + ``{id, name}`` dicts. Both forms must work.""" + client = _client_with_config() + result = client._parse_result({'title': 'X', 'categories': [3000, 3010]}) + assert result.categories == [3000, 3010] + + +def test_parse_result_skips_garbage_category_entries() -> None: + client = _client_with_config() + result = client._parse_result({'title': 'X', 'categories': [{'name': 'no-id'}, 'string', None]}) + assert result.categories == [] + + +# --------------------------------------------------------------------------- +# Configured-state predicates +# --------------------------------------------------------------------------- + + +def test_is_configured_requires_both_fields() -> None: + assert _client_with_config('http://x', '').is_configured() is False + assert _client_with_config('', 'key').is_configured() is False + assert _client_with_config('http://x', 'key').is_configured() is True + + +def test_check_connection_returns_false_when_not_configured() -> None: + client = _client_with_config('', '') + assert _run(client.check_connection()) is False + + +# --------------------------------------------------------------------------- +# HTTP plumbing +# --------------------------------------------------------------------------- + + +def _mock_response(status_code: int, json_body): + resp = MagicMock() + resp.ok = 200 <= status_code < 400 + resp.status_code = status_code + resp.json.return_value = json_body + return resp + + +def test_search_passes_repeated_categories_and_indexer_ids() -> None: + """Prowlarr's search endpoint expects repeated query keys — + ``categories=3000&categories=3010&indexerIds=1``. ``requests`` + serializes a list of tuples into that exact form, so we assert + the params are passed as a list-of-tuples (not a dict).""" + client = _client_with_config() + captured_params = {} + + def fake_get(url, headers=None, params=None, timeout=None): + captured_params['url'] = url + captured_params['params'] = params + return _mock_response(200, []) + + with patch('core.prowlarr_client.http_requests.get', side_effect=fake_get): + _run(client.search('the query', categories=[3000, 3010], indexer_ids=[1, 5])) + + assert captured_params['url'] == 'http://prowlarr:9696/api/v1/search' + params = captured_params['params'] + # Convert to a frozenset of pairs for order-independent comparison + pair_set = set(params) + assert ('query', 'the query') in pair_set + assert ('type', 'search') in pair_set + assert ('categories', 3000) in pair_set + assert ('categories', 3010) in pair_set + assert ('indexerIds', 1) in pair_set + assert ('indexerIds', 5) in pair_set + + +def test_search_returns_empty_on_blank_query() -> None: + client = _client_with_config() + # No HTTP mock — call must short-circuit without touching the network. + results = _run(client.search('')) + assert results == [] + results = _run(client.search(' ')) + assert results == [] + + +def test_search_parses_response_list() -> None: + client = _client_with_config() + with patch('core.prowlarr_client.http_requests.get', + return_value=_mock_response(200, [ + {'guid': 'a', 'title': 'Album A', 'protocol': 'torrent'}, + {'guid': 'b', 'title': 'Album B', 'protocol': 'usenet'}, + ])): + results = _run(client.search('q')) + assert [r.title for r in results] == ['Album A', 'Album B'] + assert [r.protocol for r in results] == ['torrent', 'usenet'] + + +def test_check_connection_hits_system_status() -> None: + client = _client_with_config() + with patch('core.prowlarr_client.http_requests.get', + return_value=_mock_response(200, {'version': '1.13.0'})) as mock_get: + ok = _run(client.check_connection()) + assert ok is True + called_url = mock_get.call_args.args[0] + assert called_url == 'http://prowlarr:9696/api/v1/system/status' + assert mock_get.call_args.kwargs['headers']['X-Api-Key'] == 'secret' + + +def test_check_connection_returns_false_on_http_error() -> None: + client = _client_with_config() + with patch('core.prowlarr_client.http_requests.get', + return_value=_mock_response(401, {'error': 'unauthorized'})): + ok = _run(client.check_connection()) + assert ok is False + + +def test_default_music_categories_match_newznab_tree() -> None: + """The Newznab Music category IDs are a stable convention across + Prowlarr / Jackett / every indexer. Pin the defaults so a typo + here doesn't silently broaden / narrow what SoulSync queries.""" + assert 3000 in DEFAULT_MUSIC_CATEGORIES # Audio (parent) + assert 3010 in DEFAULT_MUSIC_CATEGORIES # MP3 + assert 3040 in DEFAULT_MUSIC_CATEGORIES # Lossless diff --git a/tests/test_torrent_client_adapters.py b/tests/test_torrent_client_adapters.py new file mode 100644 index 00000000..0f813b69 --- /dev/null +++ b/tests/test_torrent_client_adapters.py @@ -0,0 +1,339 @@ +"""Tests for the three torrent client adapters. + +Pins state-mapping behavior (each client has a different native state +vocabulary that must collapse onto the adapter-uniform set) and basic +HTTP / RPC plumbing so a future protocol-spec drift fails CI instead +of silently breaking downloads. +""" + +from __future__ import annotations + +import asyncio +from unittest.mock import MagicMock, patch + +import pytest + +from core.torrent_clients import adapter_for_type, get_active_adapter +from core.torrent_clients.base import TorrentClientAdapter, TorrentStatus +from core.torrent_clients.deluge import DelugeAdapter, _map_state as deluge_map +from core.torrent_clients.qbittorrent import QBittorrentAdapter, _map_state as qbit_map +from core.torrent_clients.transmission import TransmissionAdapter, _map_state as trans_map + + +def _run(coro): + return asyncio.new_event_loop().run_until_complete(coro) + + +def _mock_response(status_code: int, json_body=None, text=None, headers=None): + resp = MagicMock() + resp.ok = 200 <= status_code < 400 + resp.status_code = status_code + resp.headers = headers or {} + if json_body is not None: + resp.json.return_value = json_body + resp.text = text or '' + return resp + + +# --------------------------------------------------------------------------- +# Factory +# --------------------------------------------------------------------------- + + +def test_adapter_for_type_returns_concrete_classes() -> None: + assert isinstance(adapter_for_type('qbittorrent'), QBittorrentAdapter) + assert isinstance(adapter_for_type('transmission'), TransmissionAdapter) + assert isinstance(adapter_for_type('deluge'), DelugeAdapter) + + +def test_adapter_for_type_returns_none_for_unknown() -> None: + assert adapter_for_type('utorrent') is None + assert adapter_for_type('') is None + + +def test_adapters_conform_to_protocol() -> None: + """``isinstance`` checks the runtime_checkable Protocol — catches + adapters that lose a required method during refactors.""" + for adapter in (QBittorrentAdapter(), TransmissionAdapter(), DelugeAdapter()): + assert isinstance(adapter, TorrentClientAdapter) + + +# --------------------------------------------------------------------------- +# State mapping +# --------------------------------------------------------------------------- + + +def test_qbittorrent_state_mapping() -> None: + assert qbit_map('downloading') == 'downloading' + assert qbit_map('forcedDL') == 'downloading' + assert qbit_map('stalledDL') == 'stalled' + assert qbit_map('uploading') == 'seeding' + assert qbit_map('pausedUP') == 'completed' + assert qbit_map('pausedDL') == 'paused' + assert qbit_map('error') == 'error' + assert qbit_map('missingFiles') == 'error' + # Unknown native value → error rather than swallowing silently. + assert qbit_map('not-a-real-state') == 'error' + + +def test_transmission_state_mapping() -> None: + assert trans_map(4, 0.5) == 'downloading' + assert trans_map(6, 1.0) == 'seeding' + # Status 0 is the ambiguous one: paused vs completed-but-not-seeding. + assert trans_map(0, 0.3) == 'paused' + assert trans_map(0, 1.0) == 'completed' + assert trans_map(2, 0.0) == 'queued' # checking files + # Unknown numeric code → error. + assert trans_map(99, 0.0) == 'error' + + +def test_deluge_state_mapping() -> None: + assert deluge_map('Downloading', 0.5) == 'downloading' + assert deluge_map('Seeding', 1.0) == 'seeding' + assert deluge_map('Paused', 0.4) == 'paused' + # Deluge reports 'Paused' for completed-not-seeding too. + assert deluge_map('Paused', 1.0) == 'completed' + assert deluge_map('Error', 0.0) == 'error' + assert deluge_map('', 0.0) == 'error' + + +# --------------------------------------------------------------------------- +# qBittorrent adapter +# --------------------------------------------------------------------------- + + +def _qbit_with_config(url='http://qbit:8080', username='admin', password='x'): + adapter = QBittorrentAdapter.__new__(QBittorrentAdapter) + import threading + adapter._session = None + adapter._session_lock = threading.Lock() + adapter._url = url.rstrip('/') + adapter._username = username + adapter._password = password + adapter._category = 'soulsync' + adapter._save_path = '' + return adapter + + +def test_qbit_is_configured_requires_only_url() -> None: + # qBittorrent allows no-auth LAN setups — URL is enough. + assert _qbit_with_config('http://x', '', '').is_configured() is True + assert _qbit_with_config('', 'u', 'p').is_configured() is False + + +def test_qbit_login_sends_referer_for_csrf() -> None: + """qBittorrent rejects login attempts without a Referer matching + its host — pin the header to catch regressions.""" + adapter = _qbit_with_config() + fake_session = MagicMock() + fake_session.post.return_value = _mock_response(200, text='Ok.') + fake_session.post.return_value.text = 'Ok.' + with patch('core.torrent_clients.qbittorrent.http_requests.Session', + return_value=fake_session): + sess = adapter._ensure_session_sync() + assert sess is not None + args, kwargs = fake_session.post.call_args + assert args[0].endswith('/api/v2/auth/login') + assert kwargs['headers']['Referer'] == 'http://qbit:8080' + assert kwargs['data'] == {'username': 'admin', 'password': 'x'} + + +def test_qbit_login_failure_returns_none() -> None: + adapter = _qbit_with_config() + fake_session = MagicMock() + bad_resp = _mock_response(200, text='Fails.') + bad_resp.text = 'Fails.' + fake_session.post.return_value = bad_resp + with patch('core.torrent_clients.qbittorrent.http_requests.Session', + return_value=fake_session): + sess = adapter._ensure_session_sync() + assert sess is None + + +def test_qbit_parse_status_normalises_native_fields() -> None: + adapter = _qbit_with_config() + status = adapter._parse_status({ + 'hash': 'abc123', 'name': 'Album', + 'state': 'downloading', 'progress': 0.5, + 'size': 1024, 'downloaded': 512, + 'dlspeed': 200, 'upspeed': 50, + 'num_seeds': 4, 'num_leechs': 1, + 'eta': 60, 'save_path': '/data/torrents', + }) + assert status == TorrentStatus( + id='abc123', name='Album', state='downloading', + progress=0.5, size=1024, downloaded=512, + download_speed=200, upload_speed=50, seeders=4, peers=1, + eta=60, save_path='/data/torrents', + ) + + +def test_qbit_parse_status_zeros_eta_when_unknown() -> None: + adapter = _qbit_with_config() + # qBittorrent uses 8640000 for "unknown" but the adapter just + # treats anything <= 0 as unknown; pin that 0 maps to None. + status = adapter._parse_status({ + 'hash': 'x', 'name': 'X', 'state': 'stalledDL', + 'progress': 0.0, 'size': 100, 'downloaded': 0, + 'dlspeed': 0, 'upspeed': 0, 'eta': 0, + }) + assert status.eta is None + + +# --------------------------------------------------------------------------- +# Transmission adapter +# --------------------------------------------------------------------------- + + +def _trans_with_config(url='http://trans:9091/transmission/rpc'): + adapter = TransmissionAdapter.__new__(TransmissionAdapter) + import threading + adapter._session_id = None + adapter._session_id_lock = threading.Lock() + adapter._url = url + adapter._username = '' + adapter._password = '' + adapter._category = 'soulsync' + adapter._save_path = '' + return adapter + + +def test_transmission_normalises_bare_host_to_rpc_path() -> None: + """Users sometimes paste ``http://host:9091``; the adapter must + append ``/transmission/rpc`` so the request hits the right + endpoint.""" + adapter = TransmissionAdapter.__new__(TransmissionAdapter) + with patch('core.torrent_clients.transmission.config_manager') as cm: + cm.get.side_effect = lambda key, default='': { + 'torrent_client.url': 'http://host:9091', + 'torrent_client.username': '', + 'torrent_client.password': '', + 'torrent_client.category': 'soulsync', + 'torrent_client.save_path': '', + }.get(key, default) + import threading + adapter._session_id = None + adapter._session_id_lock = threading.Lock() + adapter._load_config() + assert adapter._url == 'http://host:9091/transmission/rpc' + + +def test_transmission_session_id_renegotiation() -> None: + """Transmission rejects the first call with 409 and a fresh + ``X-Transmission-Session-Id`` header; the adapter must store it + and retry the same call exactly once.""" + adapter = _trans_with_config() + first = _mock_response(409, headers={'X-Transmission-Session-Id': 'sid-2'}) + second = _mock_response(200, json_body={'result': 'success', 'arguments': {'session-id': 1}}) + with patch('core.torrent_clients.transmission.http_requests.post', + side_effect=[first, second]) as mock_post: + result = adapter._rpc('session-get', {}) + assert result == {'session-id': 1} + assert mock_post.call_count == 2 + # Second call carried the new session id. + second_call_kwargs = mock_post.call_args_list[1].kwargs + assert second_call_kwargs['headers']['X-Transmission-Session-Id'] == 'sid-2' + + +def test_transmission_rpc_returns_none_on_failure_result() -> None: + adapter = _trans_with_config() + with patch('core.torrent_clients.transmission.http_requests.post', + return_value=_mock_response(200, json_body={'result': 'unknown method'})): + assert adapter._rpc('bogus', {}) is None + + +def test_transmission_add_torrent_handles_duplicate() -> None: + """torrent-add returns either ``torrent-added`` (new) or + ``torrent-duplicate`` (already-there) — both must surface the hash.""" + adapter = _trans_with_config() + with patch.object(adapter, '_rpc', return_value={'torrent-duplicate': {'hashString': 'dup'}}): + hash_id = adapter._add_torrent_sync('magnet:?xt=urn:btih:abc', 'cat', None) + assert hash_id == 'dup' + + +def test_transmission_parse_status() -> None: + adapter = _trans_with_config() + status = adapter._parse_status({ + 'hashString': 'h', 'name': 'X', 'status': 4, 'percentDone': 0.42, + 'totalSize': 100, 'downloadedEver': 42, + 'rateDownload': 10, 'rateUpload': 5, + 'peersSendingToUs': 2, 'peersGettingFromUs': 0, + 'eta': 300, 'downloadDir': '/dl', 'errorString': '', + }) + assert status.id == 'h' + assert status.state == 'downloading' + assert status.progress == 0.42 + assert status.eta == 300 + + +def test_transmission_parse_status_negative_eta_is_none() -> None: + """Transmission reports -1 / -2 for 'unknown' ETA — must normalise to None.""" + adapter = _trans_with_config() + status = adapter._parse_status({ + 'hashString': 'h', 'name': 'X', 'status': 4, 'percentDone': 0.0, + 'totalSize': 100, 'downloadedEver': 0, + 'rateDownload': 0, 'rateUpload': 0, + 'peersSendingToUs': 0, 'peersGettingFromUs': 0, + 'eta': -1, 'downloadDir': '/dl', + }) + assert status.eta is None + + +# --------------------------------------------------------------------------- +# Deluge adapter +# --------------------------------------------------------------------------- + + +def _deluge_with_config(url='http://deluge:8112', password='delugepass'): + adapter = DelugeAdapter.__new__(DelugeAdapter) + import threading + from itertools import count + adapter._session = None + adapter._session_lock = threading.Lock() + adapter._id_counter = count(1) + adapter._url = url.rstrip('/') + adapter._password = password + adapter._category = 'soulsync' + adapter._save_path = '' + return adapter + + +def test_deluge_is_configured_requires_password() -> None: + assert _deluge_with_config('http://x', '').is_configured() is False + assert _deluge_with_config('http://x', 'pw').is_configured() is True + + +def test_deluge_add_torrent_uses_magnet_method() -> None: + adapter = _deluge_with_config() + with patch.object(adapter, '_ensure_session_sync', return_value=MagicMock()), \ + patch.object(adapter, '_rpc_sync', return_value='hash123') as mock_rpc: + hash_id = adapter._add_torrent_sync('magnet:?xt=urn:btih:abc', 'cat', None) + assert hash_id == 'hash123' + # First call was core.add_torrent_magnet, not the URL variant. + first_method = mock_rpc.call_args_list[0].args[0] + assert first_method == 'core.add_torrent_magnet' + + +def test_deluge_add_torrent_uses_url_method_for_http() -> None: + adapter = _deluge_with_config() + with patch.object(adapter, '_ensure_session_sync', return_value=MagicMock()), \ + patch.object(adapter, '_rpc_sync', return_value='hash456') as mock_rpc: + hash_id = adapter._add_torrent_sync('https://example.com/x.torrent', 'cat', None) + assert hash_id == 'hash456' + first_method = mock_rpc.call_args_list[0].args[0] + assert first_method == 'core.add_torrent_url' + + +def test_deluge_parse_status_normalises_percent_progress() -> None: + """Deluge reports progress as 0-100 (not 0-1) — adapter must + normalise.""" + adapter = _deluge_with_config() + status = adapter._parse_status({ + 'hash': 'abc', 'name': 'X', 'state': 'Downloading', + 'progress': 42.0, + 'total_size': 1000, 'total_done': 420, + 'download_payload_rate': 100, 'upload_payload_rate': 0, + 'num_seeds': 1, 'num_peers': 0, 'eta': 0, + }) + assert status.progress == pytest.approx(0.42) + assert status.state == 'downloading' diff --git a/tests/test_usenet_client_adapters.py b/tests/test_usenet_client_adapters.py new file mode 100644 index 00000000..8dfb28c6 --- /dev/null +++ b/tests/test_usenet_client_adapters.py @@ -0,0 +1,297 @@ +"""Tests for the SABnzbd and NZBGet usenet adapters. + +Pins state-mapping behavior and the queue-vs-history merge logic so +get_all returns both active and completed jobs without losing +either bucket. +""" + +from __future__ import annotations + +import asyncio +import base64 +from unittest.mock import MagicMock, patch + +import pytest + +from core.usenet_clients import adapter_for_type +from core.usenet_clients.base import UsenetClientAdapter, UsenetStatus +from core.usenet_clients.nzbget import NZBGetAdapter, _map_state as nzbget_map +from core.usenet_clients.sabnzbd import SABnzbdAdapter, _map_state as sab_map + + +def _run(coro): + return asyncio.new_event_loop().run_until_complete(coro) + + +def _mock_response(status_code: int, json_body=None): + resp = MagicMock() + resp.ok = 200 <= status_code < 400 + resp.status_code = status_code + if json_body is not None: + resp.json.return_value = json_body + return resp + + +# --------------------------------------------------------------------------- +# Factory + protocol conformance +# --------------------------------------------------------------------------- + + +def test_adapter_for_type_returns_concrete_classes() -> None: + assert isinstance(adapter_for_type('sabnzbd'), SABnzbdAdapter) + assert isinstance(adapter_for_type('nzbget'), NZBGetAdapter) + + +def test_adapter_for_type_returns_none_for_unknown() -> None: + assert adapter_for_type('unknown') is None + + +def test_adapters_conform_to_protocol() -> None: + for adapter in (SABnzbdAdapter(), NZBGetAdapter()): + assert isinstance(adapter, UsenetClientAdapter) + + +# --------------------------------------------------------------------------- +# SABnzbd +# --------------------------------------------------------------------------- + + +def _sab_with_config(url='http://sab:8080', api_key='k'): + adapter = SABnzbdAdapter.__new__(SABnzbdAdapter) + adapter._url = url.rstrip('/') + adapter._api_key = api_key + adapter._category = 'soulsync' + return adapter + + +def test_sab_is_configured_requires_url_and_key() -> None: + assert _sab_with_config('http://x', '').is_configured() is False + assert _sab_with_config('', 'k').is_configured() is False + assert _sab_with_config('http://x', 'k').is_configured() is True + + +def test_sab_state_mapping_covers_queue_states() -> None: + assert sab_map('Downloading') == 'downloading' + assert sab_map('Verifying') == 'verifying' + assert sab_map('Repairing') == 'repairing' + assert sab_map('Extracting') == 'extracting' + assert sab_map('Paused') == 'paused' + assert sab_map('Failed') == 'failed' + # Case-insensitive — SAB sometimes returns lowercase. + assert sab_map('downloading') == 'downloading' + assert sab_map('') == 'error' + + +def test_sab_parse_timeleft_handles_hhmmss() -> None: + # SABnzbd's timeleft is always HH:MM:SS (or H:MM:SS). + assert SABnzbdAdapter._parse_timeleft('01:30:00') == 5400 + assert SABnzbdAdapter._parse_timeleft('00:05:30') == 330 + assert SABnzbdAdapter._parse_timeleft('00:30:00') == 1800 + # 2-part fallback covers MM:SS for robustness. + assert SABnzbdAdapter._parse_timeleft('05:30') == 330 + assert SABnzbdAdapter._parse_timeleft('garbage') is None + assert SABnzbdAdapter._parse_timeleft('') is None + assert SABnzbdAdapter._parse_timeleft(None) is None + + +def test_sab_parse_queue_slot_converts_mb_to_bytes() -> None: + adapter = _sab_with_config() + status = adapter._parse_queue_slot({ + 'nzo_id': 'SABnzbd_nzo_1', + 'filename': 'Album.nzb', + 'status': 'Downloading', + 'percentage': '42', + 'mb': '100', + 'mbleft': '58', + 'timeleft': '0:01:00', + 'cat': 'soulsync', + }) + assert status.id == 'SABnzbd_nzo_1' + assert status.state == 'downloading' + assert status.progress == pytest.approx(0.42) + assert status.size == 100 * 1024 * 1024 + assert status.downloaded == 42 * 1024 * 1024 + assert status.eta == 60 + + +def test_sab_parse_history_slot_marks_failures() -> None: + adapter = _sab_with_config() + failed = adapter._parse_history_slot({ + 'nzo_id': 'x', 'name': 'X', 'status': 'Failed', + 'bytes': 1024, 'fail_message': 'Damaged', + }) + assert failed.state == 'failed' + assert failed.progress == 0.0 + assert failed.error == 'Damaged' + + success = adapter._parse_history_slot({ + 'nzo_id': 'y', 'name': 'Y', 'status': 'Completed', + 'bytes': 1024, 'storage': '/done', + }) + assert success.state == 'completed' + assert success.progress == 1.0 + assert success.save_path == '/done' + + +def test_sab_add_nzb_via_url_returns_first_nzo_id() -> None: + adapter = _sab_with_config() + with patch('core.usenet_clients.sabnzbd.http_requests.get', + return_value=_mock_response(200, {'status': True, 'nzo_ids': ['SABnzbd_1']})) as mock_get: + job_id = _run(adapter.add_nzb('https://example.com/x.nzb', category='cat')) + assert job_id == 'SABnzbd_1' + params = mock_get.call_args.kwargs['params'] + assert params['mode'] == 'addurl' + assert params['apikey'] == 'k' + assert params['name'] == 'https://example.com/x.nzb' + assert params['cat'] == 'cat' + + +def test_sab_add_nzb_via_bytes_uses_addfile_multipart() -> None: + adapter = _sab_with_config() + with patch('core.usenet_clients.sabnzbd.http_requests.post', + return_value=_mock_response(200, {'status': True, 'nzo_ids': ['SABnzbd_2']})) as mock_post: + job_id = _run(adapter.add_nzb(b'', category='cat')) + assert job_id == 'SABnzbd_2' + assert mock_post.call_args.kwargs['params']['mode'] == 'addfile' + files = mock_post.call_args.kwargs['files'] + assert 'name' in files + assert files['name'][1] == b'' + + +def test_sab_get_all_merges_queue_and_history() -> None: + """SAB's queue and history are separate endpoints; the adapter + must hit both so completed jobs surface in the global list.""" + adapter = _sab_with_config() + queue_resp = _mock_response(200, {'queue': {'slots': [ + {'nzo_id': 'q1', 'filename': 'A.nzb', 'status': 'Downloading', + 'percentage': '10', 'mb': '100', 'mbleft': '90', 'timeleft': '0:01:00'}, + ]}}) + history_resp = _mock_response(200, {'history': {'slots': [ + {'nzo_id': 'h1', 'name': 'B.nzb', 'status': 'Completed', 'bytes': 1024}, + ]}}) + with patch('core.usenet_clients.sabnzbd.http_requests.get', + side_effect=[queue_resp, history_resp]): + statuses = adapter._get_all_sync() + assert [s.id for s in statuses] == ['q1', 'h1'] + assert [s.state for s in statuses] == ['downloading', 'completed'] + + +# --------------------------------------------------------------------------- +# NZBGet +# --------------------------------------------------------------------------- + + +def _nzbget_with_config(url='http://nzbget:6789', username='u', password='p'): + adapter = NZBGetAdapter.__new__(NZBGetAdapter) + from itertools import count + adapter._id_counter = count(1) + adapter._url = url.rstrip('/') + adapter._username = username + adapter._password = password + adapter._category = 'soulsync' + return adapter + + +def test_nzbget_is_configured_requires_all_three() -> None: + assert _nzbget_with_config('', 'u', 'p').is_configured() is False + assert _nzbget_with_config('http://x', '', 'p').is_configured() is False + assert _nzbget_with_config('http://x', 'u', '').is_configured() is False + assert _nzbget_with_config('http://x', 'u', 'p').is_configured() is True + + +def test_nzbget_state_mapping_covers_post_process_phases() -> None: + assert nzbget_map('DOWNLOADING') == 'downloading' + assert nzbget_map('PAUSED') == 'paused' + assert nzbget_map('LOADING_PARS') == 'verifying' + assert nzbget_map('REPAIRING') == 'repairing' + assert nzbget_map('UNPACKING') == 'extracting' + assert nzbget_map('PP_FINISHED') == 'completed' + assert nzbget_map('') == 'error' + + +def test_nzbget_mb_value_prefers_64bit_split() -> None: + """NZBGet ships size as FileSizeHi << 32 | FileSizeLo for clients + that need precision past 2 GB. Prefer that over the legacy MB + field when both are present.""" + val = NZBGetAdapter._mb_value({'FileSizeLo': 1024 * 1024, 'FileSizeHi': 0, 'FileSizeMB': 999}, 'FileSize') + assert val == 1.0 + + +def test_nzbget_mb_value_falls_back_to_mb() -> None: + val = NZBGetAdapter._mb_value({'FileSizeMB': 500}, 'FileSize') + assert val == 500.0 + + +def test_nzbget_add_nzb_url_passes_through_unchanged() -> None: + adapter = _nzbget_with_config() + captured = {} + + def fake_post(url, json=None, auth=None, headers=None, timeout=None): + captured['payload'] = json + return _mock_response(200, {'result': 42}) + + with patch('core.usenet_clients.nzbget.http_requests.post', side_effect=fake_post): + job_id = _run(adapter.add_nzb('https://x/x.nzb', category='cat')) + + assert job_id == '42' + params = captured['payload']['params'] + assert params[0] == '' # NZBFilename empty when content is a URL + assert params[1] == 'https://x/x.nzb' + assert params[2] == 'cat' + + +def test_nzbget_add_nzb_bytes_base64_encodes() -> None: + adapter = _nzbget_with_config() + captured = {} + + def fake_post(url, json=None, auth=None, headers=None, timeout=None): + captured['payload'] = json + return _mock_response(200, {'result': 7}) + + with patch('core.usenet_clients.nzbget.http_requests.post', side_effect=fake_post): + _run(adapter.add_nzb(b'', category='cat')) + + params = captured['payload']['params'] + assert params[0] == 'soulsync.nzb' + assert params[1] == base64.b64encode(b'').decode('ascii') + + +def test_nzbget_remove_uses_groupfinal_when_deleting_files() -> None: + """``GroupFinalDelete`` deletes downloaded data on disk; + ``GroupDelete`` just removes the queue entry. The adapter must + pick the right one based on the ``delete_files`` flag.""" + adapter = _nzbget_with_config() + with patch.object(adapter, '_rpc_sync', return_value=True) as mock_rpc: + adapter._remove_sync('42', delete_files=True) + adapter._remove_sync('42', delete_files=False) + cmds = [c.args[1][0] for c in mock_rpc.call_args_list] + assert cmds == ['GroupFinalDelete', 'GroupDelete'] + + +def test_nzbget_parse_group_computes_progress() -> None: + adapter = _nzbget_with_config() + status = adapter._parse_group({ + 'NZBID': 99, + 'NZBName': 'Album.nzb', + 'Status': 'DOWNLOADING', + 'FileSizeLo': 200 * 1024 * 1024, 'FileSizeHi': 0, + 'RemainingSizeLo': 100 * 1024 * 1024, 'RemainingSizeHi': 0, + 'DownloadRate': 500_000, + 'DestDir': '/incomplete', + 'Category': 'soulsync', + }) + assert status.id == '99' + assert status.state == 'downloading' + assert status.size == 200 * 1024 * 1024 + assert status.downloaded == 100 * 1024 * 1024 + assert status.progress == pytest.approx(0.5) + assert status.download_speed == 500_000 + + +def test_nzbget_remove_rejects_non_numeric_id() -> None: + """NZBGet IDs are ints; passing a string id like 'abc' must + fail fast instead of corrupting the editqueue call.""" + adapter = _nzbget_with_config() + with patch.object(adapter, '_rpc_sync') as mock_rpc: + assert adapter._remove_sync('not-a-number', delete_files=False) is False + mock_rpc.assert_not_called() diff --git a/web_server.py b/web_server.py index e87b4006..ddf3ebe4 100644 --- a/web_server.py +++ b/web_server.py @@ -40,7 +40,7 @@ logger = setup_logging(_log_level, _log_path) # App version — single source of truth for backup metadata, system-info, update check, etc. # Semver: MAJOR.MINOR.PATCH. Bump at each dev→main release. -_SOULSYNC_BASE_VERSION = "2.5.7" +_SOULSYNC_BASE_VERSION = "2.5.8" def _build_version_string(): """Append short commit hash to version when available (e.g. 2.35+abc1234).""" @@ -2753,7 +2753,7 @@ def handle_settings(): if 'active_media_server' in new_settings: config_manager.set_active_media_server(new_settings['active_media_server']) - for service in ['spotify', 'plex', 'jellyfin', 'navidrome', 'soulseek', 'download_source', 'settings', 'database', 'metadata_enhancement', 'file_organization', 'playlist_sync', 'tidal', 'tidal_download', 'qobuz', 'hifi_download', 'deezer_download', 'amazon_download', 'lidarr_download', 'listenbrainz', 'acoustid', 'lastfm', 'genius', 'import', 'lossy_copy', 'listening_stats', 'ui_appearance', 'youtube', 'content_filter', 'itunes', 'm3u_export', 'musicbrainz', 'deezer', 'audiodb', 'metadata', 'hydrabase', 'security', 'discogs', 'library', 'discover', 'wishlist', 'genre_whitelist', 'post_processing']: + for service in ['spotify', 'plex', 'jellyfin', 'navidrome', 'soulseek', 'download_source', 'settings', 'database', 'metadata_enhancement', 'file_organization', 'playlist_sync', 'tidal', 'tidal_download', 'qobuz', 'hifi_download', 'deezer_download', 'amazon_download', 'lidarr_download', 'prowlarr', 'torrent_client', 'usenet_client', 'listenbrainz', 'acoustid', 'lastfm', 'genius', 'import', 'lossy_copy', 'listening_stats', 'ui_appearance', 'youtube', 'content_filter', 'itunes', 'm3u_export', 'musicbrainz', 'deezer', 'audiodb', 'metadata', 'hydrabase', 'security', 'discogs', 'library', 'discover', 'wishlist', 'genre_whitelist', 'post_processing']: if service in new_settings: for key, value in new_settings[service].items(): config_manager.set(f'{service}.{key}', value) @@ -3043,6 +3043,36 @@ def hydrabase_send(): _hydrabase_ws = None return jsonify({"success": False, "error": str(e)}), 500 +@app.route('/api/prowlarr/indexers', methods=['GET']) +def prowlarr_indexers_endpoint(): + """List indexers Prowlarr currently exposes — name, protocol, enabled state. + + Drives the Indexers panel on Settings → Indexers & Downloaders so + the user can see what they're searching against without leaving + SoulSync. Returns ``[]`` if Prowlarr isn't configured / reachable. + """ + try: + from core.prowlarr_client import ProwlarrClient + client = ProwlarrClient() + if not client.is_configured(): + return jsonify({"success": False, "error": "Prowlarr not configured", "indexers": []}), 200 + indexers = run_async(client.get_indexers()) + items = [ + { + 'id': idx.id, + 'name': idx.name, + 'protocol': idx.protocol, + 'enable': idx.enable, + 'privacy': idx.privacy, + } + for idx in indexers + ] + return jsonify({"success": True, "indexers": items}) + except Exception as e: + logger.error(f"prowlarr indexers fetch error: {e}") + return jsonify({"success": False, "error": str(e), "indexers": []}), 500 + + @app.route('/api/settings/log-level', methods=['GET', 'POST']) @admin_only def handle_log_level(): @@ -28015,45 +28045,52 @@ def get_your_artist_info(artist_id): @app.route('/api/image-proxy', methods=['GET']) def image_proxy(): - """Proxy external images to avoid CORS issues for canvas rendering.""" + """Proxy external images to avoid CORS issues for canvas rendering. + + Kept for backwards compatibility; new normalized artwork URLs use + /api/image-cache/, but older browser sessions may still hold this + query-string form. + """ url = request.args.get('url', '') if not url or not url.startswith('http'): return '', 400 - host = urlparse(url).hostname or '' - allowed_hosts = [ - 'i.scdn.co', 'mosaic.scdn.co', # Spotify - 'lastfm.freetls.fastly.net', 'lastfm-img2.akamaized.net', # Last.fm - 'e-cdns-images.dzcdn.net', 'cdns-images.dzcdn.net', 'api.deezer.com', # Deezer - 'is1-ssl.mzstatic.com', 'is2-ssl.mzstatic.com', 'is3-ssl.mzstatic.com', - 'is4-ssl.mzstatic.com', 'is5-ssl.mzstatic.com', # iTunes/Apple - 'img.discogs.com', 'i.discogs.com', # Discogs - 'localhost', '127.0.0.1', 'host.docker.internal', # Local/Docker media servers - ] - if not any(host == h or host.endswith('.' + h) for h in allowed_hosts) and not is_internal_image_host(url): - return '', 403 try: - resp = requests.get(url, timeout=10, stream=True, headers={ - 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', - 'Referer': 'https://www.deezer.com/', - }) - if resp.status_code != 200: - return '', resp.status_code - content_type = resp.headers.get('Content-Type', 'image/jpeg') - if not content_type.startswith('image/'): - return '', 400 - return Response( - resp.content, - content_type=content_type, - headers={ - 'Cache-Control': 'public, max-age=86400', - 'Access-Control-Allow-Origin': '*', - } - ) - except Exception: + from core.image_cache import get_image_cache + + cached = get_image_cache().get_url(url) + response = send_file(cached.path, mimetype=cached.mime_type, conditional=True) + max_age = int(config_manager.get("image_cache.ttl_seconds", 2592000)) + response.headers['Cache-Control'] = f'private, max-age={max_age}' + response.headers['Access-Control-Allow-Origin'] = '*' + response.headers['X-SoulSync-Image-Cache'] = cached.status + return response + except Exception as exc: + logger.debug("image proxy failed: %s", exc) return '', 502 +@app.route('/api/image-cache/', methods=['GET']) +def serve_cached_image(cache_key): + """Serve a registered image URL from SoulSync's disk cache.""" + if not re.fullmatch(r'[a-f0-9]{64}', cache_key or ''): + return '', 404 + + try: + from core.image_cache import get_image_cache + + cached = get_image_cache().get(cache_key) + response = send_file(cached.path, mimetype=cached.mime_type, conditional=True) + max_age = int(config_manager.get("image_cache.ttl_seconds", 2592000)) + response.headers['Cache-Control'] = f'private, max-age={max_age}' + response.headers['Access-Control-Allow-Origin'] = '*' + response.headers['X-SoulSync-Image-Cache'] = cached.status + return response + except Exception as exc: + logger.debug("cached image serve failed for %s: %s", cache_key, exc) + return '', 404 + + from core.artists.map import ( _artmap_cache_invalidate, _artmap_cache_get, diff --git a/webui/index.html b/webui/index.html index 4bcfba98..f6b0cb1b 100644 --- a/webui/index.html +++ b/webui/index.html @@ -3635,6 +3635,7 @@
+ @@ -4945,6 +4946,211 @@
+ + +
+
Indexers & Downloaders
+
+ 1 Indexers find releases + → + 2 Downloader fetches them + → + 3 SoulSync imports to library +
+
+ Configure your Prowlarr instance to feed search results, and pick a torrent client and/or usenet client to handle the actual downloads. You only need to set up the protocols you actually use. +
+
+ + +
+ ▼ +

🔎 Indexers

+ Prowlarr — search aggregator + +
+
+
+
+

⬢ Prowlarr

+
+ Prowlarr manages your Usenet and torrent indexers and exposes them through one API. SoulSync uses Prowlarr to search across every indexer at once for the Torrent and Usenet download sources. +

+ Don't have it? Grab Prowlarr from prowlarr.com (or your *arr stack). You point Prowlarr at your indexers, then point SoulSync at Prowlarr. +
+
+ + +
+ Full URL to your Prowlarr instance (e.g. http://192.168.1.100:9696). +
+
+
+ + +
+ Found in Prowlarr → Settings → General → Security → API Key. +
+
+
+ + +
+ Comma-separated Prowlarr indexer IDs. Leave blank to search every enabled indexer. Restrict when you have a private tracker you want to prioritise or a noisy public one to exclude. +
+
+
+ +
+ + + +
+
+
+ +
+ Connect to Prowlarr and click Refresh Indexer List to see the indexers SoulSync can search. +
+
+
+
+
+ + + + + + + + +