Downloads: add per-track detail endpoint for the track-detail modal
New GET /api/downloads/task/<id>/detail merges the live download task with its library_history row (the data the Download History cards show) into one payload the upcoming track-detail modal renders: status kind, title/artist/album, source, quality, final location, AcoustID verdict, and expected-vs-downloaded. Assembly + status classification live in core/downloads/track_detail.py as a pure, importable build_track_detail()/classify_status_kind() (9 unit tests); the endpoint is thin glue that looks up the matching history row by track.
This commit is contained in:
parent
ba6c39bae3
commit
e4bbcfda1b
3 changed files with 237 additions and 0 deletions
106
core/downloads/track_detail.py
Normal file
106
core/downloads/track_detail.py
Normal file
|
|
@ -0,0 +1,106 @@
|
|||
"""Assemble a per-track detail view for the download-modal "track detail" modal.
|
||||
|
||||
Merges a live download task with its ``library_history`` record (the same data
|
||||
the Download History cards render) into one dict the frontend modal consumes.
|
||||
Kept pure + importable so the merge + status classification are unit-tested
|
||||
without Flask or the DB; the web endpoint is thin glue around build_track_detail.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
# error_message substrings that mean "quarantined" (file recoverable) rather
|
||||
# than a plain failure. Mirrors the download-modal status renderer.
|
||||
_QUARANTINE_KEYWORDS = (
|
||||
'integrity check failed',
|
||||
'bit depth filter',
|
||||
'verification failed',
|
||||
'quarantin',
|
||||
)
|
||||
|
||||
|
||||
def classify_status_kind(status: str, error_message: str = '') -> str:
|
||||
"""Map a raw task status to a UI 'kind' that drives the modal layout:
|
||||
completed / quarantined / failed / not_found / in_progress.
|
||||
"""
|
||||
s = (status or '').lower()
|
||||
if s == 'completed':
|
||||
return 'completed'
|
||||
if s in ('failed', 'cancelled'):
|
||||
em = (error_message or '').lower()
|
||||
if any(k in em for k in _QUARANTINE_KEYWORDS):
|
||||
return 'quarantined'
|
||||
return 'failed'
|
||||
if s == 'not_found':
|
||||
return 'not_found'
|
||||
return 'in_progress'
|
||||
|
||||
|
||||
def _first_artist(track_info: Dict[str, Any]) -> str:
|
||||
artists = track_info.get('artists') or []
|
||||
if isinstance(artists, list) and artists:
|
||||
first = artists[0]
|
||||
if isinstance(first, dict):
|
||||
return (first.get('name') or '').strip()
|
||||
return str(first).strip()
|
||||
return (track_info.get('artist') or track_info.get('artist_name') or '').strip()
|
||||
|
||||
|
||||
def _album_name(track_info: Dict[str, Any]) -> str:
|
||||
album = track_info.get('album')
|
||||
if isinstance(album, dict):
|
||||
return (album.get('name') or '').strip()
|
||||
return (album or '').strip() if isinstance(album, str) else ''
|
||||
|
||||
|
||||
def build_track_detail(task: Dict[str, Any], history: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||||
"""Merge a download task (+ optional library_history row) into one detail dict.
|
||||
|
||||
The task supplies live status/source/reason/quarantine id; the history row
|
||||
(when found) supplies the durable provenance — final file path, quality,
|
||||
AcoustID verdict, source, and the expected-vs-downloaded comparison.
|
||||
"""
|
||||
ti = task.get('track_info') if isinstance(task.get('track_info'), dict) else {}
|
||||
status = task.get('status', '') or ''
|
||||
kind = classify_status_kind(status, task.get('error_message', '') or '')
|
||||
|
||||
detail: Dict[str, Any] = {
|
||||
'task_id': task.get('task_id') or task.get('id') or '',
|
||||
'status': status,
|
||||
'status_kind': kind,
|
||||
'title': (ti.get('name') or '').strip(),
|
||||
'artist': _first_artist(ti),
|
||||
'album': _album_name(ti),
|
||||
'source': (task.get('username') or '').strip(),
|
||||
'reason': (task.get('error_message') or '').strip(),
|
||||
'quarantine_entry_id': task.get('quarantine_entry_id') or '',
|
||||
'file_path': (task.get('filename') or '').strip(),
|
||||
'quality': '',
|
||||
'acoustid_result': '',
|
||||
'thumb_url': '',
|
||||
'expected': {},
|
||||
'downloaded': {},
|
||||
}
|
||||
|
||||
if history:
|
||||
detail['file_path'] = (history.get('file_path') or detail['file_path'])
|
||||
detail['quality'] = history.get('quality') or ''
|
||||
detail['acoustid_result'] = history.get('acoustid_result') or ''
|
||||
detail['source'] = history.get('download_source') or detail['source']
|
||||
detail['thumb_url'] = history.get('thumb_url') or ''
|
||||
detail['downloaded'] = {
|
||||
'title': history.get('title') or '',
|
||||
'artist': history.get('artist_name') or '',
|
||||
'album': history.get('album_name') or '',
|
||||
}
|
||||
detail['expected'] = {
|
||||
'title': history.get('source_track_title') or '',
|
||||
'artist': history.get('source_artist') or '',
|
||||
}
|
||||
# Fall back to history values when the task had none.
|
||||
detail['title'] = detail['title'] or detail['downloaded']['title']
|
||||
detail['artist'] = detail['artist'] or detail['downloaded']['artist']
|
||||
detail['album'] = detail['album'] or detail['downloaded']['album']
|
||||
|
||||
return detail
|
||||
89
tests/downloads/test_track_detail.py
Normal file
89
tests/downloads/test_track_detail.py
Normal file
|
|
@ -0,0 +1,89 @@
|
|||
"""Tests for the track-detail assembly (core/downloads/track_detail.py)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from core.downloads.track_detail import build_track_detail, classify_status_kind
|
||||
|
||||
|
||||
# ── status classification ──────────────────────────────────────────────────
|
||||
|
||||
def test_classify_completed():
|
||||
assert classify_status_kind('completed') == 'completed'
|
||||
|
||||
|
||||
def test_classify_quarantined_from_integrity_message():
|
||||
assert classify_status_kind('failed', 'File integrity check failed: Duration mismatch') == 'quarantined'
|
||||
|
||||
|
||||
def test_classify_plain_failure():
|
||||
assert classify_status_kind('failed', 'No sources found') == 'failed'
|
||||
|
||||
|
||||
def test_classify_not_found():
|
||||
assert classify_status_kind('not_found') == 'not_found'
|
||||
|
||||
|
||||
def test_classify_in_progress():
|
||||
assert classify_status_kind('downloading') == 'in_progress'
|
||||
|
||||
|
||||
# ── merge: task only ────────────────────────────────────────────────────────
|
||||
|
||||
def test_build_from_task_only():
|
||||
task = {
|
||||
'task_id': 't1',
|
||||
'status': 'completed',
|
||||
'track_info': {'name': 'HUMBLE.', 'artists': [{'name': 'Kendrick Lamar'}], 'album': {'name': 'DAMN.'}},
|
||||
'username': 'soulseek',
|
||||
'filename': '/music/HUMBLE.flac',
|
||||
}
|
||||
d = build_track_detail(task)
|
||||
assert d['status_kind'] == 'completed'
|
||||
assert d['title'] == 'HUMBLE.'
|
||||
assert d['artist'] == 'Kendrick Lamar'
|
||||
assert d['album'] == 'DAMN.'
|
||||
assert d['file_path'] == '/music/HUMBLE.flac'
|
||||
assert d['source'] == 'soulseek'
|
||||
|
||||
|
||||
def test_quarantined_task_carries_entry_id_and_reason():
|
||||
task = {
|
||||
'task_id': 't2',
|
||||
'status': 'failed',
|
||||
'error_message': 'File integrity check failed: Duration mismatch',
|
||||
'quarantine_entry_id': '20260531_120000_song',
|
||||
'track_info': {'name': 'Clean', 'artists': ['Taylor Swift']},
|
||||
}
|
||||
d = build_track_detail(task)
|
||||
assert d['status_kind'] == 'quarantined'
|
||||
assert d['quarantine_entry_id'] == '20260531_120000_song'
|
||||
assert 'integrity' in d['reason'].lower()
|
||||
assert d['artist'] == 'Taylor Swift' # string-form artist handled
|
||||
|
||||
|
||||
# ── merge: history enriches ─────────────────────────────────────────────────
|
||||
|
||||
def test_history_enriches_provenance():
|
||||
task = {'task_id': 't3', 'status': 'completed', 'track_info': {'name': 'N95', 'artists': [{'name': 'Kendrick Lamar'}]}}
|
||||
history = {
|
||||
'title': 'N95', 'artist_name': 'Kendrick Lamar', 'album_name': 'Mr. Morale',
|
||||
'quality': 'FLAC 16bit', 'file_path': '/lib/N95.flac', 'acoustid_result': 'error',
|
||||
'download_source': 'Soulseek', 'thumb_url': 'http://x/cover.jpg',
|
||||
'source_track_title': 'N95', 'source_artist': 'Kendrick Lamar',
|
||||
}
|
||||
d = build_track_detail(task, history)
|
||||
assert d['file_path'] == '/lib/N95.flac'
|
||||
assert d['quality'] == 'FLAC 16bit'
|
||||
assert d['acoustid_result'] == 'error'
|
||||
assert d['source'] == 'Soulseek'
|
||||
assert d['thumb_url'] == 'http://x/cover.jpg'
|
||||
assert d['downloaded'] == {'title': 'N95', 'artist': 'Kendrick Lamar', 'album': 'Mr. Morale'}
|
||||
assert d['expected'] == {'title': 'N95', 'artist': 'Kendrick Lamar'}
|
||||
|
||||
|
||||
def test_history_fills_missing_title_from_task():
|
||||
task = {'task_id': 't4', 'status': 'completed', 'track_info': {}} # task has no track name
|
||||
history = {'title': 'Recovered Title', 'artist_name': 'Some Artist'}
|
||||
d = build_track_detail(task, history)
|
||||
assert d['title'] == 'Recovered Title'
|
||||
assert d['artist'] == 'Some Artist'
|
||||
|
|
@ -6712,6 +6712,48 @@ def _list_available_download_sources() -> tuple:
|
|||
return download_mode, sources
|
||||
|
||||
|
||||
def _norm_track_key(s: str) -> str:
|
||||
"""Loose normalization for matching a task to its library_history row."""
|
||||
import re
|
||||
return re.sub(r'[^a-z0-9]+', '', (s or '').lower())
|
||||
|
||||
|
||||
@app.route('/api/downloads/task/<task_id>/detail', methods=['GET'])
|
||||
def get_task_detail(task_id):
|
||||
"""Full per-track detail for the track-detail modal: live task state merged
|
||||
with the durable library_history provenance (location, quality, AcoustID
|
||||
verdict, source, expected-vs-downloaded). Thin glue over build_track_detail."""
|
||||
try:
|
||||
from core.downloads.track_detail import build_track_detail
|
||||
with tasks_lock:
|
||||
t = download_tasks.get(task_id)
|
||||
task = dict(t) if isinstance(t, dict) else None
|
||||
if task is None:
|
||||
return jsonify({"success": False, "error": "Task not found"}), 404
|
||||
task['task_id'] = task_id
|
||||
|
||||
# Enrich from the most recent download-history row matching this track.
|
||||
history = None
|
||||
try:
|
||||
ti = task.get('track_info') if isinstance(task.get('track_info'), dict) else {}
|
||||
want_title = _norm_track_key(ti.get('name', ''))
|
||||
if want_title:
|
||||
db = get_database()
|
||||
entries, _ = db.get_library_history(event_type='download', page=1, limit=100)
|
||||
for e in entries:
|
||||
if _norm_track_key(e.get('title', '')) == want_title:
|
||||
history = e
|
||||
break
|
||||
except Exception as hist_err:
|
||||
logger.debug(f"track-detail history lookup failed: {hist_err}")
|
||||
|
||||
detail = build_track_detail(task, history)
|
||||
return jsonify({"success": True, "detail": detail})
|
||||
except Exception as e:
|
||||
logger.error(f"get_task_detail error: {e}")
|
||||
return jsonify({"success": False, "error": str(e)}), 500
|
||||
|
||||
|
||||
@app.route('/api/downloads/task/<task_id>/candidates', methods=['GET'])
|
||||
def get_task_candidates(task_id):
|
||||
"""Returns the cached search candidates for a download task so the UI can show what was found."""
|
||||
|
|
|
|||
Loading…
Reference in a new issue