diff --git a/core/downloads/track_detail.py b/core/downloads/track_detail.py new file mode 100644 index 00000000..806de655 --- /dev/null +++ b/core/downloads/track_detail.py @@ -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 diff --git a/tests/downloads/test_track_detail.py b/tests/downloads/test_track_detail.py new file mode 100644 index 00000000..96179ca6 --- /dev/null +++ b/tests/downloads/test_track_detail.py @@ -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' diff --git a/web_server.py b/web_server.py index 07bda0ea..b3d6499f 100644 --- a/web_server.py +++ b/web_server.py @@ -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//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//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."""