danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Merge pull request #311 from kettui/feat/remove-desktop-app

Browse source 
Remove desktop app, clean up test files, remove unused dependencies
This commit is contained in:
BoulderBadgeDad 2026-04-17 12:52:10 -07:00 committed by GitHub
commit 0cf5cbe9cd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
32 changed files with 77 additions and 42836 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
Dockerfile
View file

@ -15,9 +15,9 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
COPY requirements-webui.txt .
COPY requirements.txt .
RUN pip install --no-cache-dir --upgrade pip && \
pip install --no-cache-dir -r requirements-webui.txt
pip install --no-cache-dir -r requirements.txt
# Stage 2: Runtime — only runtime dependencies, no build tools
FROM python:3.11-slim

2
README.md
View file

@ -219,7 +219,7 @@ PUID/PGID are exposed in the template — set them to match your Unraid permissi
```bash
git clone https://github.com/Nezreka/SoulSync
cd SoulSync
pip install -r requirements-webui.txt
pip install -r requirements.txt
python web_server.py
# Open http://localhost:8008
```

154
core/database_update_worker.py
View file

@ -1,35 +1,5 @@
#!/usr/bin/env python3
# Conditional PyQt6 import for backward compatibility with GUI version
try:
from PyQt6.QtCore import QThread, pyqtSignal
QT_AVAILABLE = True
except ImportError:
QT_AVAILABLE = False
# Define dummy classes for headless operation
class QThread:
def __init__(self):
self.callbacks = {}
def start(self):
import threading
self.thread = threading.Thread(target=self.run)
self.thread.daemon = True
self.thread.start()
def wait(self):
if hasattr(self, 'thread'):
self.thread.join()
def emit_signal(self, signal_name, *args):
if signal_name in self.callbacks:
for callback in self.callbacks[signal_name]:
try:
callback(*args)
except Exception as e:
logger.error(f"Error in callback for {signal_name}: {e}")
def connect_signal(self, signal_name, callback):
if signal_name not in self.callbacks:
self.callbacks[signal_name] = []
self.callbacks[signal_name].append(callback)
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Optional, List, Callable
@ -42,32 +12,19 @@ from config.settings import config_manager
logger = get_logger("database_update_worker")
class DatabaseUpdateWorker(QThread):
"""Worker thread for updating SoulSync database with media server library data (Plex or Jellyfin)"""
# Qt signals (only available when PyQt6 is installed)
if QT_AVAILABLE:
progress_updated = pyqtSignal(str, int, int, float) # current_item, processed, total, percentage
artist_processed = pyqtSignal(str, bool, str, int, int) # artist_name, success, details, albums_count, tracks_count
finished = pyqtSignal(int, int, int, int, int) # total_artists, total_albums, total_tracks, successful, failed
error = pyqtSignal(str) # error_message
phase_changed = pyqtSignal(str) # current_phase (artists, albums, tracks)
class DatabaseUpdateWorker:
"""Worker for updating SoulSync database with media server library data."""
def __init__(self, media_client, database_path: str = "database/music_library.db", full_refresh: bool = False, server_type: str = "plex", force_sequential: bool = False):
super().__init__()
# Force sequential processing for web server mode to avoid threading issues
self.force_sequential = force_sequential
# Initialize signal callbacks for headless mode
if not QT_AVAILABLE:
self.callbacks = {
'progress_updated': [],
'artist_processed': [],
'finished': [],
'error': [],
'phase_changed': []
}
self.callbacks = {
'progress_updated': [],
'artist_processed': [],
'finished': [],
'error': [],
'phase_changed': [],
}
# Support both old plex_client parameter and new media_client parameter for backward compatibility
if hasattr(media_client, '__class__') and 'plex' in media_client.__class__.__name__.lower():
@ -116,19 +73,16 @@ class DatabaseUpdateWorker(QThread):
self.database: Optional[MusicDatabase] = None
def _emit_signal(self, signal_name: str, *args):
"""Emit a signal in both Qt and headless modes"""
if QT_AVAILABLE and hasattr(self, signal_name):
# Qt mode - use actual signal
getattr(self, signal_name).emit(*args)
elif not QT_AVAILABLE:
# Headless mode - use callback system
self.emit_signal(signal_name, *args)
"""Emit a signal through the callback registry."""
for callback in self.callbacks.get(signal_name, []):
try:
callback(*args)
except Exception as e:
logger.error(f"Error in callback for {signal_name}: {e}")
def connect_callback(self, signal_name: str, callback: Callable):
"""Connect a callback for headless mode"""
if not QT_AVAILABLE:
self.connect_signal(signal_name, callback)
# In Qt mode, use the normal signal.connect() method
"""Connect a callback for progress notifications."""
self.callbacks.setdefault(signal_name, []).append(callback)
def stop(self):
"""Stop the database update process"""
@ -1261,8 +1215,8 @@ class DatabaseUpdateWorker(QThread):
logger.error(f"Error processing artist {getattr(artist, 'title', 'Unknown')}: {e}")
return (getattr(artist, 'title', 'Unknown'), False, f"Error: {str(e)}", 0, 0)
# Process artists - use sequential processing in web server mode to avoid threading issues
if not QT_AVAILABLE or self.force_sequential:
# Process artists sequentially when requested (the web server uses this path).
if self.force_sequential:
# Sequential processing for web server mode
for i, artist in enumerate(artists):
if self.should_stop:
@ -1277,7 +1231,7 @@ class DatabaseUpdateWorker(QThread):
# Emit progress signal
self._emit_signal('artist_processed', artist_name, success, details, album_count, track_count)
else:
# Process artists in parallel using ThreadPoolExecutor (Qt mode only)
# Parallel processing for local/manual runs
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
# Submit all tasks
future_to_artist = {executor.submit(process_single_artist, artist): artist
@ -1409,71 +1363,3 @@ class DatabaseUpdateWorker(QThread):
except Exception as e:
logger.error(f"Error in run_with_callback: {e}")
class DatabaseStatsWorker(QThread):
"""Simple worker for getting database statistics without blocking UI"""
# Qt signals (only available when PyQt6 is installed)
if QT_AVAILABLE:
stats_updated = pyqtSignal(dict) # Database statistics
def __init__(self, database_path: str = "database/music_library.db"):
super().__init__()
self.database_path = database_path
self.should_stop = False
# Initialize signal callbacks for headless mode
if not QT_AVAILABLE:
self.callbacks = {
'stats_updated': []
}
def stop(self):
"""Stop the worker"""
self.should_stop = True
def _emit_signal(self, signal_name: str, *args):
"""Emit a signal in both Qt and headless modes"""
if QT_AVAILABLE and hasattr(self, signal_name):
# Qt mode - use actual signal
getattr(self, signal_name).emit(*args)
elif not QT_AVAILABLE:
# Headless mode - use callback system
self.emit_signal(signal_name, *args)
def connect_callback(self, signal_name: str, callback: Callable):
"""Connect a callback for headless mode"""
if not QT_AVAILABLE:
self.connect_signal(signal_name, callback)
# In Qt mode, use the normal signal.connect() method
def run(self):
"""Get database statistics and full info including last refresh"""
try:
if self.should_stop:
return
database = get_database(self.database_path)
if self.should_stop:
return
# Get database info for active server (server-aware statistics)
info = database.get_database_info_for_server()
if not self.should_stop:
self._emit_signal('stats_updated', info)
except Exception as e:
logger.error(f"Error getting database stats: {e}")
if not self.should_stop:
# Import here to avoid circular imports
from config.settings import config_manager
active_server = config_manager.get_active_media_server()
self._emit_signal('stats_updated', {
'artists': 0,
'albums': 0,
'tracks': 0,
'database_size_mb': 0.0,
'last_update': None,
'last_full_refresh': None,
'server_source': active_server
})

389
core/media_scan_manager.py
View file

@ -1,389 +0,0 @@
#!/usr/bin/env python3
import threading
import time
from utils.logging_config import get_logger
logger = get_logger("media_scan_manager")
class MediaScanManager:
"""
Smart media library scan manager with debouncing and scan-aware follow-up logic.
Supports both Plex and Jellyfin servers based on active configuration.
Features:
- Debounces multiple scan requests to prevent spam
- Tracks downloads that happen during active scans
- Automatically triggers follow-up scans when needed
- Thread-safe operation
- Works with both Plex and Jellyfin
"""
def __init__(self, delay_seconds: int = 60):
"""
Initialize the scan manager.
Args:
delay_seconds: Debounce delay in seconds (default 60s)
"""
self.delay = delay_seconds
self._timer = None
self._scan_in_progress = False
self._downloads_during_scan = False
self._lock = threading.Lock()
self._scan_completion_callbacks = [] # List of callback functions to call when scan completes
self._scan_start_time = None # Track when scan started for timeout
self._max_scan_time = 1800 # Maximum scan time in seconds (30 minutes)
# New periodic update system
self._periodic_update_timer = None # Timer for 5-minute periodic updates
self._periodic_update_interval = 300 # 5 minutes in seconds
self._is_doing_periodic_updates = False # Track if we're in periodic update mode
self._shutting_down = False
logger.info(f"MediaScanManager initialized with {delay_seconds}s debounce delay")
def _get_active_media_client(self):
"""Get the active media client based on config settings"""
try:
from config.settings import config_manager
active_server = config_manager.get_active_media_server()
# Try to get client instances from app
try:
# Try PyQt6 first (GUI mode)
try:
from PyQt6.QtWidgets import QApplication
app = QApplication.instance()
if app:
# Try to find the main window from top-level widgets
main_window = None
for widget in app.topLevelWidgets():
if (hasattr(widget, 'plex_client') and hasattr(widget, 'jellyfin_client') and
hasattr(widget, 'navidrome_client')):
main_window = widget
break
if main_window:
server_attr_map = {
'jellyfin': 'jellyfin_client',
'navidrome': 'navidrome_client',
'plex': 'plex_client'
}
client_attr = server_attr_map.get(active_server)
if client_attr:
client = getattr(main_window, client_attr, None)
if client and client.is_connected():
return client, active_server
else:
logger.warning(f"{active_server.title()} client not connected — scan skipped")
else:
logger.debug("No main window found in Qt application")
else:
logger.debug("No QApplication instance found")
except ImportError:
logger.debug("PyQt6 not available, trying headless mode")
# Headless mode - try to get clients from global instances
import sys
server_attr_map = {
'jellyfin': 'jellyfin_client',
'navidrome': 'navidrome_client',
'plex': 'plex_client'
}
client_attr = server_attr_map.get(active_server)
if client_attr:
for module_name, module in sys.modules.items():
if hasattr(module, client_attr):
client = getattr(module, client_attr, None)
if client and hasattr(client, 'is_connected') and client.is_connected():
return client, active_server
break
except Exception as e:
logger.debug(f"Could not access clients: {e}")
logger.error("No active media client available")
return None, None
except Exception as e:
logger.error(f"Error determining active media server: {e}")
return None, None
def request_scan(self, reason: str = "Download completed"):
"""
Request a library scan with smart debouncing logic.
Args:
reason: Optional reason for the scan request (for logging)
"""
logger.info(f"DEBUG: Media scan requested - reason: {reason}")
with self._lock:
if self._shutting_down:
logger.debug("Media scan request ignored during shutdown")
return
if self._scan_in_progress:
# Server is currently scanning - mark that we need another scan later
self._downloads_during_scan = True
logger.info(f"Media scan in progress - queueing follow-up scan ({reason})")
return
# Cancel any existing timer and start a new one
if self._timer:
self._timer.cancel()
logger.debug(f"Resetting scan timer ({reason})")
else:
logger.info(f"Media scan queued - will execute in {self.delay}s ({reason})")
# Start the debounce timer
self._timer = threading.Timer(self.delay, self._execute_scan)
self._timer.daemon = True
self._timer.start()
def add_scan_completion_callback(self, callback):
"""
Add a callback function to be called when scan completes.
Args:
callback: Function to call when scan completes (no arguments)
"""
with self._lock:
if callback not in self._scan_completion_callbacks:
self._scan_completion_callbacks.append(callback)
logger.info(f"DEBUG: Added scan completion callback: {callback.__name__}")
logger.info(f"DEBUG: Total callbacks registered: {len(self._scan_completion_callbacks)}")
def remove_scan_completion_callback(self, callback):
"""
Remove a previously registered callback.
Args:
callback: Function to remove from callbacks
"""
with self._lock:
if callback in self._scan_completion_callbacks:
self._scan_completion_callbacks.remove(callback)
logger.debug(f"Removed scan completion callback: {callback.__name__}")
def _execute_scan(self):
"""Execute the actual media library scan"""
with self._lock:
if self._shutting_down:
logger.debug("Media scan execution skipped during shutdown")
return
if self._scan_in_progress:
logger.warning("Scan already in progress - skipping duplicate execution")
return
self._scan_in_progress = True
self._downloads_during_scan = False
self._timer = None
self._scan_start_time = time.time()
# Get the active media client
media_client, server_type = self._get_active_media_client()
if not media_client:
logger.error("No active media client available for library scan")
self._reset_scan_state()
return
logger.info(f"Starting {server_type.upper()} library scan...")
try:
success = media_client.trigger_library_scan()
if success:
logger.info(f"{server_type.upper()} library scan initiated successfully")
# Start new periodic update system instead of completion detection
self._start_periodic_updates()
else:
logger.error(f"Failed to initiate {server_type.upper()} library scan")
self._reset_scan_state()
except Exception as e:
logger.error(f"Exception during {server_type.upper()} library scan: {e}")
self._reset_scan_state()
def _start_periodic_updates(self):
"""Start periodic database updates while media server is scanning"""
try:
with self._lock:
if self._is_doing_periodic_updates:
logger.debug("Periodic updates already in progress")
return
self._is_doing_periodic_updates = True
logger.info(f"Starting periodic database updates - will check/update every {self._periodic_update_interval//60} minutes")
# Schedule first periodic update after 5 minutes
self._periodic_update_timer = threading.Timer(self._periodic_update_interval, self._do_periodic_update)
self._periodic_update_timer.daemon = True
self._periodic_update_timer.start()
except Exception as e:
logger.error(f"Error starting periodic updates: {e}")
self._reset_scan_state()
def _do_periodic_update(self):
"""Execute periodic database update and check if scanning continues"""
try:
with self._lock:
if not self._scan_in_progress:
logger.debug("Scan no longer in progress, stopping periodic updates")
return
# Check for timeout
if self._scan_start_time and (time.time() - self._scan_start_time) > self._max_scan_time:
logger.warning(f"Media scan timeout reached ({self._max_scan_time}s), stopping periodic updates")
self._stop_periodic_updates()
return
# Get the active media client
media_client, server_type = self._get_active_media_client()
if not media_client:
logger.warning("No active media client available for scan status check")
self._stop_periodic_updates()
return
# Check if media server is still scanning
is_scanning = media_client.is_library_scanning("Music")
elapsed_time = time.time() - self._scan_start_time if self._scan_start_time else 0
logger.info(f"PERIODIC UPDATE: After {elapsed_time//60:.0f} minutes - {server_type.upper()} scanning: {is_scanning}")
if is_scanning:
# Still scanning - trigger database update and continue periodic updates
logger.info(f"{server_type.upper()} still scanning - triggering database update")
if self._shutting_down:
return
self._call_completion_callbacks()
# Schedule next periodic update
if self._shutting_down:
return
logger.info(f"Scheduling next periodic update in {self._periodic_update_interval//60} minutes")
self._periodic_update_timer = threading.Timer(self._periodic_update_interval, self._do_periodic_update)
self._periodic_update_timer.daemon = True
self._periodic_update_timer.start()
else:
# Scanning stopped - final update and cleanup
logger.info(f"{server_type.upper()} scanning completed - doing final database update")
if self._shutting_down:
return
self._call_completion_callbacks()
self._stop_periodic_updates()
except Exception as e:
logger.error(f"Error during periodic update: {e}")
self._stop_periodic_updates()
def _stop_periodic_updates(self):
"""Stop periodic updates and clean up"""
try:
with self._lock:
self._is_doing_periodic_updates = False
if self._periodic_update_timer:
self._periodic_update_timer.cancel()
self._periodic_update_timer = None
logger.info("Stopped periodic database updates")
self._scan_completed()
except Exception as e:
logger.error(f"Error stopping periodic updates: {e}")
def _scan_completed(self):
"""Called when we assume the scan has completed"""
with self._lock:
was_in_progress = self._scan_in_progress
downloads_during_scan = self._downloads_during_scan
# Reset scan state
self._scan_in_progress = False
if not was_in_progress:
logger.debug("Scan completion callback called but scan was not in progress")
return
logger.info("Media library scan completed")
# Call registered completion callbacks
self._call_completion_callbacks()
# Check if we need a follow-up scan
if downloads_during_scan:
logger.info("Downloads occurred during scan - triggering follow-up scan")
self.request_scan("Follow-up scan for downloads during previous scan")
else:
logger.info("No downloads during scan - scan cycle complete")
def _call_completion_callbacks(self):
"""Call all registered scan completion callbacks"""
with self._lock:
callbacks = self._scan_completion_callbacks.copy() # Copy to avoid lock issues
logger.info(f"DEBUG: Calling {len(callbacks)} scan completion callbacks")
for callback in callbacks:
try:
logger.info(f"DEBUG: Executing callback: {callback.__name__}")
callback()
logger.info(f"DEBUG: Callback {callback.__name__} completed successfully")
except Exception as e:
logger.error(f"Error in scan completion callback {callback.__name__}: {e}")
def _reset_scan_state(self):
"""Reset scan state after an error"""
with self._lock:
self._scan_in_progress = False
self._scan_start_time = None
# Cancel periodic updates if running
if self._periodic_update_timer:
self._periodic_update_timer.cancel()
self._periodic_update_timer = None
self._is_doing_periodic_updates = False
def force_scan(self):
"""
Force an immediate scan, bypassing debouncing.
Use sparingly - mainly for manual/administrative triggers.
"""
with self._lock:
if self._timer:
self._timer.cancel()
self._timer = None
if self._scan_in_progress:
logger.warning("Force scan requested but scan already in progress")
return
logger.info("Force scan requested - executing immediately")
self._execute_scan()
def get_status(self) -> dict:
"""Get current status of the scan manager"""
with self._lock:
return {
'scan_in_progress': self._scan_in_progress,
'downloads_during_scan': self._downloads_during_scan,
'timer_active': self._timer is not None,
'delay_seconds': self.delay
}
def shutdown(self):
"""Clean shutdown - cancel any pending timers"""
with self._lock:
self._shutting_down = True
if self._timer:
self._timer.cancel()
self._timer = None
if self._periodic_update_timer:
self._periodic_update_timer.cancel()
self._periodic_update_timer = None
self._is_doing_periodic_updates = False
logger.info("MediaScanManager shutdown - cancelled all pending timers")

342
core/plex_scan_manager.py
View file

@ -1,342 +0,0 @@
#!/usr/bin/env python3
import threading
import time
from utils.logging_config import get_logger
logger = get_logger("plex_scan_manager")
class PlexScanManager:
"""
Smart Plex library scan manager with debouncing and scan-aware follow-up logic.
Features:
- Debounces multiple scan requests to prevent spam
- Tracks downloads that happen during active scans
- Automatically triggers follow-up scans when needed
- Thread-safe operation
"""
def __init__(self, plex_client, delay_seconds: int = 60):
"""
Initialize the scan manager.
Args:
plex_client: PlexClient instance with trigger_library_scan method
delay_seconds: Debounce delay in seconds (default 60s)
"""
self.plex_client = plex_client
self.delay = delay_seconds
self._timer = None
self._scan_in_progress = False
self._downloads_during_scan = False
self._lock = threading.Lock()
self._scan_completion_callbacks = [] # List of callback functions to call when scan completes
self._scan_start_time = None # Track when scan started for timeout
self._max_scan_time = 1800 # Maximum scan time in seconds (30 minutes)
# New periodic update system
self._periodic_update_timer = None # Timer for 5-minute periodic updates
self._periodic_update_interval = 300 # 5 minutes in seconds
self._is_doing_periodic_updates = False # Track if we're in periodic update mode
self._shutting_down = False
logger.info(f"PlexScanManager initialized with {delay_seconds}s debounce delay")
def request_scan(self, reason: str = "Download completed"):
"""
Request a library scan with smart debouncing logic.
Args:
reason: Optional reason for the scan request (for logging)
"""
logger.info(f"DEBUG: Plex scan requested - reason: {reason}")
with self._lock:
if self._shutting_down:
logger.debug("Plex scan request ignored during shutdown")
return
if self._scan_in_progress:
# Plex is currently scanning - mark that we need another scan later
self._downloads_during_scan = True
logger.info(f"Plex scan in progress - queueing follow-up scan ({reason})")
return
# Cancel any existing timer and start a new one
if self._timer:
self._timer.cancel()
logger.debug(f"Resetting scan timer ({reason})")
else:
logger.info(f"Plex scan queued - will execute in {self.delay}s ({reason})")
# Start the debounce timer
self._timer = threading.Timer(self.delay, self._execute_scan)
self._timer.daemon = True
self._timer.start()
def add_scan_completion_callback(self, callback):
"""
Add a callback function to be called when scan completes.
Args:
callback: Function to call when scan completes (no arguments)
"""
with self._lock:
if callback not in self._scan_completion_callbacks:
self._scan_completion_callbacks.append(callback)
logger.info(f"DEBUG: Added scan completion callback: {callback.__name__}")
logger.info(f"DEBUG: Total callbacks registered: {len(self._scan_completion_callbacks)}")
def remove_scan_completion_callback(self, callback):
"""
Remove a previously registered callback.
Args:
callback: Function to remove from callbacks
"""
with self._lock:
if callback in self._scan_completion_callbacks:
self._scan_completion_callbacks.remove(callback)
logger.debug(f"Removed scan completion callback: {callback.__name__}")
def _execute_scan(self):
"""Execute the actual Plex library scan"""
with self._lock:
if self._shutting_down:
logger.debug("Plex scan execution skipped during shutdown")
return
if self._scan_in_progress:
logger.warning("Scan already in progress - skipping duplicate execution")
return
self._scan_in_progress = True
self._downloads_during_scan = False
self._timer = None
self._scan_start_time = time.time()
logger.info("Starting Plex library scan...")
try:
success = self.plex_client.trigger_library_scan()
if success:
logger.info("Plex library scan initiated successfully")
# Start new periodic update system instead of completion detection
self._start_periodic_updates()
else:
logger.error("Failed to initiate Plex library scan")
self._reset_scan_state()
except Exception as e:
logger.error(f"Exception during Plex library scan: {e}")
self._reset_scan_state()
def _start_periodic_updates(self):
"""Start periodic database updates while Plex is scanning"""
try:
with self._lock:
if self._is_doing_periodic_updates:
logger.debug("Periodic updates already in progress")
return
self._is_doing_periodic_updates = True
logger.info(f"Starting periodic database updates - will check/update every {self._periodic_update_interval//60} minutes")
# Schedule first periodic update after 5 minutes
self._periodic_update_timer = threading.Timer(self._periodic_update_interval, self._do_periodic_update)
self._periodic_update_timer.daemon = True
self._periodic_update_timer.start()
except Exception as e:
logger.error(f"Error starting periodic updates: {e}")
self._reset_scan_state()
def _do_periodic_update(self):
"""Execute periodic database update and check if scanning continues"""
try:
with self._lock:
if not self._scan_in_progress:
logger.debug("Scan no longer in progress, stopping periodic updates")
return
# Check for timeout
if self._scan_start_time and (time.time() - self._scan_start_time) > self._max_scan_time:
logger.warning(f"Plex scan timeout reached ({self._max_scan_time}s), stopping periodic updates")
self._stop_periodic_updates()
return
# Check if Plex is still scanning
is_scanning = self.plex_client.is_library_scanning("Music")
elapsed_time = time.time() - self._scan_start_time if self._scan_start_time else 0
logger.info(f"PERIODIC UPDATE: After {elapsed_time//60:.0f} minutes - Plex scanning: {is_scanning}")
if is_scanning:
# Still scanning - trigger database update and continue periodic updates
logger.info("Plex still scanning - triggering database update")
if self._shutting_down:
return
self._call_completion_callbacks()
# Schedule next periodic update
if self._shutting_down:
return
logger.info(f"Scheduling next periodic update in {self._periodic_update_interval//60} minutes")
self._periodic_update_timer = threading.Timer(self._periodic_update_interval, self._do_periodic_update)
self._periodic_update_timer.daemon = True
self._periodic_update_timer.start()
else:
# Scanning stopped - final update and cleanup
logger.info("Plex scanning completed - doing final database update")
if self._shutting_down:
return
self._call_completion_callbacks()
self._stop_periodic_updates()
except Exception as e:
logger.error(f"Error during periodic update: {e}")
self._stop_periodic_updates()
def _stop_periodic_updates(self):
"""Stop periodic updates and clean up"""
try:
with self._lock:
self._is_doing_periodic_updates = False
if self._periodic_update_timer:
self._periodic_update_timer.cancel()
self._periodic_update_timer = None
logger.info("Stopped periodic database updates")
self._scan_completed()
except Exception as e:
logger.error(f"Error stopping periodic updates: {e}")
def _poll_scan_status(self):
"""Poll Plex to check if library scan is still running"""
try:
with self._lock:
if not self._scan_in_progress:
logger.debug("Scan no longer in progress, stopping polling")
return
# Check for timeout
if self._scan_start_time and (time.time() - self._scan_start_time) > self._max_scan_time:
logger.warning(f"Plex scan timeout reached ({self._max_scan_time}s), assuming completion")
self._scan_completed()
return
# Check if Plex is still scanning
is_scanning = self.plex_client.is_library_scanning("Music")
logger.info(f"DEBUG: Plex scan status check - is_scanning: {is_scanning}")
if is_scanning:
# Still scanning, poll again in 30 seconds
logger.info("DEBUG: Plex library still scanning, will check again in 30 seconds")
timer = threading.Timer(30, self._poll_scan_status)
timer.daemon = True
timer.start()
else:
# Scan completed!
elapsed_time = time.time() - self._scan_start_time if self._scan_start_time else 0
logger.info(f"Plex library scan detected as completed (took {elapsed_time:.1f} seconds)")
self._scan_completed()
except Exception as e:
logger.error(f"Error polling scan status: {e}")
# Fallback to assuming completion after error
self._scan_completed()
def _scan_completed(self):
"""Called when we assume the scan has completed"""
with self._lock:
was_in_progress = self._scan_in_progress
downloads_during_scan = self._downloads_during_scan
# Reset scan state
self._scan_in_progress = False
if not was_in_progress:
logger.debug("Scan completion callback called but scan was not in progress")
return
logger.info("Plex library scan completed")
# Call registered completion callbacks
self._call_completion_callbacks()
# Check if we need a follow-up scan
if downloads_during_scan:
logger.info("Downloads occurred during scan - triggering follow-up scan")
self.request_scan("Follow-up scan for downloads during previous scan")
else:
logger.info("No downloads during scan - scan cycle complete")
def _call_completion_callbacks(self):
"""Call all registered scan completion callbacks"""
with self._lock:
callbacks = self._scan_completion_callbacks.copy() # Copy to avoid lock issues
logger.info(f"DEBUG: Calling {len(callbacks)} scan completion callbacks")
for callback in callbacks:
try:
logger.info(f"DEBUG: Executing callback: {callback.__name__}")
callback()
logger.info(f"DEBUG: Callback {callback.__name__} completed successfully")
except Exception as e:
logger.error(f"Error in scan completion callback {callback.__name__}: {e}")
def _reset_scan_state(self):
"""Reset scan state after an error"""
with self._lock:
self._scan_in_progress = False
self._scan_start_time = None
# Cancel periodic updates if running
if self._periodic_update_timer:
self._periodic_update_timer.cancel()
self._periodic_update_timer = None
self._is_doing_periodic_updates = False
def force_scan(self):
"""
Force an immediate scan, bypassing debouncing.
Use sparingly - mainly for manual/administrative triggers.
"""
with self._lock:
if self._timer:
self._timer.cancel()
self._timer = None
if self._scan_in_progress:
logger.warning("Force scan requested but scan already in progress")
return
logger.info("Force scan requested - executing immediately")
self._execute_scan()
def get_status(self) -> dict:
"""Get current status of the scan manager"""
with self._lock:
return {
'scan_in_progress': self._scan_in_progress,
'downloads_during_scan': self._downloads_during_scan,
'timer_active': self._timer is not None,
'delay_seconds': self.delay
}
def shutdown(self):
"""Clean shutdown - cancel any pending timers"""
with self._lock:
self._shutting_down = True
if self._timer:
self._timer.cancel()
self._timer = None
if self._periodic_update_timer:
self._periodic_update_timer.cancel()
self._periodic_update_timer = None
self._is_doing_periodic_updates = False
logger.info("PlexScanManager shutdown - cancelled all pending timers")

461
main.py
View file

@ -1,461 +0,0 @@
#!/usr/bin/env python3
import sys
import asyncio
import time
from pathlib import Path
from PyQt6.QtWidgets import QApplication, QMainWindow, QVBoxLayout, QHBoxLayout, QWidget, QStackedWidget
from PyQt6.QtCore import QThread, pyqtSignal, QTimer, QThreadPool
from PyQt6.QtGui import QFont, QPalette, QColor
from config.settings import config_manager
from utils.logging_config import setup_logging, get_logger
from core.spotify_client import SpotifyClient
from core.plex_client import PlexClient
from core.jellyfin_client import JellyfinClient
from core.navidrome_client import NavidromeClient
from core.soulseek_client import SoulseekClient
from ui.sidebar import ModernSidebar
from ui.pages.dashboard import DashboardPage
from ui.pages.sync import SyncPage
from ui.pages.downloads import DownloadsPage
from ui.pages.artists import ArtistsPage
from ui.pages.settings import SettingsPage
from ui.components.toast_manager import ToastManager
logger = get_logger("main")
class ServiceStatusThread(QThread):
status_updated = pyqtSignal(str, bool)
def __init__(self, spotify_client, plex_client, jellyfin_client, navidrome_client, soulseek_client):
super().__init__()
self.spotify_client = spotify_client
self.plex_client = plex_client
self.jellyfin_client = jellyfin_client
self.navidrome_client = navidrome_client
self.soulseek_client = soulseek_client
self.running = True
# Import here to avoid circular imports
from config.settings import config_manager
self.config_manager = config_manager
def run(self):
while self.running:
try:
# Check Spotify authentication - but don't trigger OAuth
spotify_status = self.spotify_client.sp is not None
self.status_updated.emit("spotify", spotify_status)
# Check active media server connection
active_server = self.config_manager.get_active_media_server()
if active_server == "plex":
server_status = self.plex_client.is_connected()
self.status_updated.emit("plex", server_status)
elif active_server == "jellyfin":
# Use the JellyfinClient for status checking
jellyfin_status = self.jellyfin_client.is_connected()
self.status_updated.emit("jellyfin", jellyfin_status)
elif active_server == "navidrome":
# Use the NavidromeClient for status checking
navidrome_status = self.navidrome_client.is_connected()
self.status_updated.emit("navidrome", navidrome_status)
# Check Soulseek connection (simplified check to avoid event loop issues)
soulseek_status = self.soulseek_client.is_configured()
self.status_updated.emit("soulseek", soulseek_status)
self.msleep(10000) # Check every 10 seconds (less aggressive)
except Exception as e:
logger.error(f"Error checking service status: {e}")
self.msleep(10000)
def stop(self):
self.running = False
self.quit()
self.wait(2000) # Wait max 2 seconds
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
# Track application start time for uptime calculation
self.app_start_time = time.time()
self.spotify_client = SpotifyClient()
self.plex_client = PlexClient()
self.jellyfin_client = JellyfinClient()
self.navidrome_client = NavidromeClient()
self.soulseek_client = SoulseekClient()
self.status_thread = None
self.init_ui()
self.setup_status_monitoring()
# Setup periodic search maintenance (rolling 50-search window)
self.setup_search_maintenance()
def setup_search_maintenance(self):
"""Setup periodic search history maintenance to keep only the 50 most recent searches"""
try:
# Create timer for periodic search maintenance
self.search_maintenance_timer = QTimer()
self.search_maintenance_timer.timeout.connect(self._run_search_maintenance)
# Run maintenance every 2 minutes (120 seconds)
# This keeps search history clean without being too frequent
self.search_maintenance_timer.start(120000)
logger.info("Search maintenance timer started (every 2 minutes, keeps 200 most recent searches)")
except Exception as e:
logger.error(f"Error setting up search maintenance: {e}")
def _run_search_maintenance(self):
"""Run search maintenance in background thread to avoid blocking UI"""
try:
# Only run if Soulseek client seems to be available
if hasattr(self.soulseek_client, 'base_url') and self.soulseek_client.base_url:
# Run maintenance in background thread
import threading
def maintenance_thread():
try:
import asyncio
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# Run the maintenance (keep 200 most recent searches)
success = loop.run_until_complete(self.soulseek_client.maintain_search_history(200))
if not success:
logger.warning("Search maintenance completed with some failures")
except Exception as e:
logger.error(f"Error in search maintenance thread: {e}")
finally:
loop.close()
thread = threading.Thread(target=maintenance_thread, daemon=True)
thread.start()
else:
logger.debug("Soulseek client not configured, skipping search maintenance")
except Exception as e:
logger.error(f"Error running search maintenance: {e}")
def init_ui(self):
self.setWindowTitle("SoulSync - Music Sync & Manager")
self.setGeometry(100, 100, 1400, 900)
# Set dark theme palette
self.setStyleSheet("""
QMainWindow {
background: #121212;
}
""")
# Create central widget
central_widget = QWidget()
self.setCentralWidget(central_widget)
# Main layout
main_layout = QHBoxLayout(central_widget)
main_layout.setContentsMargins(0, 0, 0, 0)
main_layout.setSpacing(0)
# Create sidebar
self.sidebar = ModernSidebar()
self.sidebar.page_changed.connect(self.change_page)
main_layout.addWidget(self.sidebar)
# Create stacked widget for pages
self.stacked_widget = QStackedWidget()
# Create toast manager
self.toast_manager = ToastManager(self)
# Create and add pages
self.dashboard_page = DashboardPage()
self.downloads_page = DownloadsPage(self.soulseek_client)
self.sync_page = SyncPage(
spotify_client=self.spotify_client,
plex_client=self.plex_client,
soulseek_client=self.soulseek_client,
downloads_page=self.downloads_page,
jellyfin_client=self.jellyfin_client,
navidrome_client=self.navidrome_client
)
self.artists_page = ArtistsPage(downloads_page=self.downloads_page)
self.settings_page = SettingsPage()
# Set toast manager for pages that need direct access
self.downloads_page.set_toast_manager(self.toast_manager)
self.sync_page.set_toast_manager(self.toast_manager)
self.artists_page.set_toast_manager(self.toast_manager)
self.settings_page.set_toast_manager(self.toast_manager)
# Configure dashboard with service clients and page references
self.dashboard_page.set_service_clients(self.spotify_client, self.plex_client, self.jellyfin_client, self.navidrome_client, self.soulseek_client)
self.dashboard_page.set_page_references(self.downloads_page, self.sync_page)
self.dashboard_page.set_app_start_time(self.app_start_time)
self.dashboard_page.set_toast_manager(self.toast_manager)
# Connect download completion signal for session tracking
self.downloads_page.download_session_completed.connect(
self.dashboard_page.data_provider.increment_completed_downloads
)
# Connect sync activities to dashboard
self.sync_page.sync_activity.connect(
self.dashboard_page.add_activity_item
)
# Connect download activities to dashboard
self.downloads_page.download_activity.connect(
self.dashboard_page.add_activity_item
)
# --- ADD THESE TWO LINES TO FIX THE UI UPDATE ---
self.sync_page.database_updated_externally.connect(self.dashboard_page.database_updated_externally)
self.artists_page.database_updated_externally.connect(self.dashboard_page.database_updated_externally)
# ------------------------------------------------
self.stacked_widget.addWidget(self.dashboard_page)
self.stacked_widget.addWidget(self.sync_page)
self.stacked_widget.addWidget(self.downloads_page)
self.stacked_widget.addWidget(self.artists_page)
self.stacked_widget.addWidget(self.settings_page)
main_layout.addWidget(self.stacked_widget)
# Set dashboard as default page
self.change_page("dashboard")
# Connect media player signals between sidebar and downloads page
self.setup_media_player_connections()
# Connect settings change signals for live updates
self.setup_settings_connections()
def setup_status_monitoring(self):
# Start status monitoring thread
self.status_thread = ServiceStatusThread(
self.spotify_client,
self.plex_client,
self.jellyfin_client,
self.navidrome_client,
self.soulseek_client
)
self.status_thread.status_updated.connect(self.update_service_status)
self.status_thread.start()
def setup_media_player_connections(self):
"""Connect signals between downloads page and sidebar media player"""
# Connect downloads page signals to sidebar media player
self.downloads_page.track_started.connect(self.sidebar.media_player.set_track_info)
self.downloads_page.track_paused.connect(lambda: self.sidebar.media_player.set_playing_state(False))
self.downloads_page.track_resumed.connect(lambda: self.sidebar.media_player.set_playing_state(True))
self.downloads_page.track_stopped.connect(self.sidebar.media_player.clear_track)
self.downloads_page.track_finished.connect(self.sidebar.media_player.clear_track)
# Connect loading animation signals
self.downloads_page.track_loading_started.connect(lambda result: self.sidebar.media_player.show_loading())
self.downloads_page.track_loading_finished.connect(lambda result: self.sidebar.media_player.hide_loading())
self.downloads_page.track_loading_progress.connect(lambda progress, result: self.sidebar.media_player.set_loading_progress(progress))
# Connect sidebar media player signals to downloads page
self.sidebar.media_player.play_pause_requested.connect(self.downloads_page.handle_sidebar_play_pause)
self.sidebar.media_player.stop_requested.connect(self.downloads_page.handle_sidebar_stop)
self.sidebar.media_player.volume_changed.connect(self.downloads_page.handle_sidebar_volume)
logger.info("Media player connections established between sidebar and downloads page")
def setup_settings_connections(self):
"""Connect settings change signals for live updates across pages"""
self.settings_page.settings_changed.connect(self.on_settings_changed)
logger.info("Settings change connections established")
def on_settings_changed(self, key: str, value: str):
"""Handle settings changes and broadcast to relevant pages"""
# Reinitialize service clients when their settings change
if key.startswith('spotify.'):
try:
self.spotify_client._setup_client()
except Exception as e:
logger.error("Failed to reinitialize Spotify client")
elif key.startswith('plex.'):
try:
# Reset Plex connection to force reconnection with new settings
self.plex_client.server = None
self.plex_client.music_library = None
self.plex_client._connection_attempted = False
except Exception as e:
logger.error("Failed to reset Plex client")
elif key.startswith('soulseek.'):
try:
self.soulseek_client._setup_client()
except Exception as e:
logger.error("Failed to reinitialize Soulseek client")
# Broadcast to all pages that need to know about path changes
if hasattr(self.downloads_page, 'on_paths_updated'):
self.downloads_page.on_paths_updated(key, value)
if hasattr(self.artists_page, 'on_paths_updated'):
self.artists_page.on_paths_updated(key, value)
def change_page(self, page_id: str):
page_map = {
"dashboard": 0,
"sync": 1,
"downloads": 2,
"artists": 3,
"settings": 4
}
if page_id in page_map:
self.stacked_widget.setCurrentIndex(page_map[page_id])
logger.info(f"Changed to page: {page_id}")
def update_service_status(self, service: str, connected: bool):
self.sidebar.update_service_status(service, connected)
# Update dashboard with service status
if hasattr(self.dashboard_page, 'data_provider'):
self.dashboard_page.data_provider.update_service_status(service, connected)
# Force a refresh of the Spotify client if needed
if service == "spotify" and not connected:
try:
self.spotify_client._setup_client()
except Exception as e:
logger.error(f"Error refreshing Spotify client: {e}")
def closeEvent(self, event):
logger.info("Closing application...")
try:
# Stop all page threads first
if hasattr(self, 'downloads_page') and self.downloads_page:
logger.info("Cleaning up Downloads page threads...")
self.downloads_page.cleanup_all_threads()
# Stop dashboard threads
if hasattr(self, 'dashboard_page') and self.dashboard_page:
logger.info("Cleaning up Dashboard page threads...")
self.dashboard_page.cleanup_threads()
# Stop other page threads and background tasks
if hasattr(self, 'artists_page') and self.artists_page:
logger.info("Cleaning up Artists page threads...")
if hasattr(self.artists_page, 'cleanup_threads'):
self.artists_page.cleanup_threads()
if hasattr(self, 'sync_page') and self.sync_page:
logger.info("Cleaning up Sync page threads...")
if hasattr(self.sync_page, 'cleanup_threads'):
self.sync_page.cleanup_threads()
if hasattr(self, 'downloads_page') and self.downloads_page:
logger.info("Cleaning up Downloads page threads...")
if hasattr(self.downloads_page, 'cleanup_threads'):
self.downloads_page.cleanup_threads()
# Stop all QThreadPool tasks
logger.info("Stopping global thread pool...")
QThreadPool.globalInstance().clear()
QThreadPool.globalInstance().waitForDone(1000) # Wait max 1 second
# Stop status monitoring thread
if self.status_thread:
logger.info("Stopping status monitoring thread...")
self.status_thread.stop()
# Stop search maintenance timer
if hasattr(self, 'search_maintenance_timer') and self.search_maintenance_timer:
logger.info("Stopping search maintenance timer...")
self.search_maintenance_timer.stop()
# Close Soulseek client
try:
logger.info("Closing Soulseek client...")
# Use modern asyncio approach instead of deprecated get_event_loop
try:
loop = asyncio.get_running_loop()
# Create a new task to close the client
task = asyncio.create_task(self.soulseek_client.close())
# Wait for it to complete
asyncio.run_coroutine_threadsafe(self.soulseek_client.close(), loop).result(timeout=3.0)
except RuntimeError:
# No running loop, create new one
asyncio.run(self.soulseek_client.close())
except Exception as e:
logger.error(f"Error closing Soulseek client: {e}")
# Close database connection
try:
logger.info("Closing database connection...")
from database import close_database
close_database()
except Exception as e:
logger.error(f"Error closing database: {e}")
logger.info("Application closed successfully")
event.accept()
except Exception as e:
logger.error(f"Error during application shutdown: {e}")
# Force accept the event to prevent hanging
event.accept()
def main():
# Check for saved log level preference in database
try:
from database.music_database import MusicDatabase
db = MusicDatabase()
saved_log_level = db.get_preference('log_level')
if saved_log_level:
log_level = saved_log_level
else:
# Fall back to config file
logging_config = config_manager.get_logging_config()
log_level = logging_config.get('level', 'INFO')
except:
# If database isn't available yet, use config file
logging_config = config_manager.get_logging_config()
log_level = logging_config.get('level', 'INFO')
logging_config = config_manager.get_logging_config()
log_file = logging_config.get('path', 'logs/newmusic.log')
setup_logging(level=log_level, log_file=log_file)
logger.info("Starting Soulsync application")
if not config_manager.config_path.exists():
logger.error("Configuration file not found. Please check config/config.json")
sys.exit(1)
app = QApplication(sys.argv)
app.setApplicationName("SoulSync")
app.setApplicationVersion("0.6")
main_window = MainWindow()
main_window.show()
try:
sys.exit(app.exec())
except KeyboardInterrupt:
logger.info("Application interrupted by user")
sys.exit(0)
except Exception as e:
logger.error(f"Unexpected error: {e}")
sys.exit(1)
if __name__ == "__main__":
main()

2
requirements-dev.txt
View file

@ -1,7 +1,7 @@
# SoulSync development requirements
# Runtime web dependencies + test runner
-r requirements-webui.txt
-r requirements.txt
# Test runner
pytest>=9.0.0

52
requirements-webui.txt
View file

@ -1,52 +0,0 @@
# SoulSync WebUI Requirements
# Docker-compatible requirements without PyQt6 dependencies
# Core web framework
Flask>=3.0.0
Flask-Limiter>=3.5.0
# Music service APIs
spotipy>=2.23.0
PlexAPI>=4.17.0
# HTTP and async support
requests>=2.31.0
aiohttp>=3.9.0
# Configuration management
python-dotenv>=1.0.0
# Security and encryption
cryptography>=41.0.0
# Media metadata handling
mutagen>=1.47.0
Pillow>=10.0.0
# Text processing
unidecode>=1.3.8
beautifulsoup4>=4.12.0
# System monitoring
psutil>=6.0.0
# YouTube support
yt-dlp>=2024.12.13
# Lyrics support
lrclibapi>=0.3.1
# Optional: MQTT support (for future features)
asyncio-mqtt>=0.16.0
# Audio fingerprinting for download verification
pyacoustid>=1.3.0
# WebSocket client for Hydrabase connection
websocket-client>=1.7.0
# Tidal download support
tidalapi>=0.7.6
# WebSocket server for real-time UI updates
flask-socketio>=5.3.0

56
requirements.txt
View file

@ -1,16 +1,46 @@
PyQt6[multimedia]>=6.6.0
spotipy>=2.23.0
PlexAPI>=4.17.0
requests>=2.31.0
asyncio-mqtt>=0.16.0
python-dotenv>=1.0.0
cryptography>=41.0.0
mutagen>=1.47.0
Pillow>=10.0.0
aiohttp>=3.9.0
unidecode>=1.3.8
yt-dlp>=2024.12.13
# SoulSync requirements
# Web application dependencies only
# Core web framework
Flask>=3.0.0
Flask-Limiter>=3.5.0
# Music service APIs
spotipy>=2.23.0
PlexAPI>=4.17.0
# HTTP and async support
requests>=2.31.0
aiohttp>=3.9.0
# Security and encryption
cryptography>=41.0.0
# Media metadata handling
mutagen>=1.47.0
Pillow>=10.0.0
# Text processing
unidecode>=1.3.8
beautifulsoup4>=4.12.0
# System monitoring
psutil>=6.0.0
# YouTube support
yt-dlp>=2024.12.13
# Lyrics support
lrclibapi>=0.3.1
pyacoustid>=1.3.0
# Audio fingerprinting for download verification
pyacoustid>=1.3.0
# WebSocket client for Hydrabase connection
websocket-client>=1.7.0
# Tidal download support
tidalapi>=0.7.6
# WebSocket server for real-time UI updates
flask-socketio>=5.3.0

295
test_acoustid.py
View file

@ -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()

1271
test_hifi_client.py
View file

File diff suppressed because it is too large Load diff

128
tests/test_phase2_dashboard.py → tests/test_dashboard_events.py
View file

@ -21,16 +21,6 @@ import time
class TestSystemStats:
"""dashboard:stats socket events match GET /api/system/stats."""
def test_stats_event_received(self, test_app, shared_state):
"""Client receives a dashboard:stats event."""
app, socketio = test_app
client = socketio.test_client(app)
build = shared_state['build_system_stats']
socketio.emit('dashboard:stats', build())
received = client.get_received()
stats_events = [e for e in received if e['name'] == 'dashboard:stats']
assert len(stats_events) >= 1
def test_stats_shape(self, test_app, shared_state):
"""dashboard:stats event data has expected keys."""
app, socketio = test_app
@ -70,16 +60,6 @@ class TestSystemStats:
assert ws_data['uptime'] == http_data['uptime']
assert ws_data['memory_usage'] == http_data['memory_usage']
def test_http_stats_still_works(self, flask_client):
"""GET /api/system/stats returns 200 with expected keys."""
resp = flask_client.get('/api/system/stats')
assert resp.status_code == 200
data = resp.get_json()
assert 'active_downloads' in data
assert 'finished_downloads' in data
assert 'download_speed' in data
# =========================================================================
# Group B — Activity Feed
# =========================================================================
@ -87,16 +67,6 @@ class TestSystemStats:
class TestActivityFeed:
"""dashboard:activity socket events match GET /api/activity/feed."""
def test_activity_event_received(self, test_app, shared_state):
"""Client receives a dashboard:activity event."""
app, socketio = test_app
client = socketio.test_client(app)
build = shared_state['build_activity_feed_payload']
socketio.emit('dashboard:activity', build())
received = client.get_received()
events = [e for e in received if e['name'] == 'dashboard:activity']
assert len(events) >= 1
def test_activity_shape(self, test_app, shared_state):
"""dashboard:activity event data has activities array."""
app, socketio = test_app
@ -136,14 +106,6 @@ class TestActivityFeed:
if ws_data['activities']:
assert ws_data['activities'][0]['title'] == http_data['activities'][0]['title']
def test_http_activity_still_works(self, flask_client):
"""GET /api/activity/feed returns 200 with expected structure."""
resp = flask_client.get('/api/activity/feed')
assert resp.status_code == 200
data = resp.get_json()
assert 'activities' in data
# =========================================================================
# Group C — Toasts (instant push)
# =========================================================================
@ -201,18 +163,6 @@ class TestToasts:
assert 'show_toast' in data
assert data['show_toast'] is True
def test_http_toasts_still_works(self, flask_client, shared_state):
"""GET /api/activity/toasts returns 200 with expected structure."""
# Add a toast-worthy activity first
add_item = shared_state['add_activity_item']
add_item('', 'Test', 'Sub', show_toast=True)
resp = flask_client.get('/api/activity/toasts')
assert resp.status_code == 200
data = resp.get_json()
assert 'toasts' in data
# =========================================================================
# Group D — DB Stats
# =========================================================================
@ -220,16 +170,6 @@ class TestToasts:
class TestDbStats:
"""dashboard:db_stats socket events match GET /api/database/stats."""
def test_db_stats_event_received(self, test_app, shared_state):
"""Client receives a dashboard:db_stats event."""
app, socketio = test_app
client = socketio.test_client(app)
build = shared_state['build_db_stats']
socketio.emit('dashboard:db_stats', build())
received = client.get_received()
events = [e for e in received if e['name'] == 'dashboard:db_stats']
assert len(events) >= 1
def test_db_stats_shape(self, test_app, shared_state):
"""dashboard:db_stats event data has expected keys."""
app, socketio = test_app
@ -266,16 +206,6 @@ class TestDbStats:
assert ws_data['database_size_mb'] == http_data['database_size_mb']
assert ws_data['server_source'] == http_data['server_source']
def test_http_db_stats_still_works(self, flask_client):
"""GET /api/database/stats returns 200 with expected keys."""
resp = flask_client.get('/api/database/stats')
assert resp.status_code == 200
data = resp.get_json()
assert 'artists' in data
assert 'albums' in data
assert 'tracks' in data
# =========================================================================
# Group E — Wishlist Count
# =========================================================================
@ -283,16 +213,6 @@ class TestDbStats:
class TestWishlistCount:
"""dashboard:wishlist_count socket events match GET /api/wishlist/count."""
def test_wishlist_count_received(self, test_app, shared_state):
"""Client receives a dashboard:wishlist_count event."""
app, socketio = test_app
client = socketio.test_client(app)
build = shared_state['build_wishlist_count_payload_ws']
socketio.emit('dashboard:wishlist_count', build())
received = client.get_received()
events = [e for e in received if e['name'] == 'dashboard:wishlist_count']
assert len(events) >= 1
def test_wishlist_count_shape(self, test_app, shared_state):
"""dashboard:wishlist_count event data has count key."""
app, socketio = test_app
@ -322,54 +242,6 @@ class TestWishlistCount:
assert ws_data['count'] == http_data['count']
def test_http_wishlist_count_still_works(self, flask_client):
"""GET /api/wishlist/count returns 200 with count."""
resp = flask_client.get('/api/wishlist/count')
assert resp.status_code == 200
data = resp.get_json()
assert 'count' in data
assert isinstance(data['count'], int)
# =========================================================================
# Group F — Backward Compatibility
# =========================================================================
class TestBackwardCompat:
"""HTTP endpoints work when no WebSocket is connected."""
def test_all_http_endpoints_work_without_socket(self, flask_client):
"""All 5 Phase 2 HTTP endpoints work without any WebSocket connection."""
# System stats
resp = flask_client.get('/api/system/stats')
assert resp.status_code == 200
data = resp.get_json()
assert data['active_downloads'] == 2
# Activity feed
resp = flask_client.get('/api/activity/feed')
assert resp.status_code == 200
data = resp.get_json()
assert 'activities' in data
# Toasts
resp = flask_client.get('/api/activity/toasts')
assert resp.status_code == 200
data = resp.get_json()
assert 'toasts' in data
# DB stats
resp = flask_client.get('/api/database/stats')
assert resp.status_code == 200
data = resp.get_json()
assert data['artists'] == 350
# Wishlist count
resp = flask_client.get('/api/wishlist/count')
assert resp.status_code == 200
data = resp.get_json()
assert data['count'] == 5
def test_multiple_clients_get_dashboard_updates(self, test_app, shared_state):
"""Multiple WebSocket clients each receive dashboard broadcast events."""
app, socketio = test_app

44
tests/test_phase3_enrichment.py → tests/test_enrichment_events.py
View file

@ -39,19 +39,6 @@ ENDPOINTS = {
class TestEnrichmentEventDelivery:
"""enrichment:<worker> socket events are received by the client."""
@pytest.mark.parametrize('worker', WORKERS)
def test_enrichment_event_received(self, test_app, shared_state, worker):
"""Client receives an enrichment:<worker> event."""
app, socketio = test_app
client = socketio.test_client(app)
build = shared_state['build_enrichment_status']
socketio.emit(f'enrichment:{worker}', build(worker))
received = client.get_received()
events = [e for e in received if e['name'] == f'enrichment:{worker}']
assert len(events) >= 1
# =========================================================================
# Group B — Data Shape (parameterized)
# =========================================================================
@ -160,38 +147,11 @@ class TestEnrichmentHttpParity:
# =========================================================================
# Group D — HTTP Still Works (parameterized)
# =========================================================================
class TestEnrichmentHttpStillWorks:
"""HTTP endpoints return 200 with expected structure."""
@pytest.mark.parametrize('worker', WORKERS)
def test_http_enrichment_still_works(self, flask_client, worker):
"""GET /api/<worker>/status returns 200."""
endpoint = ENDPOINTS[worker]
resp = flask_client.get(endpoint)
assert resp.status_code == 200
data = resp.get_json()
assert 'running' in data
assert 'paused' in data
# =========================================================================
# Group E — Backward Compatibility
# Group D — Backward Compatibility
# =========================================================================
class TestEnrichmentBackwardCompat:
"""HTTP endpoints work when no WebSocket is connected."""
def test_all_http_endpoints_work_without_socket(self, flask_client):
"""All 7 enrichment HTTP endpoints work without any WebSocket connection."""
for worker in WORKERS:
endpoint = ENDPOINTS[worker]
resp = flask_client.get(endpoint)
assert resp.status_code == 200
data = resp.get_json()
assert data['running'] is True
"""WebSocket clients still receive broadcast enrichment updates."""
def test_multiple_clients_get_enrichment_updates(self, test_app, shared_state):
"""Multiple WebSocket clients each receive enrichment events."""

191
tests/test_phase5_sync.py → tests/test_sync_discovery_scan_events.py
View file

@ -29,28 +29,6 @@ SYNC_PLAYLIST_IDS = ['test-playlist-1']
class TestSyncEventDelivery:
"""sync:progress socket events are received by subscribed clients."""
def test_sync_event_received(self, test_app, shared_state):
"""Client subscribes and receives sync:progress event."""
app, socketio = test_app
client = socketio.test_client(app)
build = shared_state['build_sync_status']
# Subscribe to sync room
client.emit('sync:subscribe', {'playlist_ids': ['test-playlist-1']})
# Server emits to the room
socketio.emit('sync:progress', {
'playlist_id': 'test-playlist-1',
**build('test-playlist-1')
}, room='sync:test-playlist-1')
received = client.get_received()
events = [e for e in received if e['name'] == 'sync:progress']
assert len(events) >= 1
client.disconnect()
def test_sync_only_subscribed(self, test_app, shared_state):
"""Unsubscribed client does NOT receive sync:progress events."""
app, socketio = test_app
@ -155,30 +133,6 @@ class TestSyncDataShape:
class TestDiscoveryEventDelivery:
"""discovery:progress socket events are received by subscribed clients."""
@pytest.mark.parametrize('platform,pid', [
('tidal', 'test-tidal-1'),
('youtube', 'test-yt-hash'),
])
def test_discovery_event_received(self, test_app, shared_state, platform, pid):
"""Client subscribes and receives discovery:progress event."""
app, socketio = test_app
client = socketio.test_client(app)
build = shared_state['build_discovery_status']
client.emit('discovery:subscribe', {'ids': [pid]})
payload = build(platform, pid)
payload['platform'] = platform
payload['id'] = pid
socketio.emit('discovery:progress', payload, room=f'discovery:{pid}')
received = client.get_received()
events = [e for e in received if e['name'] == 'discovery:progress']
assert len(events) >= 1
client.disconnect()
def test_discovery_only_subscribed(self, test_app, shared_state):
"""Unsubscribed client does NOT receive discovery:progress events."""
app, socketio = test_app
@ -281,24 +235,6 @@ class TestDiscoveryDataShape:
class TestScanEventDelivery:
"""Broadcast scan events are received by all connected clients."""
def test_watchlist_scan_received(self, test_app, shared_state):
"""All clients receive scan:watchlist event."""
app, socketio = test_app
client1 = socketio.test_client(app)
client2 = socketio.test_client(app)
build = shared_state['build_watchlist_scan_status']
socketio.emit('scan:watchlist', build())
for client in [client1, client2]:
received = client.get_received()
events = [e for e in received if e['name'] == 'scan:watchlist']
assert len(events) >= 1
client1.disconnect()
client2.disconnect()
def test_watchlist_scan_shape(self, test_app, shared_state):
"""scan:watchlist data has success, status, current_artist_name."""
app, socketio = test_app
@ -321,23 +257,6 @@ class TestScanEventDelivery:
client.disconnect()
def test_media_scan_received(self, test_app, shared_state):
"""All clients receive scan:media event."""
app, socketio = test_app
client1 = socketio.test_client(app)
client2 = socketio.test_client(app)
build = shared_state['build_media_scan_status']
socketio.emit('scan:media', build())
for client in [client1, client2]:
received = client.get_received()
events = [e for e in received if e['name'] == 'scan:media']
assert len(events) >= 1
client1.disconnect()
client2.disconnect()
def test_media_scan_shape(self, test_app, shared_state):
"""scan:media data has success and status with is_scanning."""
app, socketio = test_app
@ -491,33 +410,11 @@ class TestScanHttpParity:
class TestHttpStillWorks:
"""HTTP endpoints return 200 with expected structure."""
def test_sync_http_works(self, flask_client):
"""GET /api/sync/status/:id returns 200."""
resp = flask_client.get('/api/sync/status/test-playlist-1')
assert resp.status_code == 200
data = resp.get_json()
assert 'status' in data
assert 'progress' in data
def test_sync_http_404_unknown(self, flask_client):
"""GET /api/sync/status/:id returns 404 for unknown playlist."""
resp = flask_client.get('/api/sync/status/nonexistent')
assert resp.status_code == 404
@pytest.mark.parametrize('platform,endpoint', [
('tidal', '/api/tidal/discovery/status/test-tidal-1'),
('youtube', '/api/youtube/discovery/status/test-yt-hash'),
])
def test_discovery_http_works(self, flask_client, platform, endpoint):
"""GET /api/<platform>/discovery/status/:id returns 200."""
resp = flask_client.get(endpoint)
assert resp.status_code == 200
data = resp.get_json()
assert 'phase' in data
assert 'status' in data
assert 'results' in data
def test_discovery_http_not_found(self, flask_client):
"""GET /api/tidal/discovery/status/:id returns error for unknown ID."""
resp = flask_client.get('/api/tidal/discovery/status/nonexistent')
@ -525,42 +422,12 @@ class TestHttpStillWorks:
data = resp.get_json()
assert 'error' in data
def test_watchlist_scan_http_works(self, flask_client):
"""GET /api/watchlist/scan/status returns 200."""
resp = flask_client.get('/api/watchlist/scan/status')
assert resp.status_code == 200
data = resp.get_json()
assert data['success'] is True
assert 'status' in data
def test_media_scan_http_works(self, flask_client):
"""GET /api/scan/status returns 200."""
resp = flask_client.get('/api/scan/status')
assert resp.status_code == 200
data = resp.get_json()
assert data['success'] is True
assert 'status' in data
# =========================================================================
# Group H — Backward Compatibility
# Group H — Broadcast Updates
# =========================================================================
class TestBackwardCompat:
"""HTTP endpoints work when no WebSocket is connected."""
def test_all_http_endpoints_work_without_socket(self, flask_client):
"""All Phase 5 HTTP endpoints work without any WebSocket connection."""
endpoints = [
'/api/sync/status/test-playlist-1',
'/api/tidal/discovery/status/test-tidal-1',
'/api/youtube/discovery/status/test-yt-hash',
'/api/watchlist/scan/status',
'/api/scan/status',
]
for endpoint in endpoints:
resp = flask_client.get(endpoint)
assert resp.status_code == 200
"""Broadcast scan events still reach multiple clients."""
def test_multiple_clients_get_scan_updates(self, test_app, shared_state):
"""Multiple WebSocket clients each receive scan events."""
@ -668,29 +535,6 @@ PLATFORM_SYNC_IDS = [
class TestPlatformSyncEventDelivery:
"""Platform sync pollers receive sync:progress via WS rooms."""
@pytest.mark.parametrize('platform,sync_id', PLATFORM_SYNC_IDS)
def test_platform_sync_event_received(self, test_app, shared_state, platform, sync_id):
"""Client subscribes to platform sync_playlist_id and receives sync:progress."""
app, socketio = test_app
client = socketio.test_client(app)
build = shared_state['build_sync_status']
client.emit('sync:subscribe', {'playlist_ids': [sync_id]})
socketio.emit('sync:progress', {
'playlist_id': sync_id, **build(sync_id)
}, room=f'sync:{sync_id}')
received = client.get_received()
events = [e for e in received if e['name'] == 'sync:progress']
assert len(events) >= 1
data = events[0]['args'][0]
assert data['playlist_id'] == sync_id
assert data['status'] == 'syncing'
client.disconnect()
@pytest.mark.parametrize('platform,sync_id', PLATFORM_SYNC_IDS)
def test_platform_sync_not_received_when_unsubscribed(
self, test_app, shared_state, platform, sync_id):
@ -796,35 +640,12 @@ class TestPlatformSyncEventDelivery:
client.disconnect()
@pytest.mark.parametrize('platform,sync_id', PLATFORM_SYNC_IDS)
def test_platform_sync_http_still_works(self, flask_client, platform, sync_id):
"""GET /api/sync/status/<platform_sync_id> returns 200."""
resp = flask_client.get(f'/api/sync/status/{sync_id}')
assert resp.status_code == 200
data = resp.get_json()
assert 'status' in data
assert 'progress' in data
# =========================================================================
# Group H — Wishlist Stats (broadcast)
# =========================================================================
class TestWishlistStatsEventDelivery:
"""wishlist:stats broadcast events are received by the client."""
def test_wishlist_stats_event_received(self, test_app, shared_state):
"""Client receives a wishlist:stats event."""
app, socketio = test_app
client = socketio.test_client(app)
build = shared_state['build_wishlist_stats']
socketio.emit('wishlist:stats', build())
received = client.get_received()
events = [e for e in received if e['name'] == 'wishlist:stats']
assert len(events) >= 1
client.disconnect()
def test_wishlist_stats_data_shape(self, test_app, shared_state):
"""wishlist:stats has is_auto_processing and next_run_in_seconds."""
app, socketio = test_app
@ -843,14 +664,6 @@ class TestWishlistStatsEventDelivery:
assert isinstance(data['next_run_in_seconds'], (int, float))
client.disconnect()
def test_wishlist_stats_http_still_works(self, flask_client):
"""GET /api/wishlist/stats returns 200 with expected keys."""
resp = flask_client.get('/api/wishlist/stats')
assert resp.status_code == 200
data = resp.get_json()
assert 'is_auto_processing' in data
assert 'next_run_in_seconds' in data
def test_wishlist_stats_auto_processing_detection(self, test_app, shared_state):
"""When is_auto_processing is True, client detects it."""
app, socketio = test_app

47
tests/test_phase4_tools.py → tests/test_tool_progress_events.py
View file

@ -38,19 +38,6 @@ ENDPOINTS = {
class TestToolEventDelivery:
"""tool:<name> socket events are received by the client."""
@pytest.mark.parametrize('tool', TOOLS)
def test_tool_event_received(self, test_app, shared_state, tool):
"""Client receives a tool:<name> event."""
app, socketio = test_app
client = socketio.test_client(app)
build = shared_state['build_tool_status']
socketio.emit(f'tool:{tool}', build(tool))
received = client.get_received()
events = [e for e in received if e['name'] == f'tool:{tool}']
assert len(events) >= 1
# =========================================================================
# Group B — Data Shape (individual per tool)
# =========================================================================
@ -255,41 +242,11 @@ class TestToolHttpParity:
# =========================================================================
# Group D — HTTP Still Works (parameterized)
# =========================================================================
class TestToolHttpStillWorks:
"""HTTP endpoints return 200 with expected structure."""
@pytest.mark.parametrize('tool', TOOLS)
def test_http_tool_still_works(self, flask_client, tool):
"""GET /api/<tool>/status returns 200."""
endpoint = ENDPOINTS[tool]
resp = flask_client.get(endpoint)
assert resp.status_code == 200
data = resp.get_json()
if tool == 'logs':
assert 'logs' in data
elif tool == 'metadata':
assert 'success' in data
assert 'status' in data
else:
assert 'status' in data
# =========================================================================
# Group E — Backward Compatibility
# Group D — Backward Compatibility
# =========================================================================
class TestToolBackwardCompat:
"""HTTP endpoints work when no WebSocket is connected."""
def test_all_http_endpoints_work_without_socket(self, flask_client):
"""All 7 tool HTTP endpoints work without any WebSocket connection."""
for tool in TOOLS:
endpoint = ENDPOINTS[tool]
resp = flask_client.get(endpoint)
assert resp.status_code == 200
"""WebSocket clients still receive broadcast tool updates."""
def test_multiple_clients_get_tool_updates(self, test_app, shared_state):
"""Multiple WebSocket clients each receive tool events."""

95
tests/test_phase1_websocket.py → tests/test_websocket_infrastructure.py
View file

@ -35,34 +35,6 @@ class TestInfrastructure:
assert client.is_connected()
client.disconnect()
def test_http_status_still_works(self, flask_client):
"""GET /status returns 200 with expected keys."""
resp = flask_client.get('/status')
assert resp.status_code == 200
data = resp.get_json()
assert 'spotify' in data
assert 'media_server' in data
assert 'soulseek' in data
assert 'active_media_server' in data
def test_http_watchlist_count_still_works(self, flask_client):
"""GET /api/watchlist/count returns 200 with expected keys."""
resp = flask_client.get('/api/watchlist/count')
assert resp.status_code == 200
data = resp.get_json()
assert data['success'] is True
assert 'count' in data
assert 'next_run_in_seconds' in data
def test_http_download_batch_still_works(self, flask_client):
"""GET /api/download_status/batch returns 200 with expected structure."""
resp = flask_client.get('/api/download_status/batch')
assert resp.status_code == 200
data = resp.get_json()
assert 'batches' in data
assert 'metadata' in data
# =========================================================================
# Group B — Service Status Parity
# =========================================================================
@ -70,16 +42,6 @@ class TestInfrastructure:
class TestServiceStatus:
"""status:update socket events match GET /status HTTP responses."""
def test_status_update_received(self, test_app, shared_state):
"""Client receives a status:update event."""
app, socketio = test_app
client = socketio.test_client(app)
build = shared_state['build_status_payload']
socketio.emit('status:update', build())
received = client.get_received()
status_events = [e for e in received if e['name'] == 'status:update']
assert len(status_events) >= 1
def test_status_update_shape(self, test_app, shared_state):
"""status:update event data has the expected keys."""
app, socketio = test_app
@ -139,16 +101,6 @@ class TestServiceStatus:
class TestWatchlistCount:
"""watchlist:count socket events match GET /api/watchlist/count."""
def test_watchlist_count_received(self, test_app, shared_state):
"""Client receives a watchlist:count event."""
app, socketio = test_app
client = socketio.test_client(app)
build = shared_state['build_watchlist_count_payload']
socketio.emit('watchlist:count', build())
received = client.get_received()
wl_events = [e for e in received if e['name'] == 'watchlist:count']
assert len(wl_events) >= 1
def test_watchlist_count_shape(self, test_app, shared_state):
"""watchlist:count event data has expected keys."""
app, socketio = test_app
@ -222,12 +174,6 @@ class TestDownloadBatch:
with lock:
batches[batch_id] = defaults
def test_download_subscribe(self, test_app):
"""Client can subscribe to a batch room."""
app, socketio = test_app
client = socketio.test_client(app)
client.emit('downloads:subscribe', {'batch_ids': ['batch_abc']})
def test_download_receives_updates(self, test_app, shared_state):
"""After subscribing, client receives batch_update for that batch."""
app, socketio = test_app
@ -338,47 +284,6 @@ class TestDownloadBatch:
assert 'active_count' in data
assert 'max_concurrent' in data
def test_download_http_batch_still_works(self, test_app, shared_state):
"""HTTP batch endpoint works alongside WebSocket rooms."""
app, socketio = test_app
flask_client = app.test_client()
self._add_batch(shared_state, 'batch_http')
resp = flask_client.get('/api/download_status/batch?batch_ids=batch_http')
assert resp.status_code == 200
data = resp.get_json()
assert 'batch_http' in data['batches']
assert data['batches']['batch_http']['phase'] == 'downloading'
# =========================================================================
# Group E — Fallback Behavior
# =========================================================================
class TestFallback:
"""HTTP endpoints work when no WebSocket is connected."""
def test_http_works_without_websocket(self, flask_client):
"""All three HTTP endpoints work without any WebSocket connection."""
# Status
resp = flask_client.get('/status')
assert resp.status_code == 200
data = resp.get_json()
assert data['spotify']['connected'] is True
# Watchlist
resp = flask_client.get('/api/watchlist/count')
assert resp.status_code == 200
data = resp.get_json()
assert data['count'] == 7
# Download batch (empty — no active batches)
resp = flask_client.get('/api/download_status/batch')
assert resp.status_code == 200
data = resp.get_json()
assert data['batches'] == {}
def test_multiple_clients_get_updates(self, test_app, shared_state):
"""Multiple WebSocket clients each receive broadcast events."""
app, socketio = test_app

1248
tools/test_youtube_download.py
View file

File diff suppressed because it is too large Load diff

BIN
ui/assets/jellyfin_icon.png
View file

Binary file not shown.
Side by side

Before

Width:  |  Height:  |  Size: 230 KiB

BIN
ui/assets/navidrome_icon.png
View file

Binary file not shown.
Side by side

Before

Width:  |  Height:  |  Size: 13 KiB

BIN
ui/assets/plex_icon.png
View file

Binary file not shown.
Side by side

Before

Width:  |  Height:  |  Size: 52 KiB

396
ui/components/database_updater_widget.py
View file

@ -1,396 +0,0 @@
#!/usr/bin/env python3
from PyQt6.QtWidgets import (QFrame, QVBoxLayout, QHBoxLayout, QLabel,
QPushButton, QProgressBar, QComboBox, QGroupBox)
from PyQt6.QtCore import Qt
from PyQt6.QtGui import QFont
from utils.logging_config import get_logger
logger = get_logger("database_updater_widget")
class DatabaseUpdaterWidget(QFrame):
"""UI widget for updating SoulSync database with media server library data (Plex, Jellyfin, or Navidrome)"""
def __init__(self, parent=None):
super().__init__(parent)
self.setup_ui()
def setup_ui(self):
self.setStyleSheet("""
DatabaseUpdaterWidget {
background: #282828;
border-radius: 8px;
border: 1px solid #404040;
}
""")
layout = QVBoxLayout(self)
layout.setContentsMargins(20, 15, 20, 15)
layout.setSpacing(12)
# Header
header_label = QLabel("Update SoulSync Database")
header_label.setFont(QFont("Arial", 14, QFont.Weight.Bold))
header_label.setStyleSheet("color: #ffffff;")
# Info label - dynamic based on active server
try:
from config.settings import config_manager
active_server = config_manager.get_active_media_server()
if active_server == "jellyfin":
server_name = "Jellyfin"
elif active_server == "navidrome":
server_name = "Navidrome"
else:
server_name = "Plex"
except:
server_name = "Plex" # Fallback
info_label = QLabel(f"Syncs your {server_name} music library into the local database for faster searches and analytics")
info_label.setFont(QFont("Arial", 9))
info_label.setStyleSheet("color: #b3b3b3; margin-bottom: 5px;")
info_label.setWordWrap(True)
# Recommendation label
self.recommendation_label = QLabel("Tip: Run a Full Refresh every 1-2 weeks to ensure database accuracy")
self.recommendation_label.setFont(QFont("Arial", 9))
self.recommendation_label.setStyleSheet("color: #ffaa00; margin-bottom: 8px; padding: 6px 8px; background: #332200; border-radius: 4px;")
self.recommendation_label.setWordWrap(True)
# Last full refresh label
self.last_refresh_label = QLabel("")
self.last_refresh_label.setFont(QFont("Arial", 8))
self.last_refresh_label.setStyleSheet("color: #888888; margin-bottom: 5px;")
self.last_refresh_label.setWordWrap(True)
# Control section
control_layout = QVBoxLayout()
control_layout.setSpacing(12)
# Top row: Button
button_layout = QHBoxLayout()
self.start_button = QPushButton("Update Database")
self.start_button.setFixedHeight(36)
self.start_button.setFont(QFont("Arial", 10, QFont.Weight.Medium))
self.start_button.setStyleSheet("""
QPushButton {
background: #1db954;
color: white;
border: none;
border-radius: 6px;
padding: 8px 16px;
}
QPushButton:hover {
background: #1ed760;
}
QPushButton:pressed {
background: #169c46;
}
QPushButton:disabled {
background: #555555;
color: #999999;
}
""")
button_layout.addWidget(self.start_button)
button_layout.addStretch()
# Bottom row: Settings and status
settings_layout = QHBoxLayout()
settings_layout.setSpacing(25)
# Update type dropdown
update_type_layout = QVBoxLayout()
update_type_layout.setSpacing(4)
type_label = QLabel("Update Type:")
type_label.setFont(QFont("Arial", 9))
type_label.setStyleSheet("color: #b3b3b3;")
self.update_type_combo = QComboBox()
self.update_type_combo.setFixedHeight(32)
self.update_type_combo.setFont(QFont("Arial", 10))
self.update_type_combo.addItems([
"Incremental Update",
"Full Refresh"
])
self.update_type_combo.setCurrentText("Incremental Update")
self.update_type_combo.setStyleSheet("""
QComboBox {
background: #333333;
color: #ffffff;
border: 1px solid #555555;
border-radius: 4px;
padding: 4px 8px;
min-width: 140px;
}
QComboBox:hover {
border: 1px solid #1db954;
}
QComboBox::drop-down {
border: none;
width: 20px;
}
QComboBox::down-arrow {
image: none;
border-left: 5px solid transparent;
border-right: 5px solid transparent;
border-top: 5px solid #ffffff;
margin-right: 5px;
}
QComboBox QAbstractItemView {
background: #333333;
color: #ffffff;
border: 1px solid #555555;
selection-background-color: #1db954;
}
""")
update_type_layout.addWidget(type_label)
update_type_layout.addWidget(self.update_type_combo)
# Current status display
status_layout = QVBoxLayout()
status_layout.setSpacing(4)
current_label = QLabel("Current Status:")
current_label.setFont(QFont("Arial", 9))
current_label.setStyleSheet("color: #b3b3b3;")
self.current_status_label = QLabel("Ready")
self.current_status_label.setFont(QFont("Arial", 11, QFont.Weight.Medium))
self.current_status_label.setStyleSheet("color: #ffffff;")
status_layout.addWidget(current_label)
status_layout.addWidget(self.current_status_label)
settings_layout.addLayout(update_type_layout)
settings_layout.addLayout(status_layout)
settings_layout.addStretch()
control_layout.addLayout(button_layout)
control_layout.addLayout(settings_layout)
# Progress section
progress_layout = QVBoxLayout()
progress_layout.setSpacing(8)
progress_info_layout = QHBoxLayout()
self.progress_label = QLabel("Progress: 0%")
self.progress_label.setFont(QFont("Arial", 10))
self.progress_label.setStyleSheet("color: #ffffff;")
self.count_label = QLabel("0 artists processed")
self.count_label.setFont(QFont("Arial", 9))
self.count_label.setStyleSheet("color: #b3b3b3;")
progress_info_layout.addWidget(self.progress_label)
progress_info_layout.addStretch()
progress_info_layout.addWidget(self.count_label)
self.progress_bar = QProgressBar()
self.progress_bar.setFixedHeight(8)
self.progress_bar.setRange(0, 100)
self.progress_bar.setValue(0)
self.progress_bar.setStyleSheet("""
QProgressBar {
border: none;
border-radius: 4px;
background: #555555;
}
QProgressBar::chunk {
background: #1db954;
border-radius: 4px;
}
""")
progress_layout.addLayout(progress_info_layout)
progress_layout.addWidget(self.progress_bar)
# Statistics section (shows current database info)
stats_group = QGroupBox("Database Statistics")
stats_group.setFont(QFont("Arial", 10, QFont.Weight.Bold))
stats_group.setStyleSheet("""
QGroupBox {
color: #ffffff;
border: 1px solid #555555;
border-radius: 6px;
margin-top: 6px;
padding-top: 10px;
}
QGroupBox::title {
subcontrol-origin: margin;
left: 10px;
padding: 0 5px 0 5px;
}
""")
stats_layout = QHBoxLayout(stats_group)
stats_layout.setSpacing(20)
# Artists stat
artists_layout = QVBoxLayout()
self.artists_count_label = QLabel("0")
self.artists_count_label.setFont(QFont("Arial", 16, QFont.Weight.Bold))
self.artists_count_label.setStyleSheet("color: #1db954;")
self.artists_count_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
artists_text_label = QLabel("Artists")
artists_text_label.setFont(QFont("Arial", 9))
artists_text_label.setStyleSheet("color: #b3b3b3;")
artists_text_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
artists_layout.addWidget(self.artists_count_label)
artists_layout.addWidget(artists_text_label)
# Albums stat
albums_layout = QVBoxLayout()
self.albums_count_label = QLabel("0")
self.albums_count_label.setFont(QFont("Arial", 16, QFont.Weight.Bold))
self.albums_count_label.setStyleSheet("color: #1db954;")
self.albums_count_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
albums_text_label = QLabel("Albums")
albums_text_label.setFont(QFont("Arial", 9))
albums_text_label.setStyleSheet("color: #b3b3b3;")
albums_text_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
albums_layout.addWidget(self.albums_count_label)
albums_layout.addWidget(albums_text_label)
# Tracks stat
tracks_layout = QVBoxLayout()
self.tracks_count_label = QLabel("0")
self.tracks_count_label.setFont(QFont("Arial", 16, QFont.Weight.Bold))
self.tracks_count_label.setStyleSheet("color: #1db954;")
self.tracks_count_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
tracks_text_label = QLabel("Tracks")
tracks_text_label.setFont(QFont("Arial", 9))
tracks_text_label.setStyleSheet("color: #b3b3b3;")
tracks_text_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
tracks_layout.addWidget(self.tracks_count_label)
tracks_layout.addWidget(tracks_text_label)
# Database size stat
size_layout = QVBoxLayout()
self.size_label = QLabel("0.0 MB")
self.size_label.setFont(QFont("Arial", 16, QFont.Weight.Bold))
self.size_label.setStyleSheet("color: #1db954;")
self.size_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
size_text_label = QLabel("DB Size")
size_text_label.setFont(QFont("Arial", 9))
size_text_label.setStyleSheet("color: #b3b3b3;")
size_text_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
size_layout.addWidget(self.size_label)
size_layout.addWidget(size_text_label)
stats_layout.addLayout(artists_layout)
stats_layout.addLayout(albums_layout)
stats_layout.addLayout(tracks_layout)
stats_layout.addLayout(size_layout)
stats_layout.addStretch()
# Add all sections to main layout
layout.addWidget(header_label)
layout.addWidget(info_label)
layout.addWidget(self.recommendation_label)
layout.addWidget(self.last_refresh_label)
layout.addLayout(control_layout)
layout.addLayout(progress_layout)
layout.addWidget(stats_group)
def update_progress(self, is_running: bool, current_item: str, processed: int, total: int, percentage: float):
"""Update progress display during database update"""
if is_running:
self.start_button.setText("Stop Update")
self.start_button.setEnabled(True)
self.current_status_label.setText(current_item if current_item else "Processing...")
self.progress_label.setText(f"Progress: {percentage:.1f}%")
self.count_label.setText(f"{processed} / {total} artists processed")
self.progress_bar.setValue(int(percentage))
else:
self.start_button.setText("Update Database")
self.start_button.setEnabled(True)
self.current_status_label.setText("Ready")
self.progress_label.setText("Progress: 0%")
self.count_label.setText("0 artists processed")
self.progress_bar.setValue(0)
def update_statistics(self, stats: dict):
"""Update database statistics display"""
self.artists_count_label.setText(str(stats.get('artists', 0)))
self.albums_count_label.setText(str(stats.get('albums', 0)))
self.tracks_count_label.setText(str(stats.get('tracks', 0)))
self.size_label.setText(f"{stats.get('database_size_mb', 0.0):.1f} MB")
def update_phase(self, phase: str):
"""Update current phase display"""
self.current_status_label.setText(phase)
def is_full_refresh(self) -> bool:
"""Check if full refresh is selected"""
return self.update_type_combo.currentText() == "Full Refresh"
def set_button_text(self, text: str):
"""Set custom button text"""
self.start_button.setText(text)
def set_button_enabled(self, enabled: bool):
"""Enable/disable the start button"""
self.start_button.setEnabled(enabled)
def update_last_refresh_info(self, last_refresh_date: str = None):
"""Update the last refresh information with color-coded warnings"""
if not last_refresh_date:
self.last_refresh_label.setText("No full refresh recorded")
self.last_refresh_label.setStyleSheet("color: #ff6666; margin-bottom: 5px; font-style: italic;")
self._update_recommendation_urgency(urgent=True)
return
try:
from datetime import datetime
last_date = datetime.fromisoformat(last_refresh_date.replace('Z', '+00:00'))
days_ago = (datetime.now() - last_date.replace(tzinfo=None)).days
if days_ago == 0:
time_text = "today"
color = "#1db954" # Green
urgent = False
elif days_ago == 1:
time_text = "yesterday"
color = "#1db954" # Green
urgent = False
elif days_ago < 7:
time_text = f"{days_ago} days ago"
color = "#1db954" # Green
urgent = False
elif days_ago < 14:
time_text = f"{days_ago} days ago"
color = "#ffaa00" # Orange warning
urgent = False
else:
time_text = f"{days_ago} days ago"
color = "#ff6666" # Red warning
urgent = True
self.last_refresh_label.setText(f"Last full refresh: {time_text}")
self.last_refresh_label.setStyleSheet(f"color: {color}; margin-bottom: 5px;")
self._update_recommendation_urgency(urgent=urgent)
except Exception:
self.last_refresh_label.setText("Last full refresh: unknown")
self.last_refresh_label.setStyleSheet("color: #888888; margin-bottom: 5px;")
self._update_recommendation_urgency(urgent=False)
def _update_recommendation_urgency(self, urgent: bool = False):
"""Update the recommendation label styling based on urgency"""
if urgent:
self.recommendation_label.setText("Recommended: Run a Full Refresh - it's been over 2 weeks!")
self.recommendation_label.setStyleSheet("color: #ffffff; margin-bottom: 8px; padding: 6px 8px; background: #cc3300; border-radius: 4px;")
else:
self.recommendation_label.setText("Tip: Run a Full Refresh every 1-2 weeks to ensure database accuracy")
self.recommendation_label.setStyleSheet("color: #ffaa00; margin-bottom: 8px; padding: 6px 8px; background: #332200; border-radius: 4px;")

260
ui/components/toast_manager.py
View file

@ -1,260 +0,0 @@
from PyQt6.QtWidgets import QWidget, QLabel, QHBoxLayout, QVBoxLayout, QGraphicsOpacityEffect
from PyQt6.QtCore import Qt, QTimer, QPropertyAnimation, QEasingCurve, pyqtSignal, QRect
from PyQt6.QtGui import QFont, QPainter, QPaintEvent
import time
from typing import List, Optional
from enum import Enum
class ToastType(Enum):
SUCCESS = "success"
INFO = "info"
WARNING = "warning"
ERROR = "error"
class Toast(QWidget):
"""Individual toast notification widget"""
closed = pyqtSignal(object) # Emits self when closing
def __init__(self, message: str, toast_type: ToastType = ToastType.INFO, duration: int = 4000, parent=None):
super().__init__(parent)
self.message = message
self.toast_type = toast_type
self.duration = duration
self.created_time = time.time()
self.setup_ui()
self.setup_animations()
self.setup_auto_dismiss()
def setup_ui(self):
"""Setup the toast UI"""
self.setFixedHeight(60)
self.setMinimumWidth(300)
self.setMaximumWidth(400)
# Make the widget click-through for the background but clickable for the content
self.setAttribute(Qt.WidgetAttribute.WA_TranslucentBackground)
self.setWindowFlags(Qt.WindowType.FramelessWindowHint | Qt.WindowType.WindowStaysOnTopHint)
# Main layout
layout = QHBoxLayout(self)
layout.setContentsMargins(12, 12, 12, 12)
layout.setSpacing(12)
# Icon label
self.icon_label = QLabel()
self.icon_label.setFont(QFont("Segoe UI", 14))
self.icon_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.icon_label.setFixedSize(24, 24)
# Message label
self.message_label = QLabel(self.message)
self.message_label.setFont(QFont("Segoe UI", 10))
self.message_label.setWordWrap(True)
self.message_label.setAlignment(Qt.AlignmentFlag.AlignVCenter)
layout.addWidget(self.icon_label)
layout.addWidget(self.message_label, 1)
# Apply styling based on toast type
self.apply_styling()
def apply_styling(self):
"""Apply styling based on toast type"""
if self.toast_type == ToastType.SUCCESS:
icon = ""
accent_color = "#1db954" # Spotify green
bg_color = "rgba(29, 185, 84, 0.15)"
border_color = "rgba(29, 185, 84, 0.3)"
elif self.toast_type == ToastType.ERROR:
icon = ""
accent_color = "#f04747"
bg_color = "rgba(240, 71, 71, 0.15)"
border_color = "rgba(240, 71, 71, 0.3)"
elif self.toast_type == ToastType.WARNING:
icon = ""
accent_color = "#ffa500"
bg_color = "rgba(255, 165, 0, 0.15)"
border_color = "rgba(255, 165, 0, 0.3)"
else: # INFO
icon = "ℹ️"
accent_color = "#5865f2"
bg_color = "rgba(88, 101, 242, 0.15)"
border_color = "rgba(88, 101, 242, 0.3)"
self.icon_label.setText(icon)
self.setStyleSheet(f"""
Toast {{
background: qlineargradient(x1:0, y1:0, x2:0, y2:1,
stop:0 rgba(45, 45, 45, 0.95),
stop:1 rgba(35, 35, 35, 0.95));
border: 1px solid {border_color};
border-left: 3px solid {accent_color};
border-radius: 8px;
}}
""")
self.message_label.setStyleSheet(f"""
color: #ffffff;
background: transparent;
""")
def setup_animations(self):
"""Setup slide-in and fade-out animations"""
# Opacity effect for fade animations
self.opacity_effect = QGraphicsOpacityEffect()
self.setGraphicsEffect(self.opacity_effect)
# Slide-in animation (from right)
self.slide_animation = QPropertyAnimation(self, b"geometry")
self.slide_animation.setDuration(300)
self.slide_animation.setEasingCurve(QEasingCurve.Type.OutCubic)
# Fade-out animation
self.fade_animation = QPropertyAnimation(self.opacity_effect, b"opacity")
self.fade_animation.setDuration(200)
self.fade_animation.setEasingCurve(QEasingCurve.Type.OutQuad)
# Connect fade animation to close
self.fade_animation.finished.connect(self._on_fade_complete)
def setup_auto_dismiss(self):
"""Setup auto-dismiss timer"""
if self.duration > 0:
self.dismiss_timer = QTimer()
self.dismiss_timer.setSingleShot(True)
self.dismiss_timer.timeout.connect(self.dismiss)
self.dismiss_timer.start(self.duration)
def show_at_position(self, target_rect: QRect):
"""Show the toast with slide-in animation at the specified position"""
# Start position (off-screen to the right)
start_rect = QRect(target_rect.x() + 50, target_rect.y(), target_rect.width(), target_rect.height())
# Set initial position and show
self.setGeometry(start_rect)
self.show()
# Animate to target position
self.slide_animation.setStartValue(start_rect)
self.slide_animation.setEndValue(target_rect)
self.slide_animation.start()
def dismiss(self):
"""Dismiss the toast with fade-out animation"""
if hasattr(self, 'dismiss_timer'):
self.dismiss_timer.stop()
self.fade_animation.setStartValue(1.0)
self.fade_animation.setEndValue(0.0)
self.fade_animation.start()
def _on_fade_complete(self):
"""Called when fade animation completes"""
self.closed.emit(self)
self.hide()
self.deleteLater()
def mousePressEvent(self, event):
"""Handle click to dismiss"""
if event.button() == Qt.MouseButton.LeftButton:
self.dismiss()
super().mousePressEvent(event)
class ToastManager(QWidget):
"""Manages multiple toast notifications"""
def __init__(self, parent=None):
super().__init__(parent)
self.parent_widget = parent
self.active_toasts: List[Toast] = []
self.toast_spacing = 10
self.margin_from_edge = 20
# Make this widget transparent and non-interactive
self.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents)
self.setAttribute(Qt.WidgetAttribute.WA_TranslucentBackground)
def show_toast(self, message: str, toast_type: ToastType = ToastType.INFO, duration: int = 4000):
"""Show a new toast notification"""
toast = Toast(message, toast_type, duration, self.parent_widget)
toast.closed.connect(self._on_toast_closed)
# Calculate position for this toast
position = self._calculate_toast_position(len(self.active_toasts))
# Add to active toasts list
self.active_toasts.append(toast)
# Show the toast
toast.show_at_position(position)
# Reposition existing toasts if needed
self._reposition_existing_toasts()
def _calculate_toast_position(self, index: int) -> QRect:
"""Calculate position for a toast at the given index"""
if not self.parent_widget:
return QRect(0, 0, 350, 60)
parent_rect = self.parent_widget.rect()
toast_height = 60
toast_width = 350
x = parent_rect.width() - toast_width - self.margin_from_edge
y = self.margin_from_edge + (index * (toast_height + self.toast_spacing))
return QRect(x, y, toast_width, toast_height)
def _reposition_existing_toasts(self):
"""Reposition existing toasts to make room for new ones"""
for i, toast in enumerate(self.active_toasts[:-1]): # Exclude the newest toast
new_position = self._calculate_toast_position(i)
# Animate to new position if needed
current_geo = toast.geometry()
if current_geo.y() != new_position.y():
toast.slide_animation.stop()
toast.slide_animation.setStartValue(current_geo)
toast.slide_animation.setEndValue(new_position)
toast.slide_animation.start()
def _on_toast_closed(self, toast: Toast):
"""Handle when a toast is closed"""
if toast in self.active_toasts:
self.active_toasts.remove(toast)
# Reposition remaining toasts
for i, remaining_toast in enumerate(self.active_toasts):
new_position = self._calculate_toast_position(i)
current_geo = remaining_toast.geometry()
if current_geo != new_position:
remaining_toast.slide_animation.stop()
remaining_toast.slide_animation.setStartValue(current_geo)
remaining_toast.slide_animation.setEndValue(new_position)
remaining_toast.slide_animation.start()
def clear_all_toasts(self):
"""Dismiss all active toasts"""
for toast in self.active_toasts.copy():
toast.dismiss()
# Convenience methods for different toast types
def success(self, message: str, duration: int = 4000):
"""Show a success toast"""
self.show_toast(message, ToastType.SUCCESS, duration)
def error(self, message: str, duration: int = 6000):
"""Show an error toast (longer duration)"""
self.show_toast(message, ToastType.ERROR, duration)
def warning(self, message: str, duration: int = 5000):
"""Show a warning toast"""
self.show_toast(message, ToastType.WARNING, duration)
def info(self, message: str, duration: int = 4000):
"""Show an info toast"""
self.show_toast(message, ToastType.INFO, duration)

293
ui/components/version_info_modal.py
View file

@ -1,293 +0,0 @@
#!/usr/bin/env python3
from PyQt6.QtWidgets import (QDialog, QVBoxLayout, QHBoxLayout, QLabel,
QPushButton, QFrame, QScrollArea, QWidget)
from PyQt6.QtCore import Qt
from PyQt6.QtGui import QFont
from utils.logging_config import get_logger
logger = get_logger("version_info_modal")
class VersionInfoModal(QDialog):
"""Modal displaying recent changes and version information"""
def __init__(self, parent=None):
super().__init__(parent)
self.setWindowTitle("What's New in SoulSync v1.0")
self.setModal(True)
self.setFixedSize(600, 500)
self.setup_ui()
def setup_ui(self):
self.setStyleSheet("""
VersionInfoModal {
background: #1a1a1a;
border-radius: 12px;
}
""")
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
# Header
header = self.create_header()
layout.addWidget(header)
# Content area with scroll
content_area = self.create_content_area()
layout.addWidget(content_area)
# Footer with close button
footer = self.create_footer()
layout.addWidget(footer)
def create_header(self):
header = QFrame()
header.setFixedHeight(80)
header.setStyleSheet("""
QFrame {
background: #1a1a1a;
border-top-left-radius: 12px;
border-top-right-radius: 12px;
border-bottom: 1px solid rgba(255, 255, 255, 0.1);
}
""")
layout = QVBoxLayout(header)
layout.setContentsMargins(30, 20, 30, 15)
layout.setSpacing(5)
# Title
title = QLabel("What's New in SoulSync")
title.setFont(QFont("SF Pro Display", 18, QFont.Weight.Bold))
title.setStyleSheet("""
color: #ffffff;
letter-spacing: -0.5px;
font-weight: 700;
""")
# Version subtitle
version_subtitle = QLabel("Version 1.0 - Complete WebUI Rebuild")
version_subtitle.setFont(QFont("SF Pro Text", 11, QFont.Weight.Medium))
version_subtitle.setStyleSheet("""
color: rgba(255, 255, 255, 0.7);
letter-spacing: 0.1px;
margin-top: 2px;
""")
layout.addWidget(title)
layout.addWidget(version_subtitle)
return header
def create_content_area(self):
# Scroll area for content
scroll_area = QScrollArea()
scroll_area.setWidgetResizable(True)
scroll_area.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
scroll_area.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAsNeeded)
scroll_area.setStyleSheet("""
QScrollArea {
border: none;
background: #1a1a1a;
}
QScrollBar:vertical {
background: #2a2a2a;
width: 8px;
border-radius: 4px;
}
QScrollBar::handle:vertical {
background: #555555;
border-radius: 4px;
}
QScrollBar::handle:vertical:hover {
background: #666666;
}
""")
# Content widget
content_widget = QWidget()
content_layout = QVBoxLayout(content_widget)
content_layout.setContentsMargins(30, 25, 30, 25)
content_layout.setSpacing(25)
# WebUI Transformation
webui_section = self.create_feature_section(
"Complete WebUI Transformation",
"SoulSync has been completely rebuilt from the ground up as a modern web application, moving from desktop GUI to web-based interface",
[
"• Full transition from PyQt6 desktop application to responsive web interface",
"• Modern HTML5, CSS3, and JavaScript implementation with premium glassmorphic design",
"• Real-time updates and live status monitoring through WebSocket connections",
"• Cross-platform compatibility - access from any device with a web browser",
"• Mobile-responsive design optimized for tablets and smartphones",
"• Dark theme with sophisticated visual effects and smooth animations",
"• RESTful API architecture enabling future third-party integrations"
],
"Access SoulSync through your web browser at localhost:8888 - no desktop installation required!"
)
content_layout.addWidget(webui_section)
# Docker Support
docker_section = self.create_feature_section(
"Docker Container Support",
"Complete containerization with Docker for easy deployment and scalability",
[
"• Pre-built Docker images available for instant deployment",
"• Multi-architecture support (AMD64, ARM64) for various server platforms",
"• Volume mounting for persistent configuration and downloads",
"• Environment variable configuration for easy customization",
"• Docker Compose templates for simplified multi-container setups",
"• Automatic health checks and restart policies for reliability",
"• Lightweight Alpine Linux base for minimal resource usage"
]
)
content_layout.addWidget(docker_section)
# Enhanced Music Management
music_section = self.create_feature_section(
"Enhanced Music Management",
"All beloved features preserved and enhanced with new web-based capabilities",
[
"• Complete Spotify, Tidal, and YouTube Music playlist synchronization",
"• Advanced Soulseek integration with real-time download management",
"• Intelligent music matching engine with improved accuracy",
"• Plex and Jellyfin server integration with automatic library updates",
"• Artist watchlist with automatic new release detection",
"• Comprehensive metadata enhancement with high-quality album artwork",
"• Real-time download progress with detailed logging and status updates"
]
)
content_layout.addWidget(music_section)
# Performance & Reliability
performance_section = self.create_feature_section(
"Performance & Reliability",
"Significant improvements in speed, stability, and resource efficiency",
[
"• Asynchronous processing for improved responsiveness",
"• Multi-threaded download management with concurrent processing",
"• Optimized database operations with connection pooling",
"• Intelligent caching system for faster API responses",
"• Robust error handling with automatic retry mechanisms",
"• Memory-efficient architecture suitable for long-running deployments",
"• Comprehensive logging system for easy troubleshooting"
]
)
content_layout.addWidget(performance_section)
scroll_area.setWidget(content_widget)
return scroll_area
def create_feature_section(self, title, description, features, usage_note=None):
section = QFrame()
section.setStyleSheet("""
QFrame {
background: transparent;
border: none;
border-left: 3px solid rgba(29, 185, 84, 0.4);
border-radius: 0px;
padding: 0px;
margin-left: 5px;
}
""")
layout = QVBoxLayout(section)
layout.setContentsMargins(20, 18, 20, 18)
layout.setSpacing(12)
# Section title
title_label = QLabel(title)
title_label.setFont(QFont("SF Pro Text", 14, QFont.Weight.Bold))
title_label.setStyleSheet("""
color: #1ed760;
font-weight: 600;
letter-spacing: -0.2px;
margin-bottom: 3px;
""")
layout.addWidget(title_label)
# Description
desc_label = QLabel(description)
desc_label.setFont(QFont("SF Pro Text", 11))
desc_label.setStyleSheet("""
color: rgba(255, 255, 255, 0.8);
line-height: 1.4;
margin-bottom: 8px;
""")
desc_label.setWordWrap(True)
layout.addWidget(desc_label)
# Features list
for feature in features:
feature_label = QLabel(feature)
feature_label.setFont(QFont("SF Pro Text", 10))
feature_label.setStyleSheet("""
color: rgba(255, 255, 255, 0.7);
line-height: 1.5;
padding-left: 8px;
margin: 2px 0px;
""")
feature_label.setWordWrap(True)
layout.addWidget(feature_label)
# Usage note if provided
if usage_note:
usage_label = QLabel(f"{usage_note}")
usage_label.setFont(QFont("SF Pro Text", 10))
usage_label.setStyleSheet("""
color: #1ed760;
background: transparent;
border: none;
padding: 8px 0px;
margin-top: 8px;
line-height: 1.4;
font-style: italic;
""")
usage_label.setWordWrap(True)
layout.addWidget(usage_label)
return section
def create_footer(self):
footer = QFrame()
footer.setFixedHeight(65)
footer.setStyleSheet("""
QFrame {
background: rgba(255, 255, 255, 0.02);
border-top: 1px solid rgba(255, 255, 255, 0.08);
border-bottom-left-radius: 12px;
border-bottom-right-radius: 12px;
}
""")
layout = QHBoxLayout(footer)
layout.setContentsMargins(30, 15, 30, 15)
# Close button
close_button = QPushButton("Close")
close_button.setFixedSize(100, 35)
close_button.setFont(QFont("SF Pro Text", 10, QFont.Weight.Medium))
close_button.setStyleSheet("""
QPushButton {
background: #1db954;
color: white;
border: none;
border-radius: 6px;
font-weight: 500;
letter-spacing: 0.1px;
}
QPushButton:hover {
background: #1ed760;
}
QPushButton:pressed {
background: #169c46;
}
""")
close_button.clicked.connect(self.accept)
layout.addStretch()
layout.addWidget(close_button)
return footer

1314
ui/components/watchlist_status_modal.py
View file

File diff suppressed because it is too large Load diff

0
ui/pages/__init__.py
View file

5507
ui/pages/artists.py
View file

File diff suppressed because it is too large Load diff

4216
ui/pages/dashboard.py
View file

File diff suppressed because it is too large Load diff

11364
ui/pages/downloads.py
View file

File diff suppressed because it is too large Load diff

3487
ui/pages/settings.py
View file

File diff suppressed because it is too large Load diff

9920
ui/pages/sync.py
View file

File diff suppressed because it is too large Load diff

1367
ui/sidebar.py
View file

File diff suppressed because it is too large Load diff

8
web_server.py
View file

@ -68,7 +68,7 @@ from core.soulseek_client import SoulseekClient
from core.download_orchestrator import DownloadOrchestrator
from core.tidal_client import TidalClient # Added import for Tidal
from core.matching_engine import MusicMatchingEngine
from core.database_update_worker import DatabaseUpdateWorker, DatabaseStatsWorker
from core.database_update_worker import DatabaseUpdateWorker
from core.web_scan_manager import WebScanManager
from core.lyrics_client import lyrics_client
from core.metadata_cache import get_metadata_cache
@ -4066,7 +4066,7 @@ def _find_downloaded_file(download_path, track_data):
return None
# --- Refactored Logic from GUI Threads ---
# This logic is extracted from your QThread classes to be used directly by Flask.
# This logic is extracted from the database update worker to be used directly by Flask.
def run_service_test(service, test_config):
"""
@ -24238,7 +24238,7 @@ def _run_db_update_task(full_refresh, server_type):
db_update_worker.connect_callback('finished', _db_update_finished_callback)
db_update_worker.connect_callback('error', _db_update_error_callback)
# This is a blocking call that runs the QThread's logic
# This is a blocking call that runs the worker logic
db_update_worker.run()
@ -24289,7 +24289,7 @@ def _run_deep_scan_task(server_type):
def get_database_stats():
"""Endpoint to get current database statistics."""
try:
# This logic is adapted from DatabaseStatsWorker
# This endpoint returns the same stats shape the UI expects.
db = get_database()
stats = db.get_database_info_for_server()
return jsonify(stats)