f13d339584
User reported usenet album downloads getting stuck on "downloading
release" while SABnzbd reported the job as complete. Container restart
did not help; reproducible on every usenet album download.
Three independent issues all causing the same symptom — the download
modal freezes mid-flow with no error surfaced to the user:
1. SAB queue → history transition window
SAB removes a slot from its queue BEFORE adding it to the history,
and on a busy server (par2 verify, unrar, multi-file move) that
window can span several poll iterations. The poll treated a single
None status as terminal failure ("disappeared from client") and
gave up. Now the poll tolerates up to ~10s of consecutive misses
(5 polls at the default 2s interval) before declaring the job gone.
2. SAB queue states like `Pp` were unmapped
`_SAB_QUEUE_STATE_MAP` didn't cover SAB's `Pp` (post-processing
summary), `Unpacking`, `Trying`, `Deleted`, or the `Prop_paused`
/ `Prop_failed` variants. Unmapped states fell through to the
default-'error' fallback, and the poll loop only treated explicit
'failed' / 'completed' as terminal — 'error' was neither, so the
loop spun until the 6-hour timeout. Map now covers every Status
value from SAB's `sabnzbd/api.py`, and the poll treats the default-
'error' fallback as a transient miss (warn-logged, retry within
the same tolerance window) so a brand-new unmapped state can't
infinite-loop the way `Pp` did here.
3. No terminal failure emit
The poll only logged on failure / timeout / disappeared — never
called the progress callback with 'failed', so the download modal
stayed at the last 'downloading' emit forever. Plumb a 'failed'
emit through every failure exit path so the UI flips out of the
downloading state when the poll gives up.
Plus:
4. SAB direct nzo_ids lookup instead of paging all-history
`_get_status_sync` was fetching the latest 50 history entries on
every poll and iterating to find the target nzo_id. On busy
servers (many recent downloads), the target job could roll past
the 50-entry window and look like a "disappeared" job. Replaced
with a targeted `mode=queue&nzo_ids=<id>` → `mode=history&nzo_ids=<id>`
chain. Falls back to the bulk path for SAB versions that pre-date
the nzo_ids filter — the transient-miss tolerance covers any
short-lived gap there too.
Implementation:
Lifted the album-bundle poll loop out of `usenet.py` and `torrent.py`
into `core/download_plugins/album_bundle.py:poll_album_download` —
near-duplicate implementations are now a single function with deps
injected so it's testable in isolation (kettui's extract-don't-AST-parse
standard; can't unit-test a `time.sleep` loop inside a plugin method).
The lifted helper takes:
- `get_status` callable bound to job_id, so the same loop works for
usenet UsenetStatus and torrent TorrentStatus shapes
- `complete_states` set so torrent's `{'seeding', 'completed'}` and
usenet's `{'completed'}` both Just Work
- `failed_states` set so torrent's `{'error'}` is terminal while
usenet's default-'error' fallback is transient
- `transient_miss_threshold` (default 5 ≈ 10s at 2s poll)
- `sleep` / `monotonic` injectables for deterministic tests
Per-track flows in both plugins gained the same transient-miss
tolerance inline — they don't use the emit pattern (update an
`active_downloads[id]` row dict via lock instead), so reusing the
helper would have required threading a no-op emit through. Inline
fix is small enough.
Tests:
- 11 new tests in `tests/test_album_bundle.py:poll_album_download`
cover the happy path, transient-miss tolerance with recovery,
hard-failure threshold, explicit-failed surface, timeout-emit,
default-'error' transient treatment, shutdown clean exit,
torrent's `seeding`-counts-as-complete, save_path captured across
iterations, and adapter-exception treated as transient miss.
- 521 download-suite tests pass (33 in test_album_bundle, others
pin existing torrent + usenet contracts).
- Ruff clean.
Closes #706.
480 lines
18 KiB
Python
480 lines
18 KiB
Python
"""UsenetDownloadPlugin — composes Prowlarr search + usenet client
|
|
adapter + archive_pipeline into a uniform download source.
|
|
|
|
Mirrors ``TorrentDownloadPlugin`` in shape and lifecycle (see that
|
|
module's docstring for the full pipeline rationale). Differences:
|
|
|
|
- Search filters Prowlarr results to ``protocol='usenet'``.
|
|
- ``add_nzb`` replaces ``add_torrent``; for NZBs we usually have
|
|
a direct HTTP URL the indexer exposes via Prowlarr.
|
|
- Usenet clients (SABnzbd, NZBGet) typically auto-extract during
|
|
post-processing, so ``archive_pipeline.collect_audio_after_extraction``
|
|
usually has nothing to extract and just walks loose files.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from core.archive_pipeline import collect_audio_after_extraction
|
|
from core.download_plugins.album_bundle import (
|
|
DEFAULT_TRANSIENT_MISS_THRESHOLD,
|
|
copy_audio_files_atomically,
|
|
pick_best_album_release,
|
|
poll_album_download,
|
|
)
|
|
from core.download_plugins.base import DownloadSourcePlugin
|
|
from core.download_plugins.torrent import (
|
|
_adapter_state_to_display,
|
|
_decode_filename,
|
|
_guess_quality_from_title,
|
|
_parse_indexer_id_filter,
|
|
_parse_release_title,
|
|
_row_to_status,
|
|
_COMPLETE_STATES,
|
|
_FILENAME_SEP,
|
|
_POLL_INTERVAL_SECONDS,
|
|
_POLL_TIMEOUT_SECONDS,
|
|
)
|
|
from core.download_plugins.types import AlbumResult, DownloadStatus, TrackResult
|
|
from core.prowlarr_client import (
|
|
DEFAULT_MUSIC_CATEGORIES,
|
|
ProwlarrClient,
|
|
ProwlarrSearchResult,
|
|
)
|
|
from core.usenet_clients import get_active_adapter as get_active_usenet_adapter
|
|
from utils.async_helpers import run_async
|
|
from utils.logging_config import get_logger
|
|
|
|
logger = get_logger("download_plugins.usenet")
|
|
|
|
|
|
class UsenetDownloadPlugin(DownloadSourcePlugin):
|
|
"""Usenet download source backed by Prowlarr + an active usenet
|
|
client adapter (SABnzbd or NZBGet)."""
|
|
|
|
def __init__(self) -> None:
|
|
self._prowlarr = ProwlarrClient()
|
|
self.active_downloads: Dict[str, Dict[str, Any]] = {}
|
|
self._lock = threading.Lock()
|
|
self.shutdown_check = None
|
|
|
|
def set_shutdown_check(self, check_callable):
|
|
self.shutdown_check = check_callable
|
|
|
|
def reload_settings(self) -> None:
|
|
self._prowlarr.reload_settings()
|
|
|
|
def is_configured(self) -> bool:
|
|
if not self._prowlarr.is_configured():
|
|
return False
|
|
adapter = get_active_usenet_adapter()
|
|
return bool(adapter and adapter.is_configured())
|
|
|
|
async def check_connection(self) -> bool:
|
|
if not self._prowlarr.is_configured():
|
|
return False
|
|
adapter = get_active_usenet_adapter()
|
|
if not adapter or not adapter.is_configured():
|
|
return False
|
|
if not await self._prowlarr.check_connection():
|
|
return False
|
|
return await adapter.check_connection()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Search
|
|
# ------------------------------------------------------------------
|
|
|
|
async def search(
|
|
self,
|
|
query: str,
|
|
timeout: Optional[int] = None,
|
|
progress_callback=None,
|
|
) -> Tuple[List[TrackResult], List[AlbumResult]]:
|
|
if not self._prowlarr.is_configured():
|
|
return ([], [])
|
|
try:
|
|
indexer_ids = _parse_indexer_id_filter()
|
|
results = await self._prowlarr.search(
|
|
query,
|
|
categories=DEFAULT_MUSIC_CATEGORIES,
|
|
indexer_ids=indexer_ids,
|
|
)
|
|
except Exception as e:
|
|
logger.error("Usenet plugin search failed: %s", e)
|
|
return ([], [])
|
|
return self._project_results(results)
|
|
|
|
def _project_results(
|
|
self, results: List[ProwlarrSearchResult]
|
|
) -> Tuple[List[TrackResult], List[AlbumResult]]:
|
|
tracks: List[TrackResult] = []
|
|
albums: List[AlbumResult] = []
|
|
for result in results:
|
|
if result.protocol != 'usenet':
|
|
continue
|
|
if not result.download_url:
|
|
continue
|
|
filename = f"{result.download_url}{_FILENAME_SEP}{result.title}"
|
|
quality = _guess_quality_from_title(result.title)
|
|
parsed_artist, parsed_title = _parse_release_title(result.title)
|
|
tr = TrackResult(
|
|
username='usenet',
|
|
filename=filename,
|
|
size=result.size,
|
|
bitrate=None,
|
|
duration=None,
|
|
quality=quality,
|
|
# Usenet doesn't expose per-uploader concurrency the way
|
|
# Soulseek does; fill in neutral non-punishing values.
|
|
free_upload_slots=1,
|
|
upload_speed=0,
|
|
queue_length=0,
|
|
# Pre-fill artist + title so TrackResult.__post_init__
|
|
# doesn't auto-parse the filename — same URL-in-filename
|
|
# gotcha as the torrent plugin.
|
|
artist=parsed_artist or result.indexer_name or 'Usenet',
|
|
title=parsed_title or result.title,
|
|
album=parsed_title or None,
|
|
track_number=None,
|
|
_source_metadata={
|
|
'indexer': result.indexer_name,
|
|
'indexer_id': result.indexer_id,
|
|
'grabs': result.grabs,
|
|
'protocol': 'usenet',
|
|
},
|
|
)
|
|
tracks.append(tr)
|
|
albums.append(AlbumResult(
|
|
username='usenet',
|
|
album_path=f"usenet/{result.guid}",
|
|
album_title=parsed_title or result.title,
|
|
artist=parsed_artist or None,
|
|
track_count=1,
|
|
total_size=result.size,
|
|
tracks=[tr],
|
|
dominant_quality=quality,
|
|
year=None,
|
|
))
|
|
return tracks, albums
|
|
|
|
# ------------------------------------------------------------------
|
|
# Download
|
|
# ------------------------------------------------------------------
|
|
|
|
async def download(
|
|
self,
|
|
username: str,
|
|
filename: str,
|
|
file_size: int = 0,
|
|
) -> Optional[str]:
|
|
if not self.is_configured():
|
|
return None
|
|
nzb_url, display_name = _decode_filename(filename)
|
|
if not nzb_url:
|
|
logger.error("Usenet download missing URL in filename: %r", filename)
|
|
return None
|
|
|
|
download_id = str(uuid.uuid4())
|
|
with self._lock:
|
|
self.active_downloads[download_id] = {
|
|
'id': download_id,
|
|
'filename': filename,
|
|
'username': 'usenet',
|
|
'display_name': display_name,
|
|
'state': 'Initializing',
|
|
'progress': 0.0,
|
|
'size': file_size,
|
|
'transferred': 0,
|
|
'speed': 0,
|
|
'file_path': None,
|
|
'audio_files': [],
|
|
'job_id': None,
|
|
'error': None,
|
|
}
|
|
|
|
thread = threading.Thread(
|
|
target=self._download_thread,
|
|
args=(download_id, nzb_url),
|
|
daemon=True,
|
|
name=f'usenet-dl-{download_id[:8]}',
|
|
)
|
|
thread.start()
|
|
return download_id
|
|
|
|
def _download_thread(self, download_id: str, nzb_url: str) -> None:
|
|
adapter = get_active_usenet_adapter()
|
|
if adapter is None or not adapter.is_configured():
|
|
self._mark_error(download_id, "No usenet client configured")
|
|
return
|
|
|
|
try:
|
|
job_id = run_async(adapter.add_nzb(nzb_url))
|
|
except Exception as e:
|
|
self._mark_error(download_id, f"add_nzb failed: {e}")
|
|
return
|
|
if not job_id:
|
|
self._mark_error(download_id, "Usenet client refused the NZB")
|
|
return
|
|
|
|
with self._lock:
|
|
row = self.active_downloads.get(download_id)
|
|
if row is not None:
|
|
row['job_id'] = job_id
|
|
row['state'] = 'InProgress, Downloading'
|
|
|
|
deadline = time.monotonic() + _POLL_TIMEOUT_SECONDS
|
|
last_save_path: Optional[str] = None
|
|
# Tolerate transient None / 'error' (unmapped state) reads —
|
|
# SAB removes a job from the queue before adding it to history,
|
|
# and on a busy server that gap can span several polls. Same
|
|
# bug class as the album-bundle path; see
|
|
# ``core/download_plugins/album_bundle.py:poll_album_download``
|
|
# docstring for the longer rationale.
|
|
transient_misses = 0
|
|
miss_threshold = DEFAULT_TRANSIENT_MISS_THRESHOLD
|
|
while time.monotonic() < deadline:
|
|
if self.shutdown_check and self.shutdown_check():
|
|
return
|
|
try:
|
|
status = run_async(adapter.get_status(job_id))
|
|
except Exception as e:
|
|
logger.warning("Usenet poll error for %s: %s", job_id, e)
|
|
status = None
|
|
|
|
if status is None:
|
|
transient_misses += 1
|
|
if transient_misses >= miss_threshold:
|
|
self._mark_error(
|
|
download_id,
|
|
f"Usenet job disappeared from client (no status after {miss_threshold} polls)",
|
|
)
|
|
return
|
|
time.sleep(_POLL_INTERVAL_SECONDS)
|
|
continue
|
|
|
|
if status.state != 'error':
|
|
transient_misses = 0
|
|
|
|
with self._lock:
|
|
row = self.active_downloads.get(download_id)
|
|
if row is not None:
|
|
row['progress'] = status.progress * 100.0
|
|
row['transferred'] = status.downloaded
|
|
row['speed'] = status.download_speed
|
|
row['size'] = status.size or row.get('size', 0)
|
|
row['state'] = _adapter_state_to_display(status.state)
|
|
row['error'] = status.error
|
|
if status.save_path:
|
|
last_save_path = status.save_path
|
|
|
|
if status.state in _COMPLETE_STATES:
|
|
self._finalize_download(download_id, last_save_path)
|
|
return
|
|
if status.state == 'failed':
|
|
self._mark_error(download_id, status.error or "Usenet client reported failure")
|
|
return
|
|
if status.state == 'error':
|
|
logger.warning(
|
|
"Usenet poll: '%s' returned unmapped state — treating as transient",
|
|
job_id,
|
|
)
|
|
transient_misses += 1
|
|
if transient_misses >= miss_threshold:
|
|
self._mark_error(
|
|
download_id,
|
|
"Usenet client returned unmapped state repeatedly",
|
|
)
|
|
return
|
|
|
|
time.sleep(_POLL_INTERVAL_SECONDS)
|
|
|
|
self._mark_error(download_id, "Usenet download timed out")
|
|
|
|
def _finalize_download(self, download_id: str, save_path: Optional[str]) -> None:
|
|
if not save_path:
|
|
self._mark_error(download_id, "Usenet job completed but no save_path reported")
|
|
return
|
|
try:
|
|
audio_files = collect_audio_after_extraction(Path(save_path))
|
|
except Exception as e:
|
|
self._mark_error(download_id, f"Post-extract walk failed: {e}")
|
|
return
|
|
if not audio_files:
|
|
self._mark_error(download_id, f"No audio files found in {save_path}")
|
|
return
|
|
primary = audio_files[0]
|
|
with self._lock:
|
|
row = self.active_downloads.get(download_id)
|
|
if row is not None:
|
|
row['state'] = 'Completed, Succeeded'
|
|
row['progress'] = 100.0
|
|
row['file_path'] = str(primary)
|
|
row['audio_files'] = [str(path) for path in audio_files]
|
|
logger.info("Usenet download complete: %s -> %s (%d audio files)",
|
|
download_id[:8], primary.name, len(audio_files))
|
|
|
|
def _mark_error(self, download_id: str, message: str) -> None:
|
|
logger.error("Usenet download %s failed: %s", download_id[:8], message)
|
|
with self._lock:
|
|
row = self.active_downloads.get(download_id)
|
|
if row is not None:
|
|
row['state'] = 'Completed, Errored'
|
|
row['error'] = message
|
|
|
|
# ------------------------------------------------------------------
|
|
# Status / lifecycle
|
|
# ------------------------------------------------------------------
|
|
|
|
async def get_all_downloads(self) -> List[DownloadStatus]:
|
|
with self._lock:
|
|
rows = list(self.active_downloads.values())
|
|
return [_row_to_status(r) for r in rows]
|
|
|
|
async def get_download_status(self, download_id: str) -> Optional[DownloadStatus]:
|
|
with self._lock:
|
|
row = self.active_downloads.get(download_id)
|
|
if row is None:
|
|
return None
|
|
return _row_to_status(row)
|
|
|
|
async def cancel_download(
|
|
self,
|
|
download_id: str,
|
|
username: Optional[str] = None,
|
|
remove: bool = False,
|
|
) -> bool:
|
|
adapter = get_active_usenet_adapter()
|
|
with self._lock:
|
|
row = self.active_downloads.get(download_id)
|
|
job_id = row.get('job_id') if row else None
|
|
if adapter and job_id:
|
|
try:
|
|
await adapter.remove(job_id, delete_files=remove)
|
|
except Exception as e:
|
|
logger.warning("Usenet cancel via adapter failed: %s", e)
|
|
with self._lock:
|
|
if remove:
|
|
self.active_downloads.pop(download_id, None)
|
|
else:
|
|
row = self.active_downloads.get(download_id)
|
|
if row is not None:
|
|
row['state'] = 'Cancelled'
|
|
return True
|
|
|
|
async def clear_all_completed_downloads(self) -> bool:
|
|
with self._lock:
|
|
for did in list(self.active_downloads.keys()):
|
|
state = self.active_downloads[did].get('state', '')
|
|
if state.startswith('Completed') or state == 'Cancelled':
|
|
self.active_downloads.pop(did, None)
|
|
return True
|
|
|
|
# ------------------------------------------------------------------
|
|
# Album-bundle flow
|
|
# ------------------------------------------------------------------
|
|
|
|
def download_album_to_staging(
|
|
self,
|
|
album_name: str,
|
|
artist_name: str,
|
|
staging_dir: str,
|
|
progress_callback=None,
|
|
) -> Dict[str, Any]:
|
|
"""Usenet sibling of ``TorrentDownloadPlugin.download_album_to_staging``.
|
|
See that method's docstring for the contract."""
|
|
result: Dict[str, Any] = {'success': False, 'files': [], 'error': None}
|
|
if not self.is_configured():
|
|
result['error'] = 'Usenet source not configured'
|
|
return result
|
|
|
|
adapter = get_active_usenet_adapter()
|
|
if adapter is None or not adapter.is_configured():
|
|
result['error'] = 'No active usenet client'
|
|
return result
|
|
|
|
def _emit(state: str, **extra) -> None:
|
|
if progress_callback:
|
|
try:
|
|
progress_callback({'state': state, **extra})
|
|
except Exception as cb_exc:
|
|
logger.debug("[Usenet album] progress callback failed: %s", cb_exc)
|
|
|
|
query = f"{artist_name} {album_name}".strip()
|
|
_emit('searching', query=query)
|
|
try:
|
|
search_results = run_async(self._prowlarr.search(
|
|
query, categories=DEFAULT_MUSIC_CATEGORIES,
|
|
indexer_ids=_parse_indexer_id_filter(),
|
|
))
|
|
except Exception as e:
|
|
result['error'] = f'Prowlarr search failed: {e}'
|
|
return result
|
|
|
|
candidates = [r for r in search_results
|
|
if r.protocol == 'usenet' and r.download_url]
|
|
if not candidates:
|
|
result['error'] = f'No usenet results found for "{query}"'
|
|
return result
|
|
|
|
picked = pick_best_album_release(candidates, _guess_quality_from_title)
|
|
if picked is None:
|
|
result['error'] = 'No suitable NZB candidate after filtering'
|
|
return result
|
|
|
|
logger.info("[Usenet album] Picked '%s' (size=%.1fMB grabs=%s indexer=%s)",
|
|
picked.title, picked.size / 1_048_576, picked.grabs, picked.indexer_name)
|
|
_emit('queued', release=picked.title, size=picked.size, grabs=picked.grabs)
|
|
|
|
try:
|
|
job_id = run_async(adapter.add_nzb(picked.download_url))
|
|
except Exception as e:
|
|
result['error'] = f'Usenet client refused the NZB: {e}'
|
|
return result
|
|
if not job_id:
|
|
result['error'] = 'Usenet client refused the NZB'
|
|
return result
|
|
|
|
_emit('downloading', release=picked.title)
|
|
save_path = poll_album_download(
|
|
get_status=lambda: run_async(adapter.get_status(job_id)),
|
|
title=picked.title,
|
|
emit=_emit,
|
|
# Usenet completes into history as 'completed'; no 'seeding'
|
|
# equivalent. Failed is explicit on history failures.
|
|
complete_states=frozenset(['completed']),
|
|
failed_states=frozenset(['failed']),
|
|
is_shutdown=self.shutdown_check,
|
|
log_prefix='[Usenet album]',
|
|
)
|
|
if save_path is None:
|
|
# poll_album_download already emitted the terminal 'failed'
|
|
# state on every failure path (timeout / disappeared /
|
|
# explicit failure / unmapped). UI is unstuck either way.
|
|
result['error'] = 'Usenet download failed or timed out'
|
|
return result
|
|
|
|
_emit('staging', release=picked.title)
|
|
try:
|
|
audio_files = collect_audio_after_extraction(Path(save_path))
|
|
except Exception as e:
|
|
result['error'] = f'Failed to walk audio files: {e}'
|
|
return result
|
|
if not audio_files:
|
|
result['error'] = f'No audio files found in {save_path}'
|
|
return result
|
|
|
|
copied = copy_audio_files_atomically(audio_files, Path(staging_dir))
|
|
if not copied:
|
|
result['error'] = 'No audio files copied to staging'
|
|
return result
|
|
logger.info("[Usenet album] Staged %d audio files for '%s'", len(copied), album_name)
|
|
_emit('staged', count=len(copied))
|
|
result['success'] = True
|
|
result['files'] = copied
|
|
return result
|
|
|