diff --git a/config/settings.py b/config/settings.py index b2c6b8c4..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 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/image_cache.py b/core/image_cache.py index 9ee0be47..040c1126 100644 --- a/core/image_cache.py +++ b/core/image_cache.py @@ -198,8 +198,8 @@ class ImageCache: except Exception: try: tmp_path.unlink(missing_ok=True) - except Exception: - pass + except Exception as cleanup_exc: + logger.debug("image_cache tmp cleanup failed: %s", cleanup_exc) raise if total <= 0: 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/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/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 0fbb760c..ddf3ebe4 100644 --- a/web_server.py +++ b/web_server.py @@ -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(): 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. +
+
+
+
+
+ + + + + + + + +