From a0f4eea9df110a6c378b221cca1605ff4f1403d6 Mon Sep 17 00:00:00 2001 From: Antti Kettunen Date: Fri, 17 Apr 2026 21:16:34 +0300 Subject: [PATCH] Remove one-off tests Tests outside of the tests folder were deemed to be temporary development-time tests and not intended for a regular test suite --- test_acoustid.py | 295 -------- test_hifi_client.py | 1271 -------------------------------- tools/test_youtube_download.py | 1248 ------------------------------- 3 files changed, 2814 deletions(-) delete mode 100644 test_acoustid.py delete mode 100644 test_hifi_client.py delete mode 100644 tools/test_youtube_download.py diff --git a/test_acoustid.py b/test_acoustid.py deleted file mode 100644 index 1c8ed03a..00000000 --- a/test_acoustid.py +++ /dev/null @@ -1,295 +0,0 @@ -#!/usr/bin/env python3 -""" -AcoustID Integration Test Script - -Run this script to test the AcoustID verification system before using it in production. -It will check: -1. fpcalc binary availability -2. API key validation -3. Fingerprint generation (if audio file provided) -4. Full verification flow (if audio file and expected track info provided) - -Usage: - python test_acoustid.py # Basic tests - python test_acoustid.py path/to/audio.mp3 # Test with audio file - python test_acoustid.py path/to/audio.mp3 "Song Title" "Artist Name" # Full test -""" - -import sys -import os -import io - -# Fix Windows encoding issues -if sys.platform == 'win32': - sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') - sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace') - -# Add project root to path -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) - -from pathlib import Path - - -def print_header(text): - print("\n" + "=" * 60) - print(f" {text}") - print("=" * 60) - - -def print_result(success, message): - icon = "[PASS]" if success else "[FAIL]" - print(f" {icon} {message}") - - -def test_chromaprint(): - """Test if chromaprint/fpcalc is available for fingerprinting.""" - print_header("Testing fingerprint backend availability") - - from core.acoustid_client import CHROMAPRINT_AVAILABLE, ACOUSTID_AVAILABLE, FPCALC_PATH - - if not ACOUSTID_AVAILABLE: - print_result(False, "pyacoustid library not installed!") - print("\n To install:") - print(" pip install pyacoustid") - return False - - if CHROMAPRINT_AVAILABLE and FPCALC_PATH: - print_result(True, f"fpcalc ready: {FPCALC_PATH}") - return True - - if CHROMAPRINT_AVAILABLE: - print_result(True, "Fingerprint backend available") - return True - - print_result(False, "No fingerprint backend available!") - print("\n fpcalc will be auto-downloaded on first use.") - print(" Or manually install:") - print(" - Windows: Auto-download supported") - print(" - macOS: brew install chromaprint") - print(" - Linux: apt install libchromaprint-tools") - return False - - -def test_api_key(): - """Test if AcoustID API key is configured and valid.""" - print_header("Testing AcoustID API key") - - from core.acoustid_client import AcoustIDClient - from config.settings import config_manager - - api_key = config_manager.get('acoustid.api_key', '') - - if not api_key: - print_result(False, "No API key configured in settings") - print("\n To configure:") - print(" 1. Get a free API key from https://acoustid.org/new-application") - print(" 2. Add it in Settings > AcoustID section") - return False - - print(f" API key found: {api_key[:8]}...{api_key[-4:]}") - - client = AcoustIDClient() - success, message = client.test_api_key() - - print_result(success, message) - return success - - -def test_enabled(): - """Test if AcoustID verification is enabled.""" - print_header("Testing AcoustID enabled status") - - from config.settings import config_manager - - enabled = config_manager.get('acoustid.enabled', False) - - if enabled: - print_result(True, "AcoustID verification is ENABLED") - else: - print_result(False, "AcoustID verification is DISABLED") - print("\n To enable:") - print(" 1. Go to Settings > AcoustID section") - print(" 2. Check 'Enable Download Verification'") - - return enabled - - -def test_availability(): - """Test overall availability.""" - print_header("Testing overall availability") - - from core.acoustid_client import AcoustIDClient - - client = AcoustIDClient() - available, reason = client.is_available() - - print_result(available, reason) - return available - - -def test_fingerprint_and_lookup(audio_file): - """Test fingerprint generation and AcoustID lookup for an audio file.""" - print_header(f"Testing fingerprint and AcoustID lookup") - print(f" File: {audio_file}") - - if not os.path.isfile(audio_file): - print_result(False, f"File not found: {audio_file}") - return None - - from core.acoustid_client import AcoustIDClient - - client = AcoustIDClient() - - available, reason = client.is_available() - if not available: - print_result(False, f"AcoustID not available: {reason}") - return None - - print(" Fingerprinting and looking up (this may take a moment)...") - result = client.fingerprint_and_lookup(audio_file) - - if result: - recordings = result.get('recordings', []) - score = result.get('best_score', 0) - print_result(True, f"Found {len(recordings)} recording(s) (score: {score:.2f})") - - for i, rec in enumerate(recordings[:5]): # Show first 5 - title = rec.get('title', '?') - artist = rec.get('artist', '?') - mbid = rec.get('mbid', '?') - rec_score = rec.get('score', 0) - print(f" {i+1}. \"{title}\" by {artist} (score: {rec_score:.2f})") - print(f" https://musicbrainz.org/recording/{mbid}") - - if len(recordings) > 5: - print(f" ... and {len(recordings) - 5} more") - - return result - else: - print_result(False, "Track not found in AcoustID database") - print(" This may be a rare/new track not yet fingerprinted.") - return None - - -def test_musicbrainz_lookup(track_name, artist_name): - """Test MusicBrainz lookup for expected track.""" - print_header("Testing MusicBrainz lookup") - print(f" Track: '{track_name}'") - print(f" Artist: '{artist_name}'") - - try: - from database.music_database import MusicDatabase - from core.musicbrainz_service import MusicBrainzService - - db = MusicDatabase() - mb_service = MusicBrainzService(db) - - print(" Searching MusicBrainz...") - result = mb_service.match_recording(track_name, artist_name) - - if result: - mbid = result.get('mbid') - confidence = result.get('confidence', 0) - cached = result.get('cached', False) - - print_result(True, f"Found match (confidence: {confidence}%)") - print(f" MBID: {mbid}") - print(f" https://musicbrainz.org/recording/{mbid}") - print(f" Cached: {cached}") - return result - else: - print_result(False, "No match found in MusicBrainz") - return None - - except Exception as e: - print_result(False, f"Error: {e}") - return None - - -def test_full_verification(audio_file, track_name, artist_name): - """Test the full verification flow.""" - print_header("Testing full verification flow") - print(f" File: {audio_file}") - print(f" Expected: '{track_name}' by '{artist_name}'") - - from core.acoustid_verification import AcoustIDVerification, VerificationResult - - verifier = AcoustIDVerification() - - # Check availability first - available, reason = verifier.quick_check_available() - if not available: - print_result(False, f"Verification not available: {reason}") - return - - print(" Running verification (this may take a moment)...") - result, message = verifier.verify_audio_file( - audio_file, - track_name, - artist_name - ) - - if result == VerificationResult.PASS: - print_result(True, f"VERIFICATION PASSED: {message}") - elif result == VerificationResult.FAIL: - print_result(False, f"VERIFICATION FAILED: {message}") - elif result == VerificationResult.SKIP: - print(f" [SKIP] Verification skipped: {message}") - else: - print(f" [????] Unknown result: {result.value} - {message}") - - -def main(): - print("\n" + "=" * 60) - print(" ACOUSTID VERIFICATION SYSTEM TEST") - print("=" * 60) - - # Parse arguments - audio_file = sys.argv[1] if len(sys.argv) > 1 else None - track_name = sys.argv[2] if len(sys.argv) > 2 else None - artist_name = sys.argv[3] if len(sys.argv) > 3 else None - - # Run basic tests - chromaprint_ok = test_chromaprint() - api_key_ok = test_api_key() - enabled_ok = test_enabled() - available_ok = test_availability() - - # Summary of basic tests - print_header("Basic Tests Summary") - print(f" Chromaprint: {'OK' if chromaprint_ok else 'MISSING'}") - print(f" API key: {'OK' if api_key_ok else 'MISSING/INVALID'}") - print(f" Enabled: {'YES' if enabled_ok else 'NO'}") - print(f" Available: {'YES' if available_ok else 'NO'}") - - if not audio_file: - print("\n" + "-" * 60) - print(" To test fingerprinting, provide an audio file:") - print(" python test_acoustid.py path/to/audio.mp3") - print("\n To test full verification flow:") - print(" python test_acoustid.py path/to/audio.mp3 \"Song Title\" \"Artist\"") - print("-" * 60) - return - - # Test with audio file (combined fingerprint + lookup) - lookup_result = test_fingerprint_and_lookup(audio_file) - - if track_name and artist_name: - # Test MusicBrainz lookup - mb_result = test_musicbrainz_lookup(track_name, artist_name) - - # Test full verification - if available_ok: - test_full_verification(audio_file, track_name, artist_name) - else: - print("\n Skipping full verification test (not available)") - - # Point to log file - print("\n" + "-" * 60) - log_path = Path(__file__).parent / "logs" / "acoustid.log" - print(f" Detailed logs: {log_path}") - print("-" * 60 + "\n") - - -if __name__ == "__main__": - main() diff --git a/test_hifi_client.py b/test_hifi_client.py deleted file mode 100644 index 8cba1f2a..00000000 --- a/test_hifi_client.py +++ /dev/null @@ -1,1271 +0,0 @@ -""" -Comprehensive tests for HiFi API Client (core/hifi_client.py). - -Tests the client against live public hifi-api instances to validate: -- Instance availability and API versioning -- Track search (by title, artist, album, combined queries) -- Stream URL retrieval and base64 manifest decoding -- Album lookup and track listing -- Artist lookup -- Quality tier selection and fallback chain -- Instance failover when one goes down -- TrackResult / DownloadStatus compatibility with Soulseek interfaces -- Actual file download (small test track) -- Rate limiting behavior -- Error handling (bad IDs, empty queries, malformed data) -- Download lifecycle: start → progress → complete/cancel/error - -Usage: - python test_hifi_client.py # Run all tests - python test_hifi_client.py -k search # Run only search tests - python test_hifi_client.py -v # Verbose output -""" - -import os -import sys -import json -import time -import asyncio -import tempfile -import shutil -import threading -from pathlib import Path -from unittest.mock import patch, MagicMock -from dataclasses import dataclass - -import pytest - -# Add project root to path -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) - -from core.hifi_client import ( - HiFiClient, - HIFI_QUALITY_MAP, - DEFAULT_INSTANCES, -) -from core.soulseek_client import TrackResult, AlbumResult, DownloadStatus - - -# ===================== Fixtures ===================== - -@pytest.fixture(scope="session") -def temp_download_dir(): - """Create a temporary download directory for the entire test session.""" - d = tempfile.mkdtemp(prefix="hifi_test_") - yield d - shutil.rmtree(d, ignore_errors=True) - - -@pytest.fixture -def client(temp_download_dir): - """Create a fresh HiFiClient for each test.""" - c = HiFiClient(download_path=temp_download_dir) - return c - - -@pytest.fixture(scope="session") -def shared_client(temp_download_dir): - """Shared client for read-only tests (avoids excessive instance creation).""" - return HiFiClient(download_path=temp_download_dir) - - -@pytest.fixture -def event_loop(): - """Create an event loop for async tests.""" - loop = asyncio.new_event_loop() - yield loop - loop.close() - - -def run_async(coro): - """Helper to run an async function synchronously.""" - loop = asyncio.new_event_loop() - try: - return loop.run_until_complete(coro) - finally: - loop.close() - - -# ===================== 1. Instance & Availability Tests ===================== - -class TestInstanceManagement: - """Tests for API instance management and availability.""" - - def test_01_default_instances_populated(self, client): - """Client should have default instances loaded.""" - assert len(client._instances) >= 1 - assert client._current_instance is not None - - def test_02_custom_instance_priority(self, temp_download_dir): - """Custom base_url should be first in instance list.""" - custom = "https://my-custom-instance.example.com" - c = HiFiClient(download_path=temp_download_dir, base_url=custom) - assert c._instances[0] == custom - assert c._current_instance == custom - - def test_03_custom_instance_trailing_slash(self, temp_download_dir): - """Trailing slash should be stripped from custom instance URL.""" - custom = "https://my-custom-instance.example.com/" - c = HiFiClient(download_path=temp_download_dir, base_url=custom) - assert c._instances[0] == "https://my-custom-instance.example.com" - - def test_04_is_available(self, shared_client): - """At least one public instance should be reachable.""" - assert shared_client.is_available(), "No HiFi API instance is reachable" - - def test_05_get_version(self, shared_client): - """Should return a version string from the API.""" - version = shared_client.get_version() - # version may be None if endpoint doesn't expose it, that's okay - # but the call shouldn't crash - if version: - assert isinstance(version, str) - - def test_06_instance_rotation(self, temp_download_dir): - """Rotating an instance should move it to the back.""" - c = HiFiClient(download_path=temp_download_dir) - first = c._instances[0] - second = c._instances[1] if len(c._instances) > 1 else None - - c._rotate_instance(first) - - assert c._instances[-1] == first - if second: - assert c._current_instance == second - - def test_07_rotate_nonexistent_instance(self, client): - """Rotating a URL not in the list shouldn't crash.""" - original_first = client._current_instance - client._rotate_instance("https://nonexistent.example.com") - assert client._current_instance == original_first - - def test_08_all_instances_exhausted(self, temp_download_dir): - """_api_get should return None when all instances fail.""" - c = HiFiClient(download_path=temp_download_dir) - c._instances = ["https://definitely-not-real-1.invalid", "https://definitely-not-real-2.invalid"] - c._current_instance = c._instances[0] - c._min_interval = 0 # No rate limiting in test - - result = c._api_get("/", timeout=3) - assert result is None - - -# ===================== 2. Rate Limiting Tests ===================== - -class TestRateLimiting: - """Tests for rate limiting between API calls.""" - - def test_09_rate_limit_enforces_interval(self, temp_download_dir): - """Rate limiting should enforce minimum interval between calls.""" - c = HiFiClient(download_path=temp_download_dir) - c._min_interval = 0.3 - - start = time.time() - c._rate_limit() - c._rate_limit() - elapsed = time.time() - start - - assert elapsed >= 0.25, f"Rate limiting should enforce ~0.3s gap, got {elapsed:.3f}s" - - def test_10_rate_limit_first_call_no_delay(self, temp_download_dir): - """First API call shouldn't have artificial delay.""" - c = HiFiClient(download_path=temp_download_dir) - c._min_interval = 1.0 - c._last_api_call = 0 # Reset - - start = time.time() - c._rate_limit() - elapsed = time.time() - start - - assert elapsed < 0.1, f"First call should be instant, took {elapsed:.3f}s" - - -# ===================== 3. Search Tests ===================== - -class TestSearch: - """Tests for track search functionality.""" - - def test_11_search_by_title(self, shared_client): - """Search by title should return results.""" - results = shared_client.search_tracks(title="Bohemian Rhapsody") - assert len(results) > 0, "Expected results for 'Bohemian Rhapsody'" - assert results[0]['title'], "Track should have a title" - - def test_12_search_by_artist(self, shared_client): - """Search by artist alone may return empty (API returns artist objects, not tracks). - This test validates it doesn't crash.""" - results = shared_client.search_tracks(artist="Queen") - # Artist-only search hits /search/?a=Queen which returns artist objects, - # not tracks — so 0 results is expected behavior. - assert isinstance(results, list) - - def test_13_search_by_title_and_artist(self, shared_client): - """Combined title + artist search should return relevant results.""" - results = shared_client.search_tracks(title="Stairway to Heaven", artist="Led Zeppelin") - assert len(results) > 0, "Expected results for 'Stairway to Heaven' by Led Zeppelin" - - # Check that at least one result mentions the artist - found_artist = False - for r in results: - if 'led zeppelin' in r.get('artist', '').lower(): - found_artist = True - break - assert found_artist, "Expected at least one result from Led Zeppelin" - - def test_14_search_by_album(self, shared_client): - """Search by album alone may return empty (API returns album objects, not tracks). - This test validates it doesn't crash.""" - results = shared_client.search_tracks(album="Dark Side of the Moon") - # Album-only search hits /search/?al=... which returns album objects, - # not tracks — so 0 results is expected behavior. - assert isinstance(results, list) - - def test_15_search_limit(self, shared_client): - """Search should respect the limit parameter.""" - results = shared_client.search_tracks(title="Love", limit=5) - assert len(results) <= 5, f"Expected ≤5 results, got {len(results)}" - - def test_16_search_no_terms(self, client): - """Search with no terms should return empty list.""" - results = client.search_tracks() - assert results == [] - - def test_17_search_gibberish(self, shared_client): - """Search for gibberish should return empty or handle gracefully.""" - results = shared_client.search_tracks(title="xzqwkjhgf9876zzz") - assert isinstance(results, list) - - def test_18_search_generic(self, shared_client): - """Generic search_raw() should call search_tracks with title.""" - results = shared_client.search_raw("Never Gonna Give You Up") - assert len(results) > 0 - - def test_19_search_result_fields(self, shared_client): - """Search results should have all expected fields.""" - results = shared_client.search_tracks(title="Yesterday", artist="Beatles") - assert len(results) > 0 - - track = results[0] - required_fields = ['id', 'title', 'artist', 'album', 'duration_ms'] - for field in required_fields: - assert field in track, f"Missing field: {field}" - - assert track['id'] is not None, "Track ID should not be None" - assert isinstance(track['title'], str) - assert isinstance(track['artist'], str) - - def test_20_search_duration_is_milliseconds(self, shared_client): - """Duration should be in milliseconds (typically > 30000 for a normal song).""" - results = shared_client.search_tracks(title="Bohemian Rhapsody", artist="Queen") - if results: - track = results[0] - duration = track.get('duration_ms', 0) - # Bohemian Rhapsody is ~6 minutes = ~360000ms - if duration > 0: - assert duration > 10000, f"Duration {duration}ms seems too low — might be in seconds" - - def test_21_search_special_characters(self, shared_client): - """Search should handle special characters.""" - results = shared_client.search_tracks(title="What's Going On", artist="Marvin Gaye") - assert isinstance(results, list) - - def test_22_search_unicode(self, shared_client): - """Search should handle unicode characters.""" - results = shared_client.search_tracks(title="Für Elise") - assert isinstance(results, list) - - def test_23_search_very_long_query(self, shared_client): - """Very long query should not crash.""" - results = shared_client.search_tracks(title="A" * 500) - assert isinstance(results, list) - - -# ===================== 4. Track Info Tests ===================== - -class TestTrackInfo: - """Tests for individual track info retrieval.""" - - def _get_test_track_id(self, client): - """Helper: search for a track and return its ID.""" - results = client.search_tracks(title="Bohemian Rhapsody", artist="Queen") - if results: - return results[0]['id'] - return None - - def test_24_get_track_info(self, shared_client): - """Should return track info for a valid ID.""" - track_id = self._get_test_track_id(shared_client) - if not track_id: - pytest.skip("No search results to get a track ID") - - info = shared_client.get_track_info(track_id) - assert info is not None, f"Expected track info for ID {track_id}" - assert info.get('title'), "Track info should have a title" - assert info.get('artist'), "Track info should have an artist" - - def test_25_get_track_info_invalid_id(self, shared_client): - """Invalid track ID should return None, not crash.""" - info = shared_client.get_track_info(99999999999) - # May return None or an error — just shouldn't raise - assert info is None or isinstance(info, dict) - - def test_26_get_track_info_zero_id(self, shared_client): - """Zero ID should handle gracefully.""" - info = shared_client.get_track_info(0) - assert info is None or isinstance(info, dict) - - -# ===================== 5. Stream URL / Manifest Tests ===================== - -class TestStreamURL: - """Tests for stream URL retrieval and manifest decoding.""" - - def _get_test_track_id(self, client): - results = client.search_tracks(title="Bohemian Rhapsody", artist="Queen") - return results[0]['id'] if results else None - - def test_27_get_stream_url_lossless(self, shared_client): - """Should return a stream URL for lossless quality.""" - track_id = self._get_test_track_id(shared_client) - if not track_id: - pytest.skip("No search results") - - stream = shared_client.get_stream_url(track_id, quality='lossless') - if stream is None: - pytest.skip("Stream URL not available (may be geo-restricted)") - - assert 'url' in stream, "Stream info should contain 'url'" - assert stream['url'].startswith('http'), f"URL should be HTTP(S): {stream['url'][:100]}" - assert stream['quality'] == 'lossless' - - def test_28_get_stream_url_hires(self, shared_client): - """Should try to get hi-res stream URL.""" - track_id = self._get_test_track_id(shared_client) - if not track_id: - pytest.skip("No search results") - - stream = shared_client.get_stream_url(track_id, quality='hires') - # Hi-res may not be available for all tracks — just shouldn't crash - if stream: - assert 'url' in stream - - def test_29_get_stream_url_high(self, shared_client): - """Should get AAC stream URL.""" - track_id = self._get_test_track_id(shared_client) - if not track_id: - pytest.skip("No search results") - - stream = shared_client.get_stream_url(track_id, quality='high') - if stream: - assert 'url' in stream - assert stream['quality'] == 'high' - - def test_30_get_stream_url_invalid_track(self, shared_client): - """Invalid track ID should return None.""" - stream = shared_client.get_stream_url(99999999999, quality='lossless') - assert stream is None - - def test_31_get_stream_url_invalid_quality(self, shared_client): - """Invalid quality key should fall back to lossless.""" - track_id = self._get_test_track_id(shared_client) - if not track_id: - pytest.skip("No search results") - - # 'nonexistent' isn't in HIFI_QUALITY_MAP, should fall back - stream = shared_client.get_stream_url(track_id, quality='nonexistent') - # Should not crash — either returns data with lossless fallback or None - - def test_32_stream_url_has_encryption_info(self, shared_client): - """Stream info should include encryption field.""" - track_id = self._get_test_track_id(shared_client) - if not track_id: - pytest.skip("No search results") - - stream = shared_client.get_stream_url(track_id, quality='lossless') - if stream: - assert 'encryption' in stream - # Most should be 'NONE' for public instances - assert stream['encryption'] in ('NONE', 'OLD_AES', ''), \ - f"Unexpected encryption: {stream['encryption']}" - - -# ===================== 6. Album Tests ===================== - -class TestAlbum: - """Tests for album lookup.""" - - def _get_test_album_id(self, client): - """Search for a track and extract its album ID if available.""" - results = client.search_tracks(title="Bohemian Rhapsody", artist="Queen", limit=5) - for r in results: - # Album ID may be embedded in album info - if r.get('album'): - # We need an actual album ID — search the album endpoint - break - return None - - def test_33_get_album_known_id(self, shared_client): - """Should return album data for a known Tidal album ID.""" - # "A Night at the Opera" by Queen — Tidal album ID 36393265 - album = shared_client.get_album(36393265) - if album is None: - pytest.skip("Album endpoint not available or ID changed") - - assert album.get('title'), "Album should have a title" - assert isinstance(album.get('tracks', []), list) - - def test_34_get_album_tracks_have_metadata(self, shared_client): - """Album tracks should have proper metadata.""" - album = shared_client.get_album(36393265) - if not album or not album.get('tracks'): - pytest.skip("Album data not available") - - for track in album['tracks'][:3]: - assert track.get('title'), f"Track missing title: {track}" - assert track.get('id'), f"Track missing ID: {track}" - - def test_35_get_album_invalid_id(self, shared_client): - """Invalid album ID should return None.""" - album = shared_client.get_album(99999999999) - assert album is None or (isinstance(album, dict) and len(album.get('tracks', [])) == 0) - - def test_36_get_album_track_count(self, shared_client): - """Album track_count should match actual tracks returned.""" - album = shared_client.get_album(36393265) - if not album: - pytest.skip("Album data not available") - - tracks = album.get('tracks', []) - count = album.get('track_count', 0) - # These should be close (may differ if some tracks are unavailable) - if tracks and count: - assert abs(len(tracks) - count) <= 2, \ - f"Track count mismatch: {len(tracks)} tracks vs reported {count}" - - -# ===================== 7. Artist Tests ===================== - -class TestArtist: - """Tests for artist lookup.""" - - def test_37_get_artist_known_id(self, shared_client): - """Should return artist data for Queen (Tidal artist ID 30157).""" - artist = shared_client.get_artist(30157) - if artist is None: - pytest.skip("Artist endpoint not available") - - assert isinstance(artist, dict) - - def test_38_get_artist_invalid_id(self, shared_client): - """Invalid artist ID should return None.""" - artist = shared_client.get_artist(99999999999) - assert artist is None or isinstance(artist, dict) - - -# ===================== 8. Quality Map Tests ===================== - -class TestQualityMap: - """Tests for quality tier configuration.""" - - def test_39_all_quality_tiers_present(self): - """All four quality tiers should be defined.""" - assert 'hires' in HIFI_QUALITY_MAP - assert 'lossless' in HIFI_QUALITY_MAP - assert 'high' in HIFI_QUALITY_MAP - assert 'low' in HIFI_QUALITY_MAP - - def test_40_quality_tiers_have_required_fields(self): - """Each quality tier should have api_value, label, extension, bitrate, codec.""" - for key, tier in HIFI_QUALITY_MAP.items(): - assert 'api_value' in tier, f"{key} missing api_value" - assert 'label' in tier, f"{key} missing label" - assert 'extension' in tier, f"{key} missing extension" - assert 'bitrate' in tier, f"{key} missing bitrate" - assert 'codec' in tier, f"{key} missing codec" - - def test_41_flac_tiers_have_flac_extension(self): - """Lossless tiers should produce .flac files.""" - assert HIFI_QUALITY_MAP['hires']['extension'] == 'flac' - assert HIFI_QUALITY_MAP['lossless']['extension'] == 'flac' - - def test_42_aac_tiers_have_m4a_extension(self): - """Lossy tiers should produce .m4a files.""" - assert HIFI_QUALITY_MAP['high']['extension'] == 'm4a' - assert HIFI_QUALITY_MAP['low']['extension'] == 'm4a' - - def test_43_bitrate_ordering(self): - """Bitrates should descend: hires > lossless > high > low.""" - assert HIFI_QUALITY_MAP['hires']['bitrate'] > HIFI_QUALITY_MAP['lossless']['bitrate'] - assert HIFI_QUALITY_MAP['lossless']['bitrate'] > HIFI_QUALITY_MAP['high']['bitrate'] - assert HIFI_QUALITY_MAP['high']['bitrate'] > HIFI_QUALITY_MAP['low']['bitrate'] - - -# ===================== 9. Parse Track Tests ===================== - -class TestParseTrack: - """Tests for _parse_track internal method.""" - - def test_44_parse_track_basic(self, client): - """Should parse a simple track dict.""" - item = { - 'id': 12345, - 'title': 'Test Song', - 'artists': [{'name': 'Test Artist'}], - 'album': {'title': 'Test Album'}, - 'duration': 240, - 'trackNumber': 3, - 'isrc': 'USTEST0001', - } - result = client._parse_track(item) - assert result['id'] == 12345 - assert result['title'] == 'Test Song' - assert result['artist'] == 'Test Artist' - assert result['album'] == 'Test Album' - assert result['duration_ms'] == 240000 # 240s → 240000ms - assert result['track_number'] == 3 - assert result['isrc'] == 'USTEST0001' - - def test_45_parse_track_multiple_artists(self, client): - """Should join multiple artists with commas.""" - item = { - 'id': 1, - 'title': 'Collab', - 'artists': [{'name': 'Artist A'}, {'name': 'Artist B'}, {'name': 'Artist C'}], - 'duration': 180, - } - result = client._parse_track(item) - assert result['artist'] == 'Artist A, Artist B, Artist C' - - def test_46_parse_track_string_artist(self, client): - """Should handle artist as a plain string.""" - item = {'id': 1, 'title': 'Song', 'artist': 'Solo Artist', 'duration': 100} - result = client._parse_track(item) - assert result['artist'] == 'Solo Artist' - - def test_47_parse_track_dict_artist(self, client): - """Should handle artist as a dict with name key.""" - item = {'id': 1, 'title': 'Song', 'artist': {'name': 'Dict Artist'}, 'duration': 100} - result = client._parse_track(item) - assert result['artist'] == 'Dict Artist' - - def test_48_parse_track_missing_artists(self, client): - """Missing artist should default to 'Unknown Artist'.""" - item = {'id': 1, 'title': 'Orphan Song', 'duration': 100} - result = client._parse_track(item) - assert result['artist'] == 'Unknown Artist' - - def test_49_parse_track_string_album(self, client): - """Should handle album as a plain string.""" - item = {'id': 1, 'title': 'Song', 'album': 'String Album', 'duration': 100} - result = client._parse_track(item) - assert result['album'] == 'String Album' - - def test_50_parse_track_missing_album(self, client): - """Missing album should be empty string.""" - item = {'id': 1, 'title': 'Song', 'duration': 100} - result = client._parse_track(item) - assert result['album'] == '' - - def test_51_parse_track_duration_already_ms(self, client): - """Duration >= 100000 should be treated as already in milliseconds.""" - item = {'id': 1, 'title': 'Song', 'duration': 240000} - result = client._parse_track(item) - assert result['duration_ms'] == 240000 - - def test_52_parse_track_duration_seconds(self, client): - """Duration < 100000 should be converted from seconds to ms.""" - item = {'id': 1, 'title': 'Song', 'duration': 240} - result = client._parse_track(item) - assert result['duration_ms'] == 240000 - - def test_53_parse_track_zero_duration(self, client): - """Zero duration should stay zero.""" - item = {'id': 1, 'title': 'Song', 'duration': 0} - result = client._parse_track(item) - assert result['duration_ms'] == 0 - - def test_54_parse_track_name_fallback(self, client): - """Should fall back to 'name' key if 'title' missing.""" - item = {'id': 1, 'name': 'Fallback Name', 'duration': 100} - result = client._parse_track(item) - assert result['title'] == 'Fallback Name' - - def test_55_parse_track_explicit_flag(self, client): - """Should preserve explicit flag.""" - item = {'id': 1, 'title': 'Song', 'duration': 100, 'explicit': True} - result = client._parse_track(item) - assert result['explicit'] is True - - def test_56_parse_track_quality_field(self, client): - """Should parse audioQuality field.""" - item = {'id': 1, 'title': 'Song', 'duration': 100, 'audioQuality': 'LOSSLESS'} - result = client._parse_track(item) - assert result['quality'] == 'LOSSLESS' - - def test_57_parse_track_artists_with_strings(self, client): - """Should handle artists list containing plain strings.""" - item = {'id': 1, 'title': 'Song', 'artists': ['Artist A', 'Artist B'], 'duration': 100} - result = client._parse_track(item) - assert result['artist'] == 'Artist A, Artist B' - - def test_58_parse_track_empty_artist_names(self, client): - """Should skip empty artist names.""" - item = {'id': 1, 'title': 'Song', 'artists': [{'name': ''}, {'name': 'Real Artist'}], 'duration': 100} - result = client._parse_track(item) - assert result['artist'] == 'Real Artist' - - -# ===================== 10. TrackResult Compatibility Tests ===================== - -class TestTrackResultCompatibility: - """Tests for Soulseek-compatible TrackResult conversion.""" - - def test_59_to_track_result_basic(self, client): - """Should convert track dict to TrackResult.""" - track = { - 'id': 12345, - 'title': 'Test Song', - 'artist': 'Test Artist', - 'album': 'Test Album', - 'duration_ms': 240000, - 'track_number': 3, - } - q_info = HIFI_QUALITY_MAP['lossless'] - result = client._to_track_result(track, q_info) - - assert isinstance(result, TrackResult) - assert result.username == 'hifi' - assert result.artist == 'Test Artist' - assert result.title == 'Test Song' - assert result.album == 'Test Album' - assert result.bitrate == 1411 - assert result.duration == 240000 - assert result.track_number == 3 - - def test_60_track_result_filename_format(self, client): - """Filename should be 'track_id||display_name'.""" - track = {'id': 999, 'title': 'My Song', 'artist': 'My Artist'} - q_info = HIFI_QUALITY_MAP['lossless'] - result = client._to_track_result(track, q_info) - - assert '||' in result.filename - parts = result.filename.split('||', 1) - assert parts[0] == '999' - assert 'My Artist' in parts[1] - assert 'My Song' in parts[1] - - def test_61_track_result_hifi_username(self, client): - """All HiFi results should use 'hifi' username.""" - track = {'id': 1, 'title': 'S', 'artist': 'A'} - result = client._to_track_result(track, HIFI_QUALITY_MAP['lossless']) - assert result.username == 'hifi' - - def test_62_track_result_high_slots(self, client): - """HiFi results should have high slot count (always available).""" - track = {'id': 1, 'title': 'S', 'artist': 'A'} - result = client._to_track_result(track, HIFI_QUALITY_MAP['lossless']) - assert result.free_upload_slots >= 99 - - def test_63_track_result_different_qualities(self, client): - """Different quality tiers should produce different bitrates.""" - track = {'id': 1, 'title': 'S', 'artist': 'A'} - - hires = client._to_track_result(track, HIFI_QUALITY_MAP['hires']) - lossless = client._to_track_result(track, HIFI_QUALITY_MAP['lossless']) - high = client._to_track_result(track, HIFI_QUALITY_MAP['high']) - - assert hires.bitrate > lossless.bitrate > high.bitrate - - -# ===================== 11. Async Search Compatible Tests ===================== - -class TestSearchCompatible: - """Tests for the async Soulseek-compatible search interface.""" - - def test_64_search_returns_tuple(self, shared_client): - """search should return (tracks, albums) tuple.""" - tracks, albums = run_async(shared_client.search("Bohemian Rhapsody")) - assert isinstance(tracks, list) - assert isinstance(albums, list) - - def test_65_search_track_results(self, shared_client): - """Results should be TrackResult instances.""" - tracks, _ = run_async(shared_client.search("Yesterday Beatles")) - if tracks: - assert isinstance(tracks[0], TrackResult) - - def test_66_search_empty_query(self, client): - """Empty query should return empty lists.""" - tracks, albums = run_async(client.search("")) - assert tracks == [] or isinstance(tracks, list) - - def test_67_search_albums_always_empty(self, shared_client): - """Albums list should always be empty (HiFi doesn't return album results from search).""" - _, albums = run_async(shared_client.search("Dark Side of the Moon")) - assert albums == [] - - -# ===================== 12. Download Lifecycle Tests ===================== - -class TestDownloadLifecycle: - """Tests for download start, tracking, cancel, and cleanup.""" - - def test_68_download_creates_tracking_entry(self, client): - """Starting a download should create an entry in active_downloads.""" - # Use a valid-ish filename format but don't care if it actually downloads - download_id = run_async(client.download('hifi', '12345||Test Artist - Test Song')) - - if download_id: - assert download_id in client.active_downloads - info = client.active_downloads[download_id] - assert info['state'] in ('Initializing', 'InProgress, Downloading', 'Errored') - assert info['username'] == 'hifi' - - def test_69_download_invalid_filename_format(self, client): - """Filename without || separator should fail gracefully.""" - download_id = run_async(client.download('hifi', 'invalid_no_separator')) - assert download_id is None - - def test_70_download_invalid_track_id(self, client): - """Non-numeric track ID should fail gracefully.""" - download_id = run_async(client.download('hifi', 'abc||Artist - Song')) - assert download_id is None - - def test_71_cancel_download(self, client): - """Should be able to cancel a download.""" - # Start a download - download_id = run_async(client.download('hifi', '99999999||Fake - Track')) - if not download_id: - pytest.skip("Download didn't start") - - result = run_async(client.cancel_download(download_id)) - assert result is True - - info = client.active_downloads.get(download_id) - if info: - assert info['state'] == 'Cancelled' - - def test_72_cancel_nonexistent_download(self, client): - """Cancelling a non-existent download should return False.""" - result = run_async(client.cancel_download('fake-uuid-12345')) - assert result is False - - def test_73_cancel_with_remove(self, client): - """Cancel with remove=True should delete from active_downloads.""" - download_id = run_async(client.download('hifi', '99999999||Fake - Track')) - if not download_id: - pytest.skip("Download didn't start") - - run_async(client.cancel_download(download_id, remove=True)) - assert download_id not in client.active_downloads - - def test_74_get_all_downloads(self, client): - """get_all_downloads should return DownloadStatus list.""" - statuses = run_async(client.get_all_downloads()) - assert isinstance(statuses, list) - for s in statuses: - assert isinstance(s, DownloadStatus) - - def test_75_get_download_status(self, client): - """Should return status for a known download.""" - download_id = run_async(client.download('hifi', '99999999||Test - Track')) - if not download_id: - pytest.skip("Download didn't start") - - status = run_async(client.get_download_status(download_id)) - assert status is not None - assert isinstance(status, DownloadStatus) - assert status.id == download_id - - def test_76_get_download_status_unknown(self, client): - """Should return None for unknown download ID.""" - status = run_async(client.get_download_status('nonexistent-uuid')) - assert status is None - - def test_77_clear_completed_downloads(self, client): - """Should clear terminal downloads but keep active ones.""" - # Add a fake completed download - with client._download_lock: - client.active_downloads['fake-completed'] = { - 'id': 'fake-completed', - 'filename': 'test', - 'username': 'hifi', - 'state': 'Completed, Succeeded', - 'progress': 100.0, - 'size': 1000, - 'transferred': 1000, - 'speed': 0, - } - client.active_downloads['fake-active'] = { - 'id': 'fake-active', - 'filename': 'test2', - 'username': 'hifi', - 'state': 'InProgress, Downloading', - 'progress': 50.0, - 'size': 2000, - 'transferred': 1000, - 'speed': 100, - } - - run_async(client.clear_all_completed_downloads()) - - assert 'fake-completed' not in client.active_downloads - assert 'fake-active' in client.active_downloads - - def test_78_clear_errored_downloads(self, client): - """Should clear errored downloads.""" - with client._download_lock: - client.active_downloads['fake-errored'] = { - 'id': 'fake-errored', - 'filename': 'err', - 'username': 'hifi', - 'state': 'Errored', - 'progress': 0, - 'size': 0, - 'transferred': 0, - 'speed': 0, - } - - run_async(client.clear_all_completed_downloads()) - assert 'fake-errored' not in client.active_downloads - - def test_79_clear_cancelled_downloads(self, client): - """Should clear cancelled downloads.""" - with client._download_lock: - client.active_downloads['fake-cancelled'] = { - 'id': 'fake-cancelled', - 'filename': 'can', - 'username': 'hifi', - 'state': 'Cancelled', - 'progress': 0, - 'size': 0, - 'transferred': 0, - 'speed': 0, - } - - run_async(client.clear_all_completed_downloads()) - assert 'fake-cancelled' not in client.active_downloads - - -# ===================== 13. Download Sync (Quality Fallback) Tests ===================== - -class TestDownloadSync: - """Tests for the synchronous download with quality fallback chain.""" - - def test_80_quality_chain_starts_at_configured_quality(self, client): - """Quality fallback chain should start from configured quality.""" - with patch('core.hifi_client.config_manager') as mock_config: - mock_config.get.return_value = 'high' - chain = ['hires', 'lossless', 'high', 'low'] - start = chain.index('high') - assert chain[start:] == ['high', 'low'] - - def test_81_quality_chain_starts_at_hires(self, client): - """If configured quality is hires, chain starts from the top.""" - with patch('core.hifi_client.config_manager') as mock_config: - mock_config.get.return_value = 'hires' - chain = ['hires', 'lossless', 'high', 'low'] - start = chain.index('hires') - assert chain[start:] == ['hires', 'lossless', 'high', 'low'] - - def test_82_download_sync_shutdown_check(self, temp_download_dir): - """Download should abort if shutdown_check returns True.""" - c = HiFiClient(download_path=temp_download_dir) - c.set_shutdown_check(lambda: True) - - result = c._download_sync('test-id', 12345, 'Test - Track') - assert result is None - - def test_83_safe_filename_generation(self, client): - """Filenames with special chars should be sanitized.""" - import re - display = 'Artist: "Song" / ' - safe = re.sub(r'[<>:"/\\|?*]', '_', display) - assert '<' not in safe - assert '>' not in safe - assert '"' not in safe - assert ':' not in safe - assert '/' not in safe - assert '\\' not in safe - assert '|' not in safe - assert '?' not in safe - assert '*' not in safe - - -# ===================== 14. Shutdown Check Tests ===================== - -class TestShutdownCheck: - """Tests for shutdown check callback.""" - - def test_84_set_shutdown_check(self, client): - """Should store the shutdown check callback.""" - check = lambda: False - client.set_shutdown_check(check) - assert client.shutdown_check is check - - def test_85_shutdown_check_default_none(self, temp_download_dir): - """Default shutdown check should be None.""" - c = HiFiClient(download_path=temp_download_dir) - assert c.shutdown_check is None - - -# ===================== 15. Download Path Tests ===================== - -class TestDownloadPath: - """Tests for download path configuration.""" - - def test_86_custom_download_path(self, temp_download_dir): - """Should use provided download path.""" - c = HiFiClient(download_path=temp_download_dir) - assert str(c.download_path) == temp_download_dir - - def test_87_download_path_created(self): - """Should create download directory if it doesn't exist.""" - test_path = tempfile.mktemp(prefix="hifi_mkdir_test_") - try: - c = HiFiClient(download_path=test_path) - assert Path(test_path).exists() - finally: - shutil.rmtree(test_path, ignore_errors=True) - - def test_88_default_download_path(self): - """Default path should come from config_manager or './downloads'.""" - with patch('core.hifi_client.config_manager') as mock_config: - mock_config.get.return_value = './test_default_downloads' - c = HiFiClient() - expected = Path('./test_default_downloads').resolve() - # Just verify it doesn't crash; actual path depends on config - assert c.download_path is not None - # Cleanup - shutil.rmtree(str(c.download_path), ignore_errors=True) - - -# ===================== 16. Instance Failover Integration Tests ===================== - -class TestInstanceFailover: - """Tests for multi-instance failover behavior.""" - - def test_89_failover_on_connection_error(self, temp_download_dir): - """Should failover to next instance on connection error.""" - c = HiFiClient(download_path=temp_download_dir) - c._instances = [ - "https://definitely-not-real.invalid", # Will fail - c._instances[0] if c._instances else "https://triton.squid.wtf", # Should succeed - ] - c._current_instance = c._instances[0] - c._min_interval = 0 - - # After failing on first, should try second - result = c._api_get("/", timeout=3) - # If second instance is up, we should get data - # If not, at least verify no crash and first was rotated - assert c._instances[-1] == "https://definitely-not-real.invalid" - - def test_90_failover_preserves_all_instances(self, temp_download_dir): - """Failover should move failed instance to back, not remove it.""" - c = HiFiClient(download_path=temp_download_dir) - original_count = len(c._instances) - first = c._instances[0] - - c._rotate_instance(first) - - assert len(c._instances) == original_count - assert c._instances[-1] == first - - -# ===================== 17. HTTP Session Tests ===================== - -class TestHTTPSession: - """Tests for HTTP session configuration.""" - - def test_91_session_user_agent(self, client): - """Session should have SoulSync user agent.""" - ua = client.session.headers.get('User-Agent') - assert 'SoulSync' in ua - - def test_92_session_accept_json(self, client): - """Session should accept JSON.""" - accept = client.session.headers.get('Accept') - assert 'json' in accept - - -# ===================== 18. Thread Safety Tests ===================== - -class TestThreadSafety: - """Tests for thread-safe operations.""" - - def test_93_concurrent_download_tracking(self, client): - """Multiple threads should safely add/read downloads.""" - errors = [] - - def add_download(i): - try: - with client._download_lock: - client.active_downloads[f'test-{i}'] = { - 'id': f'test-{i}', - 'filename': f'track-{i}', - 'username': 'hifi', - 'state': 'InProgress, Downloading', - 'progress': 0, - 'size': 0, - 'transferred': 0, - 'speed': 0, - } - except Exception as e: - errors.append(e) - - threads = [threading.Thread(target=add_download, args=(i,)) for i in range(20)] - for t in threads: - t.start() - for t in threads: - t.join() - - assert len(errors) == 0 - assert len([k for k in client.active_downloads if k.startswith('test-')]) == 20 - - def test_94_concurrent_instance_rotation(self, temp_download_dir): - """Multiple threads rotating instances shouldn't crash.""" - c = HiFiClient(download_path=temp_download_dir) - errors = [] - - def rotate(i): - try: - if c._instances: - c._rotate_instance(c._instances[0]) - except Exception as e: - errors.append(e) - - threads = [threading.Thread(target=rotate, args=(i,)) for i in range(10)] - for t in threads: - t.start() - for t in threads: - t.join() - - assert len(errors) == 0 - - -# ===================== 19. DownloadStatus Field Tests ===================== - -class TestDownloadStatusFields: - """Tests for DownloadStatus dataclass compatibility.""" - - def test_95_download_status_all_fields(self, client): - """DownloadStatus should have all required fields.""" - with client._download_lock: - client.active_downloads['field-test'] = { - 'id': 'field-test', - 'filename': '123||Artist - Title', - 'username': 'hifi', - 'state': 'InProgress, Downloading', - 'progress': 45.5, - 'size': 50_000_000, - 'transferred': 22_750_000, - 'speed': 1_000_000, - 'time_remaining': 27, - 'file_path': '/tmp/test.flac', - } - - status = run_async(client.get_download_status('field-test')) - assert status.id == 'field-test' - assert status.filename == '123||Artist - Title' - assert status.username == 'hifi' - assert status.state == 'InProgress, Downloading' - assert status.progress == 45.5 - assert status.size == 50_000_000 - assert status.transferred == 22_750_000 - assert status.speed == 1_000_000 - assert status.time_remaining == 27 - assert status.file_path == '/tmp/test.flac' - - def test_96_download_status_optional_fields_default(self, client): - """Optional fields should default to None.""" - with client._download_lock: - client.active_downloads['optional-test'] = { - 'id': 'optional-test', - 'filename': 'test', - 'username': 'hifi', - 'state': 'Initializing', - 'progress': 0, - 'size': 0, - 'transferred': 0, - 'speed': 0, - } - - status = run_async(client.get_download_status('optional-test')) - assert status.time_remaining is None - assert status.file_path is None - - -# ===================== 20. Live Download Test ===================== - -class TestLiveDownload: - """Live download test — actually downloads a track from HiFi API.""" - - def test_97_live_search_and_get_stream(self, shared_client): - """Full flow: search → get track info → get stream URL.""" - results = shared_client.search_tracks(title="Yesterday", artist="Beatles", limit=3) - if not results: - pytest.skip("No search results available") - - track = results[0] - track_id = track['id'] - - # Get track info - info = shared_client.get_track_info(track_id) - # info might be None if endpoint not available - - # Get stream URL - stream = shared_client.get_stream_url(track_id, quality='lossless') - if not stream: - pytest.skip("Stream URL not available") - - assert stream['url'].startswith('http') - assert stream['quality'] == 'lossless' - - def test_98_live_download_small_segment(self, shared_client, temp_download_dir): - """Download the first 100KB of a track to verify the URL works.""" - results = shared_client.search_tracks(title="Yesterday", artist="Beatles", limit=1) - if not results: - pytest.skip("No search results") - - stream = shared_client.get_stream_url(results[0]['id'], quality='lossless') - if not stream: - pytest.skip("No stream URL") - - import requests - try: - resp = requests.get(stream['url'], stream=True, timeout=10, - headers={'Range': 'bytes=0-102400'}) - data = resp.content - assert len(data) > 1000, f"Expected >1KB of audio data, got {len(data)} bytes" - - # Check for FLAC magic bytes - if stream.get('codec', '') and 'flac' in stream['codec'].lower(): - assert data[:4] == b'fLaC', "FLAC file should start with 'fLaC' magic bytes" - except requests.exceptions.RequestException as e: - pytest.skip(f"CDN request failed: {e}") - - def test_99_live_full_download_lifecycle(self, temp_download_dir): - """Full download lifecycle: search → download → verify file.""" - client = HiFiClient(download_path=temp_download_dir) - - results = client.search_tracks(title="Yesterday", artist="Beatles", limit=1) - if not results: - pytest.skip("No search results") - - track = results[0] - filename = f"{track['id']}||{track['artist']} - {track['title']}" - - download_id = run_async(client.download('hifi', filename)) - if not download_id: - pytest.skip("Download failed to start") - - # Wait for download to complete (max 60s) - for _ in range(120): - time.sleep(0.5) - status = run_async(client.get_download_status(download_id)) - if not status: - break - if status.state in ('Completed, Succeeded', 'Errored', 'Cancelled'): - break - - status = run_async(client.get_download_status(download_id)) - if status and status.state == 'Completed, Succeeded': - assert status.file_path is not None - assert os.path.exists(status.file_path) - file_size = os.path.getsize(status.file_path) - assert file_size > 100 * 1024, f"Downloaded file too small: {file_size} bytes" - # Cleanup - os.unlink(status.file_path) - else: - # Download may have failed due to geo/availability — not a test failure - state = status.state if status else 'unknown' - pytest.skip(f"Download did not complete successfully (state: {state})") - - def test_100_live_download_cancel_cleans_up(self, temp_download_dir): - """Cancelling a live download should update state.""" - client = HiFiClient(download_path=temp_download_dir) - - results = client.search_tracks(title="Stairway to Heaven", artist="Led Zeppelin", limit=1) - if not results: - pytest.skip("No search results") - - track = results[0] - filename = f"{track['id']}||{track['artist']} - {track['title']}" - - download_id = run_async(client.download('hifi', filename)) - if not download_id: - pytest.skip("Download failed to start") - - # Brief pause then cancel - time.sleep(1) - result = run_async(client.cancel_download(download_id)) - assert result is True - - status = run_async(client.get_download_status(download_id)) - if status: - assert status.state == 'Cancelled' - - -# ===================== Bonus: Edge Case & Regression Tests ===================== - -class TestEdgeCases: - """Additional edge case and regression tests.""" - - def test_101_is_configured_with_instances(self, client): - """Client with instances should report as configured.""" - assert client.is_configured() is True - - def test_102_is_configured_no_instances(self, temp_download_dir): - """Client with no current instance should report not configured.""" - c = HiFiClient(download_path=temp_download_dir) - c._current_instance = None - assert c.is_configured() is False - - def test_103_check_connection_async(self, shared_client): - """check_connection should return True when instances are up.""" - result = run_async(shared_client.check_connection()) - assert result is True - - def test_104_empty_instances_list(self, temp_download_dir): - """Client with no instances should handle gracefully.""" - c = HiFiClient(download_path=temp_download_dir) - c._instances = [] - c._current_instance = None - - assert c.is_available() is False - assert c.search_tracks(title="test") == [] - - def test_105_api_get_error_response(self, client): - """API returning error dict should return None.""" - with patch.object(client.session, 'get') as mock_get: - mock_response = MagicMock() - mock_response.json.return_value = {'error': 'Something went wrong'} - mock_response.raise_for_status.return_value = None - mock_get.return_value = mock_response - - result = client._api_get('/test') - assert result is None - - def test_106_parse_track_with_track_number_alt_key(self, client): - """Should handle 'track_number' key (underscore variant).""" - item = {'id': 1, 'title': 'Song', 'track_number': 7, 'duration': 100} - result = client._parse_track(item) - assert result['track_number'] == 7 - - def test_107_download_path_type(self, client): - """Download path should be a Path object.""" - assert isinstance(client.download_path, Path) - - def test_108_download_returns_unique_ids(self, client): - """Each download should get a unique ID.""" - ids = set() - for i in range(5): - did = run_async(client.download('hifi', f'{i}||Test - Track {i}')) - if did: - ids.add(did) - assert len(ids) == 5 or len(ids) == 0 # All unique or all failed - - -if __name__ == '__main__': - pytest.main([__file__, '-v', '--tb=short', '-x']) diff --git a/tools/test_youtube_download.py b/tools/test_youtube_download.py deleted file mode 100644 index 59bbdbd6..00000000 --- a/tools/test_youtube_download.py +++ /dev/null @@ -1,1248 +0,0 @@ -#!/usr/bin/env python3 -""" -YouTube Download Test Client - Proof of Concept -Mirrors the Soulseek download flow but uses yt-dlp/YouTube instead. - -This is a standalone test file to evaluate YouTube as a fallback download source. -Does NOT modify any production code. - -Requirements: - pip install yt-dlp - -Usage: - python tools/test_youtube_download.py -""" - -import sys -import os -import json -import re -from typing import List, Optional, Dict, Any, Tuple -from dataclasses import dataclass -from pathlib import Path -from datetime import datetime - -# Fix Windows console encoding for emojis -if sys.platform == 'win32': - import io - sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') - sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace') - -# Try to import yt-dlp -try: - import yt_dlp -except ImportError: - print("yt-dlp not installed. Install with: pip install yt-dlp") - sys.exit(1) - -# Add parent directory to path to import from core -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - -from utils.logging_config import get_logger -from core.matching_engine import MusicMatchingEngine - -logger = get_logger("youtube_test") - - -@dataclass -class SpotifyTrack: - """Mock Spotify track (matches your actual SpotifyTrack)""" - id: str - name: str - artists: List[str] - album: str - duration_ms: int - popularity: int = 0 - - -@dataclass -class YouTubeSearchResult: - """YouTube search result - mirrors TrackResult structure""" - video_id: str - title: str - channel: str # Artist/uploader - duration: int # seconds - url: str - thumbnail: str - view_count: int - upload_date: str - - # Parsed metadata (attempted extraction from title) - parsed_artist: Optional[str] = None - parsed_title: Optional[str] = None - parsed_album: Optional[str] = None - - # Quality info (extracted from available formats) - available_quality: str = "unknown" # e.g., "256kbps AAC" - best_audio_format: Optional[Dict] = None - - # Confidence score (0.0 - 1.0) - confidence: float = 0.0 - match_reason: str = "" - - def __post_init__(self): - """Parse metadata from title""" - self._parse_title_metadata() - - def _parse_title_metadata(self): - """Extract artist and title from YouTube video title""" - # Common patterns: "Artist - Title", "Artist: Title", "Title by Artist" - patterns = [ - r'^(.+?)\s*[-–—]\s*(.+)$', # Artist - Title - r'^(.+?)\s*:\s*(.+)$', # Artist: Title - r'^(.+?)\s+by\s+(.+)$', # Title by Artist (reversed) - ] - - for pattern in patterns: - match = re.match(pattern, self.title, re.IGNORECASE) - if match: - if 'by' in pattern: # Reversed format - self.parsed_title = match.group(1).strip() - self.parsed_artist = match.group(2).strip() - else: - self.parsed_artist = match.group(1).strip() - self.parsed_title = match.group(2).strip() - break - - # Fallback: use entire title - if not self.parsed_title: - self.parsed_title = self.title - self.parsed_artist = self.channel - - -class YouTubeClient: - """ - YouTube download client using yt-dlp. - Mirrors SoulseekClient API structure for easy comparison. - """ - - def __init__(self, download_path: str = "./downloads/youtube"): - self.download_path = Path(download_path) - self.download_path.mkdir(parents=True, exist_ok=True) - - # Initialize production matching engine for parity with Soulseek - self.matching_engine = MusicMatchingEngine() - logger.info("Initialized production MusicMatchingEngine") - - # Check for ffmpeg (REQUIRED for MP3 conversion) - if not self._check_ffmpeg(): - print("\n" + "="*80) - print("ERROR: ffmpeg is required but not found in PATH") - print("="*80) - print("\nInstall ffmpeg:") - print(" Windows: scoop install ffmpeg") - print(" Linux: sudo apt install ffmpeg") - print(" Mac: brew install ffmpeg") - print("\nAfter installing, restart your terminal and try again.") - print("="*80 + "\n") - raise RuntimeError("ffmpeg is required for MP3 conversion but was not found in PATH") - - # yt-dlp options for search - self.search_opts = { - 'quiet': True, - 'no_warnings': True, - 'extract_flat': True, # Don't download during search - 'skip_download': True, - # Anti-bot measures - 'extractor_args': { - 'youtube': { - 'player_client': ['android', 'web'], - 'skip': ['hls', 'dash'] - } - }, - } - - # yt-dlp options for download - MP3 conversion with ffmpeg - self.download_opts = { - 'format': 'bestaudio/best', # Best audio quality - 'outtmpl': str(self.download_path / '%(artist)s - %(title)s.%(ext)s'), - 'quiet': False, - 'no_warnings': False, - 'progress_hooks': [self._download_progress_hook], - # Anti-bot measures for YouTube - 'extractor_args': { - 'youtube': { - 'player_client': ['android', 'web'], # Use mobile client to avoid restrictions - 'skip': ['hls', 'dash'] - } - }, - # Better user agent - 'http_headers': { - 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', - 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', - 'Accept-Language': 'en-us,en;q=0.5', - 'Sec-Fetch-Mode': 'navigate', - }, - # FFmpeg audio extraction to MP3 - 'postprocessors': [{ - 'key': 'FFmpegExtractAudio', - 'preferredcodec': 'mp3', - 'preferredquality': '320', - }], - # Don't download YouTube thumbnail - we'll get album art from Spotify - 'writethumbnail': False, - } - - self.current_download_progress = {} - - def _check_ffmpeg(self) -> bool: - """Check if ffmpeg is available (system PATH or auto-download to tools folder)""" - import shutil - import platform - import urllib.request - import zipfile - import tarfile - - # Check if ffmpeg is in system PATH - if shutil.which('ffmpeg'): - logger.info("Found ffmpeg in system PATH") - return True - - # Auto-download ffmpeg to tools folder if not found - tools_dir = Path(__file__).parent - system = platform.system().lower() - - if system == 'windows': - ffmpeg_path = tools_dir / 'ffmpeg.exe' - ffprobe_path = tools_dir / 'ffprobe.exe' - else: - ffmpeg_path = tools_dir / 'ffmpeg' - ffprobe_path = tools_dir / 'ffprobe' - - # If we already have both locally, use them - if ffmpeg_path.exists() and ffprobe_path.exists(): - logger.info(f"Found ffmpeg and ffprobe in tools folder") - # Add to PATH so yt-dlp can find them - import os - tools_dir_str = str(tools_dir.absolute()) - os.environ['PATH'] = tools_dir_str + os.pathsep + os.environ.get('PATH', '') - return True - - # Auto-download ffmpeg binary - logger.info(f"⬇️ ffmpeg not found - downloading for {system}...") - - try: - if system == 'windows': - # Download Windows ffmpeg (static build) - url = 'https://github.com/BtbN/FFmpeg-Builds/releases/download/latest/ffmpeg-master-latest-win64-gpl.zip' - zip_path = tools_dir / 'ffmpeg.zip' - - logger.info(f" Downloading from GitHub (this may take a minute)...") - urllib.request.urlretrieve(url, zip_path) - - logger.info(f" Extracting ffmpeg.exe and ffprobe.exe...") - with zipfile.ZipFile(zip_path, 'r') as zip_ref: - # Extract ffmpeg.exe and ffprobe.exe from the bin folder - for file in zip_ref.namelist(): - if file.endswith('bin/ffmpeg.exe'): - with zip_ref.open(file) as source, open(tools_dir / 'ffmpeg.exe', 'wb') as target: - target.write(source.read()) - elif file.endswith('bin/ffprobe.exe'): - with zip_ref.open(file) as source, open(tools_dir / 'ffprobe.exe', 'wb') as target: - target.write(source.read()) - - zip_path.unlink() # Clean up zip - - elif system == 'linux': - # Download Linux ffmpeg (static build) - url = 'https://github.com/BtbN/FFmpeg-Builds/releases/download/latest/ffmpeg-master-latest-linux64-gpl.tar.xz' - tar_path = tools_dir / 'ffmpeg.tar.xz' - - logger.info(f" Downloading from GitHub (this may take a minute)...") - urllib.request.urlretrieve(url, tar_path) - - logger.info(f" Extracting ffmpeg and ffprobe...") - with tarfile.open(tar_path, 'r:xz') as tar_ref: - for member in tar_ref.getmembers(): - if member.name.endswith('bin/ffmpeg'): - with tar_ref.extractfile(member) as source, open(tools_dir / 'ffmpeg', 'wb') as target: - target.write(source.read()) - (tools_dir / 'ffmpeg').chmod(0o755) # Make executable - elif member.name.endswith('bin/ffprobe'): - with tar_ref.extractfile(member) as source, open(tools_dir / 'ffprobe', 'wb') as target: - target.write(source.read()) - (tools_dir / 'ffprobe').chmod(0o755) # Make executable - - tar_path.unlink() # Clean up tar - - elif system == 'darwin': - # Download Mac ffmpeg and ffprobe (static builds) - logger.info(f" Downloading ffmpeg from evermeet.cx...") - ffmpeg_url = 'https://evermeet.cx/ffmpeg/getrelease/zip' - ffmpeg_zip = tools_dir / 'ffmpeg.zip' - urllib.request.urlretrieve(ffmpeg_url, ffmpeg_zip) - - logger.info(f" Downloading ffprobe from evermeet.cx...") - ffprobe_url = 'https://evermeet.cx/ffmpeg/getrelease/ffprobe/zip' - ffprobe_zip = tools_dir / 'ffprobe.zip' - urllib.request.urlretrieve(ffprobe_url, ffprobe_zip) - - logger.info(f" Extracting ffmpeg and ffprobe...") - with zipfile.ZipFile(ffmpeg_zip, 'r') as zip_ref: - zip_ref.extract('ffmpeg', tools_dir) - with zipfile.ZipFile(ffprobe_zip, 'r') as zip_ref: - zip_ref.extract('ffprobe', tools_dir) - - (tools_dir / 'ffmpeg').chmod(0o755) # Make executable - (tools_dir / 'ffprobe').chmod(0o755) # Make executable - - ffmpeg_zip.unlink() # Clean up zip - ffprobe_zip.unlink() # Clean up zip - - else: - logger.error(f"Unsupported platform: {system}") - return False - - logger.info(f"Downloaded ffmpeg to: {ffmpeg_path}") - - # Add to PATH - import os - tools_dir_str = str(tools_dir.absolute()) - os.environ['PATH'] = tools_dir_str + os.pathsep + os.environ.get('PATH', '') - - return True - - except Exception as e: - logger.error(f"Failed to download ffmpeg: {e}") - logger.error(f" Please install manually:") - logger.error(f" Windows: scoop install ffmpeg") - logger.error(f" Linux: sudo apt install ffmpeg") - logger.error(f" Mac: brew install ffmpeg") - return False - - def _download_progress_hook(self, d): - """Track download progress (like slskd monitoring)""" - if d['status'] == 'downloading': - total = d.get('total_bytes') or d.get('total_bytes_estimate', 0) - downloaded = d.get('downloaded_bytes', 0) - speed = d.get('speed') or 0 - eta = d.get('eta') or 0 - - if total > 0: - progress = (downloaded / total) * 100 - self.current_download_progress = { - 'status': 'downloading', - 'progress': progress, - 'downloaded': downloaded, - 'total': total, - 'speed': speed, - 'eta': eta - } - - # Format speed safely - speed_kb = speed / 1024 if speed else 0 - logger.info(f"Progress: {progress:.1f}% | Speed: {speed_kb:.1f} KB/s | ETA: {eta}s") - - elif d['status'] == 'finished': - self.current_download_progress = { - 'status': 'finished', - 'progress': 100.0, - 'filename': d.get('filename', '') - } - logger.info(f"Download finished: {d.get('filename', '')}") - - elif d['status'] == 'error': - self.current_download_progress = { - 'status': 'error', - 'error': str(d.get('error', 'Unknown error')) - } - logger.error(f"Download error: {d.get('error', '')}") - - def search(self, query: str, max_results: int = 10) -> List[YouTubeSearchResult]: - """ - Search YouTube for tracks (mirrors soulseek_client.search). - - Args: - query: Search query (e.g., "Artist - Track Name") - max_results: Maximum number of results to return - - Returns: - List of YouTubeSearchResult objects - """ - logger.info(f"Searching YouTube for: '{query}'") - - try: - # Use YouTube Music for better music search results - search_url = f"ytsearch{max_results}:{query}" - - with yt_dlp.YoutubeDL(self.search_opts) as ydl: - search_results = ydl.extract_info(search_url, download=False) - - if not search_results or 'entries' not in search_results: - logger.warning(f"No results found for: {query}") - return [] - - results = [] - for entry in search_results['entries']: - if not entry: - continue - - # Get detailed info for quality assessment - video_url = f"https://www.youtube.com/watch?v={entry['id']}" - try: - with yt_dlp.YoutubeDL({'quiet': True}) as detail_ydl: - video_info = detail_ydl.extract_info(video_url, download=False) - - # Find best audio format - best_audio = self._get_best_audio_format(video_info.get('formats', [])) - quality_str = self._format_quality_string(best_audio) - - result = YouTubeSearchResult( - video_id=entry['id'], - title=entry.get('title', 'Unknown'), - channel=entry.get('channel', entry.get('uploader', 'Unknown')), - duration=entry.get('duration', 0), - url=video_url, - thumbnail=entry.get('thumbnail', ''), - view_count=entry.get('view_count', 0), - upload_date=entry.get('upload_date', ''), - available_quality=quality_str, - best_audio_format=best_audio - ) - - results.append(result) - - except Exception as e: - logger.warning(f"Could not get detailed info for {entry['id']}: {e}") - continue - - logger.info(f"Found {len(results)} YouTube results") - return results - - except Exception as e: - logger.error(f"YouTube search error: {e}") - return [] - - def _get_best_audio_format(self, formats: List[Dict]) -> Optional[Dict]: - """Extract best audio format from available formats""" - audio_formats = [f for f in formats if f.get('acodec') != 'none' and f.get('vcodec') == 'none'] - - if not audio_formats: - # Fallback: get best format with audio - audio_formats = [f for f in formats if f.get('acodec') != 'none'] - - if not audio_formats: - return None - - # Sort by audio bitrate (tbr = total bitrate, abr = audio bitrate) - audio_formats.sort(key=lambda f: f.get('abr', f.get('tbr', 0)), reverse=True) - return audio_formats[0] - - def _format_quality_string(self, audio_format: Optional[Dict]) -> str: - """Format quality info string""" - if not audio_format: - return "unknown" - - abr = audio_format.get('abr', audio_format.get('tbr', 0)) - acodec = audio_format.get('acodec', 'unknown') - - if abr: - return f"{int(abr)}kbps {acodec.upper()}" - return acodec.upper() - - def calculate_match_confidence(self, spotify_track: SpotifyTrack, yt_result: YouTubeSearchResult) -> Tuple[float, str]: - """ - Calculate match confidence using PRODUCTION matching engine for parity with Soulseek. - - Returns: - (confidence_score, match_reason) tuple - """ - # Use production matching engine's normalization and similarity scoring - spotify_artist = spotify_track.artists[0] if spotify_track.artists else "" - yt_artist = yt_result.parsed_artist or yt_result.channel - - # Normalize using production engine - spotify_artist_clean = self.matching_engine.clean_artist(spotify_artist) - yt_artist_clean = self.matching_engine.clean_artist(yt_artist) - - spotify_title_clean = self.matching_engine.clean_title(spotify_track.name) - yt_title_clean = self.matching_engine.clean_title(yt_result.parsed_title) - - # Use production similarity_score (includes version detection, remaster penalties, etc.) - artist_similarity = self.matching_engine.similarity_score(spotify_artist_clean, yt_artist_clean) - title_similarity = self.matching_engine.similarity_score(spotify_title_clean, yt_title_clean) - - # Duration matching using production engine - spotify_duration_ms = spotify_track.duration_ms - yt_duration_ms = int(yt_result.duration * 1000) # Convert seconds to ms - duration_similarity = self.matching_engine.duration_similarity(spotify_duration_ms, yt_duration_ms) - - # Quality penalty (YouTube-specific) - quality_score = self._quality_score(yt_result.available_quality) - - # Weighted confidence calculation (similar to production Soulseek matching) - # Production uses: title * 0.5 + artist * 0.3 + duration * 0.2 - # Adjusted for YouTube: title * 0.4 + artist * 0.3 + duration * 0.2 + quality * 0.1 - confidence = ( - title_similarity * 0.40 + - artist_similarity * 0.30 + - duration_similarity * 0.20 + - quality_score * 0.10 - ) - - # Determine match reason - if confidence >= 0.8: - reason = "excellent_match" - elif confidence >= 0.65: - reason = "good_match" - elif confidence >= 0.58: # Match production threshold - reason = "acceptable_match" - else: - reason = "poor_match" - - # Bonus for official channels/verified - if 'vevo' in yt_artist.lower() or 'official' in yt_result.channel.lower(): - confidence = min(1.0, confidence + 0.05) - reason += "_official" - - logger.debug(f"Match confidence: {confidence:.2f} | Artist: {artist_similarity:.2f} | Title: {title_similarity:.2f} | Duration: {duration_similarity:.2f} | Quality: {quality_score:.2f}") - - return confidence, reason - - def _quality_score(self, quality_str: str) -> float: - """Score quality string (mirrors quality_score logic)""" - quality_lower = quality_str.lower() - - # Extract bitrate - bitrate_match = re.search(r'(\d+)kbps', quality_lower) - if bitrate_match: - bitrate = int(bitrate_match.group(1)) - - # Scoring based on bitrate - if bitrate >= 256: - return 1.0 - elif bitrate >= 192: - return 0.8 - elif bitrate >= 128: - return 0.6 - else: - return 0.4 - - # Codec-based scoring if no bitrate - if 'opus' in quality_lower: - return 0.9 - elif 'aac' in quality_lower: - return 0.7 - elif 'mp3' in quality_lower: - return 0.7 - - return 0.5 # Unknown quality - - def find_best_matches(self, spotify_track: SpotifyTrack, yt_results: List[YouTubeSearchResult], - min_confidence: float = 0.58) -> List[YouTubeSearchResult]: - """ - Find best YouTube matches for Spotify track (mirrors find_best_slskd_matches). - Uses production threshold of 0.58 for parity with Soulseek matching. - - Args: - spotify_track: Spotify track to match - yt_results: YouTube search results - min_confidence: Minimum confidence threshold (default: 0.58, same as production) - - Returns: - Sorted list of matches above confidence threshold - """ - matches = [] - - for yt_result in yt_results: - confidence, reason = self.calculate_match_confidence(spotify_track, yt_result) - yt_result.confidence = confidence - yt_result.match_reason = reason - - if confidence >= min_confidence: - matches.append(yt_result) - - # Sort by confidence (best first) - matches.sort(key=lambda r: r.confidence, reverse=True) - - logger.info(f"Found {len(matches)} matches above {min_confidence} confidence") - return matches - - def download(self, yt_result: YouTubeSearchResult, spotify_track: Optional[SpotifyTrack] = None) -> Optional[str]: - """ - Download YouTube video as audio with proper metadata tagging (mirrors soulseek download). - - Args: - yt_result: YouTube result to download - spotify_track: Optional Spotify track for metadata embedding - - Returns: - Path to downloaded file, or None if failed - """ - logger.info(f"Starting download: {yt_result.title}") - logger.info(f" Quality: {yt_result.available_quality}") - logger.info(f" URL: {yt_result.url}") - - try: - # Build download options - download_opts = self.download_opts.copy() - - # Get Spotify album details for proper folder structure and track numbering - track_number = 1 - disc_number = 1 - release_year = str(datetime.now().year) - album_artist = None - artist_genres = [] - - if spotify_track and spotify_track.id and not spotify_track.id.startswith('test'): - # Fetch full Spotify details to get track number, disc number, release date, genres - try: - sys.path.insert(0, str(Path(__file__).parent.parent)) - from core.spotify_client import SpotifyClient - - spotify_client = SpotifyClient() - if spotify_client.is_authenticated(): - track_details = spotify_client.get_track_details(spotify_track.id) - if track_details: - track_number = track_details.get('track_number', 1) - disc_number = track_details.get('disc_number', 1) - - # Use album artist if available, otherwise use track artist - album_data = track_details.get('album', {}) - if album_data.get('artists'): - album_artist = album_data['artists'][0] - - # Get actual release year from Spotify - release_date = album_data.get('release_date', '') - if release_date: - release_year = release_date.split('-')[0] # Extract year from YYYY-MM-DD - - # Get artist genres (for metadata parity with Soulseek flow) - try: - primary_artist = track_details.get('primary_artist') - if primary_artist: - artist_info = spotify_client.get_artist(primary_artist) - if artist_info and hasattr(artist_info, 'genres'): - artist_genres = artist_info.genres - except: - pass - - logger.info(f" Spotify track #{track_number} on album: {spotify_track.album} ({release_year})") - except Exception as e: - logger.warning(f" Could not fetch Spotify track details: {e}") - - # If we have Spotify metadata, use production file organization - if spotify_track: - artist = spotify_track.artists[0] if spotify_track.artists else yt_result.parsed_artist - title = spotify_track.name - album = spotify_track.album - - # Use album artist if found, otherwise use track artist - if not album_artist: - album_artist = artist - - # Create folder structure: $albumartist/$albumartist - $album/ - album_folder = self.download_path / album_artist / f"{album_artist} - {album}" - album_folder.mkdir(parents=True, exist_ok=True) - - # File naming: $track - $title (production format) - final_filename = f"{track_number:02d} - {title}" - - # Sanitize filename (remove invalid characters) - final_filename = re.sub(r'[<>:"/\\|?*]', '', final_filename) - - # Override output template with production folder structure - download_opts['outtmpl'] = str(album_folder / f'{final_filename}.%(ext)s') - - logger.info(f" Album folder: {album_artist}/{album_artist} - {album}/") - logger.info(f" Filename: {final_filename}.mp3") - - # Add metadata postprocessor with Spotify info - download_opts['postprocessor_args'] = { - 'ffmpeg': [ - '-metadata', f'artist={artist}', - '-metadata', f'title={title}', - '-metadata', f'album={album}', - '-metadata', f'album_artist={album_artist}', - '-metadata', f'track={track_number}/{spotify_track.total_tracks if hasattr(spotify_track, "total_tracks") else track_number}', - '-metadata', f'disc={disc_number}', - '-metadata', f'date={release_year}', - '-metadata', 'comment=Downloaded via SoulSync (YouTube fallback)', - ] - } - - # Perform download - with yt_dlp.YoutubeDL(download_opts) as ydl: - info = ydl.extract_info(yt_result.url, download=True) - - # Get final filename (will be MP3 after ffmpeg conversion) - filename = Path(ydl.prepare_filename(info)).with_suffix('.mp3') - - if filename.exists(): - logger.info(f"Download successful: {filename}") - - # Post-download: Enhance metadata with mutagen - album_art_url = self._enhance_metadata(str(filename), spotify_track, yt_result, track_number, disc_number, release_year, artist_genres) - - # Save cover.jpg to album folder (like production) - if album_art_url and spotify_track: - self._save_cover_art(filename.parent, album_art_url) - - # Create .lrc lyrics file (like production) - if spotify_track: - self._create_lyrics_file(str(filename), spotify_track) - - return str(filename) - else: - logger.error(f"Download completed but file not found: {filename}") - return None - - except Exception as e: - logger.error(f"Download failed: {e}") - import traceback - traceback.print_exc() - return None - - def _enhance_metadata(self, filepath: str, spotify_track: Optional[SpotifyTrack], yt_result: YouTubeSearchResult, track_number: int = 1, disc_number: int = 1, release_year: str = None, artist_genres: list = None): - """ - Enhance MP3 metadata using mutagen + Spotify album art (mirrors main app's metadata enhancement). - Uses full Spotify metadata including disc number, actual release year, and genre tags. - """ - try: - from mutagen.mp3 import MP3 - from mutagen.id3 import ID3, TIT2, TPE1, TALB, TDRC, COMM, APIC, TRCK, TPE2, TPOS, TCON - from mutagen.id3 import ID3NoHeaderError - import requests - - logger.info(f"Enhancing metadata for: {Path(filepath).name}") - - # Load MP3 file - audio = MP3(filepath) - - # Clear ALL existing tags and start fresh - if audio.tags is not None: - # Delete ALL existing frames - audio.tags.clear() - logger.info(f" Cleared all existing tag frames") - else: - # No tags exist, add them - audio.add_tags() - logger.info(f" Added new tag structure") - - if spotify_track: - # Use Spotify metadata - artist = spotify_track.artists[0] if spotify_track.artists else "Unknown Artist" - title = spotify_track.name - album = spotify_track.album - year = release_year or str(datetime.now().year) - - # Get album artist from Spotify - album_artist = artist - try: - if spotify_track.id and not spotify_track.id.startswith('test'): - from core.spotify_client import SpotifyClient - spotify_client = SpotifyClient() - if spotify_client.is_authenticated(): - track_details = spotify_client.get_track_details(spotify_track.id) - if track_details: - album_data = track_details.get('album', {}) - if album_data.get('artists'): - album_artist = album_data['artists'][0] - except: - pass - - logger.info(f" Setting metadata tags...") - - # Set ID3 tags (using setall to ensure they're set) - audio.tags.setall('TIT2', [TIT2(encoding=3, text=title)]) - audio.tags.setall('TPE1', [TPE1(encoding=3, text=artist)]) - audio.tags.setall('TPE2', [TPE2(encoding=3, text=album_artist)]) # Album artist - audio.tags.setall('TALB', [TALB(encoding=3, text=album)]) - audio.tags.setall('TRCK', [TRCK(encoding=3, text=str(track_number))]) # Track number - audio.tags.setall('TPOS', [TPOS(encoding=3, text=str(disc_number))]) # Disc number - audio.tags.setall('TDRC', [TDRC(encoding=3, text=year)]) - - # Genre (from Spotify artist data - matches production flow) - if artist_genres: - if len(artist_genres) == 1: - genre = artist_genres[0] - else: - # Combine up to 3 genres (matches production logic) - genre = ', '.join(artist_genres[:3]) - audio.tags.setall('TCON', [TCON(encoding=3, text=genre)]) - logger.info(f" Genre: {genre}") - - audio.tags.setall('COMM', [COMM(encoding=3, lang='eng', desc='', - text=f'Downloaded via SoulSync (YouTube)\nSource: {yt_result.url}\nConfidence: {yt_result.confidence:.2f}')]) - - logger.info(f" Artist: {artist}") - logger.info(f" Album Artist: {album_artist}") - logger.info(f" Title: {title}") - logger.info(f" Album: {album}") - logger.info(f" Track #: {track_number}") - logger.info(f" Disc #: {disc_number}") - logger.info(f" Year: {year}") - - # Fetch and embed album art from Spotify (via search) - logger.info(f" Fetching album art from Spotify...") - album_art_url = self._get_spotify_album_art(spotify_track) - - if album_art_url: - try: - # Download album art - response = requests.get(album_art_url, timeout=10) - response.raise_for_status() - - # Determine image type - if 'jpeg' in response.headers.get('Content-Type', ''): - mime_type = 'image/jpeg' - elif 'png' in response.headers.get('Content-Type', ''): - mime_type = 'image/png' - else: - mime_type = 'image/jpeg' # Default - - # Embed album art - audio.tags.add(APIC( - encoding=3, - mime=mime_type, - type=3, # Cover (front) - desc='Cover', - data=response.content - )) - - logger.info(f" Album art embedded ({len(response.content) // 1024} KB)") - except Exception as art_error: - logger.warning(f" Could not embed album art: {art_error}") - else: - logger.warning(f" No album art found on Spotify") - - # Save all tags - audio.save() - logger.info(f"Metadata enhanced successfully") - - # Return album art URL for cover.jpg creation - return album_art_url - - except ImportError: - logger.warning("mutagen not installed - skipping enhanced metadata tagging") - logger.warning(" Install with: pip install mutagen") - return None - except Exception as e: - logger.warning(f"Could not enhance metadata: {e}") - import traceback - traceback.print_exc() - return None - - def _get_spotify_album_art(self, spotify_track: SpotifyTrack) -> Optional[str]: - """ - Fetch album art URL from Spotify using your existing Spotify client. - Returns URL to highest quality album art. - """ - try: - # Import your existing Spotify client - sys.path.insert(0, str(Path(__file__).parent.parent)) - from core.spotify_client import SpotifyClient - - logger.info(f" Getting Spotify client...") - - # Get authenticated Spotify client - spotify_client = SpotifyClient() - - if not spotify_client: - logger.warning(f" Spotify client not available") - return None - - if not spotify_client.is_authenticated(): - logger.warning(f" Spotify client not authenticated") - return None - - logger.info(f" Spotify client authenticated") - - # Use the track ID if available (real Spotify IDs) - if spotify_track.id and not spotify_track.id.startswith('test'): - logger.info(f" Fetching track info for ID: {spotify_track.id}") - try: - # Get track info from Spotify API - track_info = spotify_client.sp.track(spotify_track.id) - - if track_info: - logger.info(f" Got track info from Spotify") - - if 'album' in track_info: - album_images = track_info['album'].get('images', []) - logger.info(f" Found {len(album_images)} album images") - - if album_images: - # Get highest quality image (first in list) - album_art_url = album_images[0]['url'] - logger.info(f" Album art URL: {album_art_url[:50]}...") - return album_art_url - else: - logger.warning(f" No album data in track info") - else: - logger.warning(f" Track info is empty") - - except Exception as e: - logger.warning(f" Error fetching via track ID: {e}") - import traceback - traceback.print_exc() - - # Fallback: Search for the track - query = f"track:{spotify_track.name} artist:{spotify_track.artists[0]}" - logger.info(f" Searching Spotify: {query}") - - try: - search_results = spotify_client.sp.search(q=query, type='track', limit=1) - - if search_results and 'tracks' in search_results: - tracks = search_results['tracks'].get('items', []) - logger.info(f" Search returned {len(tracks)} tracks") - - if tracks: - album_images = tracks[0].get('album', {}).get('images', []) - if album_images: - # Get highest quality image (first in list) - album_art_url = album_images[0]['url'] - logger.info(f" Found via search: {album_art_url[:50]}...") - return album_art_url - except Exception as search_error: - logger.warning(f" Search error: {search_error}") - import traceback - traceback.print_exc() - - logger.warning(f" No album art found on Spotify") - return None - - except ImportError as e: - logger.warning(f" Could not import Spotify client: {e}") - import traceback - traceback.print_exc() - return None - except Exception as e: - logger.warning(f" Error fetching album art: {e}") - import traceback - traceback.print_exc() - return None - - def _save_cover_art(self, album_folder: Path, album_art_url: str): - """ - Save cover.jpg to album folder (mirrors production _download_cover_art). - """ - try: - import requests - - cover_path = album_folder / "cover.jpg" - - # Skip if already exists - if cover_path.exists(): - logger.info(f" cover.jpg already exists") - return - - logger.info(f" Downloading cover.jpg to album folder...") - - # Download album art - response = requests.get(album_art_url, timeout=10) - response.raise_for_status() - - # Save to file - cover_path.write_bytes(response.content) - - logger.info(f" Saved cover.jpg ({len(response.content) // 1024} KB)") - - except Exception as e: - logger.warning(f" Could not save cover.jpg: {e}") - - def _create_lyrics_file(self, audio_file_path: str, spotify_track: SpotifyTrack): - """ - Create .lrc lyrics file using LRClib API (mirrors production lyrics flow). - """ - try: - # Import lyrics client - sys.path.insert(0, str(Path(__file__).parent.parent)) - from core.lyrics_client import lyrics_client - - if not lyrics_client.api: - logger.debug(f" LRClib API not available - skipping lyrics") - return - - logger.info(f" Fetching lyrics from LRClib...") - - # Get track metadata - artist_name = spotify_track.artists[0] if spotify_track.artists else "Unknown Artist" - track_name = spotify_track.name - album_name = spotify_track.album - duration_seconds = int(spotify_track.duration_ms / 1000) if spotify_track.duration_ms else None - - # Create LRC file - success = lyrics_client.create_lrc_file( - audio_file_path=audio_file_path, - track_name=track_name, - artist_name=artist_name, - album_name=album_name, - duration_seconds=duration_seconds - ) - - if success: - logger.info(f" Created .lrc lyrics file") - else: - logger.info(f" No lyrics found on LRClib") - - except ImportError: - logger.debug(f" lyrics_client not available - skipping lyrics") - except Exception as e: - logger.warning(f" Could not create lyrics file: {e}") - - def search_and_download_best(self, spotify_track: SpotifyTrack, min_confidence: float = 0.58) -> Optional[str]: - """ - Complete flow: search, find best match, download (mirrors soulseek flow). - Uses production threshold of 0.58 for parity with Soulseek matching. - - Args: - spotify_track: Spotify track to download - min_confidence: Minimum confidence threshold (default: 0.58, same as production) - - Returns: - Path to downloaded file, or None if failed - """ - logger.info(f"Starting YouTube download flow for: {spotify_track.name} by {spotify_track.artists[0]}") - - # Generate search query - query = f"{spotify_track.artists[0]} {spotify_track.name}" - - # Search YouTube - results = self.search(query, max_results=10) - - if not results: - logger.error(f"No YouTube results found for query: {query}") - return None - - # Find best matches - matches = self.find_best_matches(spotify_track, results, min_confidence=min_confidence) - - if not matches: - logger.error(f"No matches above {min_confidence} confidence threshold") - return None - - # Try downloading best match - best_match = matches[0] - logger.info(f"Best match: {best_match.title} (confidence: {best_match.confidence:.2f})") - - downloaded_file = self.download(best_match, spotify_track) - - return downloaded_file - - -# ============================================================================ -# TEST RUNNER -# ============================================================================ - -def test_youtube_download(): - """Test the YouTube download flow with a curated playlist""" - - print("=" * 80) - print("YouTube Download Test - Curated Playlist (5 Tracks)") - print("=" * 80) - print() - - # Initialize YouTube client - yt_client = YouTubeClient(download_path="./downloads/youtube_test") - - # Curated playlist - 5 diverse tracks from 5 different artists - # REAL Spotify track IDs for proper album art fetching - test_playlist = [ - SpotifyTrack( - id="5CQ30WqJwcep0pYcV4AMNc", # Real Spotify ID for Stairway to Heaven - name="Stairway to Heaven", - artists=["Led Zeppelin"], - album="Led Zeppelin IV (Remaster)", - duration_ms=482830, # ~8:03 - popularity=82 - ), - SpotifyTrack( - id="39LLxExYz6ewLAcYrzQQyP", # Real Spotify ID for Levitating - name="Levitating", - artists=["Dua Lipa"], - album="Future Nostalgia", - duration_ms=203064, # ~3:23 - popularity=88 - ), - SpotifyTrack( - id="7KXjTSCq5nL1LoYtL7XAwS", # Real Spotify ID for HUMBLE. - name="HUMBLE.", - artists=["Kendrick Lamar"], - album="DAMN.", - duration_ms=177000, # ~2:57 - popularity=89 - ), - SpotifyTrack( - id="0DiWol3AO6WpXZgp0goxAV", # Real Spotify ID for One More Time - name="One More Time", - artists=["Daft Punk"], - album="Discovery", - duration_ms=320357, # ~5:20 - popularity=85 - ), - SpotifyTrack( - id="63OQupATfueTdZMWTxW03A", # Real Spotify ID for Karma Police - name="Karma Police", - artists=["Radiohead"], - album="OK Computer", - duration_ms=263893, # ~4:24 - popularity=80 - ), - ] - - print("Curated Playlist - Testing Diverse Genres:") - print() - for i, track in enumerate(test_playlist, 1): - print(f" {i}. {track.artists[0]:20s} - {track.name}") - print() - print("This playlist tests:") - print(" • Classic Rock (Led Zeppelin)") - print(" • Modern Pop (Dua Lipa)") - print(" • Hip Hop (Kendrick Lamar)") - print(" • Electronic (Daft Punk)") - print(" • Alternative (Radiohead)") - print() - print("─" * 80) - print() - - # Auto-download all tracks - results = [] - total_start_time = datetime.now() - - for idx, track in enumerate(test_playlist, 1): - print(f"\n{'='*80}") - print(f"Track {idx}/{len(test_playlist)}: {track.name} by {track.artists[0]}") - print(f"{'='*80}\n") - - track_start_time = datetime.now() - - try: - # Use the complete search and download flow (production threshold: 0.58) - downloaded_file = yt_client.search_and_download_best( - spotify_track=track, - min_confidence=0.58 - ) - - track_end_time = datetime.now() - track_duration = (track_end_time - track_start_time).total_seconds() - - if downloaded_file: - # Get file size - file_size = Path(downloaded_file).stat().st_size / (1024 * 1024) # MB - - results.append({ - 'track': track.name, - 'artist': track.artists[0], - 'album': track.album, - 'file': downloaded_file, - 'file_size_mb': file_size, - 'duration_seconds': track_duration, - 'success': True, - 'error': None - }) - - print(f"\nSUCCESS - Downloaded in {track_duration:.1f}s") - print(f" File: {Path(downloaded_file).name}") - print(f" Size: {file_size:.2f} MB") - else: - results.append({ - 'track': track.name, - 'artist': track.artists[0], - 'album': track.album, - 'file': None, - 'file_size_mb': 0, - 'duration_seconds': track_duration, - 'success': False, - 'error': 'Download failed or no matches found' - }) - - print(f"\nFAILED - No suitable match found") - - except Exception as e: - track_end_time = datetime.now() - track_duration = (track_end_time - track_start_time).total_seconds() - - print(f"\nERROR: {e}") - import traceback - traceback.print_exc() - - results.append({ - 'track': track.name, - 'artist': track.artists[0], - 'album': track.album, - 'file': None, - 'file_size_mb': 0, - 'duration_seconds': track_duration, - 'success': False, - 'error': str(e) - }) - - total_end_time = datetime.now() - total_duration = (total_end_time - total_start_time).total_seconds() - - # ======================================================================== - # Final Summary Report - # ======================================================================== - - print("\n\n" + "=" * 80) - print("FINAL SUMMARY REPORT") - print("=" * 80) - print() - - successful = [r for r in results if r['success']] - failed = [r for r in results if not r['success']] - - print(f"⏱️ Total Time: {total_duration:.1f}s ({total_duration/60:.1f} minutes)") - print(f"Success Rate: {len(successful)}/{len(results)} ({len(successful)/len(results)*100:.1f}%)") - print() - - if successful: - total_size = sum(r['file_size_mb'] for r in successful) - avg_time = sum(r['duration_seconds'] for r in successful) / len(successful) - - print("─" * 80) - print("SUCCESSFUL DOWNLOADS:") - print("─" * 80) - for i, result in enumerate(successful, 1): - print(f"\n{i}. {result['artist']} - {result['track']}") - print(f" Album: {result['album']}") - print(f" File: {Path(result['file']).name}") - print(f" Size: {result['file_size_mb']:.2f} MB") - print(f" Time: {result['duration_seconds']:.1f}s") - - print() - print(f"Total Downloaded: {total_size:.2f} MB") - print(f"Average Download Time: {avg_time:.1f}s per track") - print() - - if failed: - print("─" * 80) - print("FAILED DOWNLOADS:") - print("─" * 80) - for i, result in enumerate(failed, 1): - print(f"\n{i}. {result['artist']} - {result['track']}") - print(f" Reason: {result['error']}") - print() - - # File list for easy access - if successful: - print("─" * 80) - print("DOWNLOAD LOCATION:") - print("─" * 80) - print(f"\n{yt_client.download_path.absolute()}\n") - print("Files:") - for result in successful: - print(f" • {Path(result['file']).name}") - print() - - print("=" * 80) - print("Test Complete!") - print("=" * 80) - print() - - # Quality check reminder - if successful: - print("Next Steps:") - print(" 1. Listen to downloaded files to verify quality") - print(" 2. Check metadata tags in your music player") - print(" 3. Compare with Soulseek downloads if available") - print(" 4. Review confidence scores vs actual match quality") - print() - - return results - - -if __name__ == "__main__": - test_youtube_download()