Add comprehensive backend services and API enhancements

- Complete Spotify integration with downloader and settings
- Advanced UX features and audio quality management
- Enhanced search capabilities and mobile offline support
- Music catalog browser and recap features
- Universal downloader and upload functionality
- Update tracking system with database models and migrations
- Comprehensive service layer architecture
- Enhanced lyrics API and streaming capabilities
- Extended application builder and startup configuration
- New logging infrastructure and services directory
This commit is contained in:
Tomas Dvorak
2026-03-17 17:56:20 +01:00
parent 65a1268dab
commit 4338dd1d9c
43 changed files with 19453 additions and 10 deletions
+9 -1
View File
@@ -21,6 +21,14 @@ from swingmusic.api import (
auth,
stream,
backup_and_restore,
spotify,
spotify_settings,
enhanced_search,
universal_downloader,
music_catalog,
update_tracking,
audio_quality,
upload,
)
from swingmusic.api.plugins import lyrics as lyrics_plugin
@@ -28,7 +36,7 @@ from swingmusic.api.plugins import mixes as mixes_plugin
__all__ = [
"album", "artist", "collections", "colors", "favorites", "folder", "imgserver", "playlist", "search", "settings",
"lyrics", "plugins", "scrobble", "home", "getall", "auth", "stream", "backup_and_restore",
"lyrics", "plugins", "scrobble", "home", "getall", "auth", "stream", "backup_and_restore", "spotify", "spotify_settings", "enhanced_search", "universal_downloader", "music_catalog", "update_tracking", "audio_quality", "upload",
"lyrics_plugin",
"mixes_plugin"
+624
View File
@@ -0,0 +1,624 @@
"""
Advanced UX API Endpoints
This module provides REST API endpoints for enhanced user experience features,
including intelligent search suggestions, recommendations, and personalization.
"""
import logging
from datetime import datetime
from typing import Dict, List, Optional, Any
from flask import Blueprint, request, jsonify
from flask_login import login_required, current_user
from swingmusic.db import db
from swingmusic.services.advanced_ux_service import advanced_ux_service, SuggestionType, SearchContext
from swingmusic.utils.request import APIError, success_response, error_response
from swingmusic.utils.validators import validate_search_query, validate_context
logger = logging.getLogger(__name__)
advanced_ux_bp = Blueprint('advanced_ux', __name__, url_prefix='/api/ux')
def get_current_user_id() -> int:
"""Get current user ID from Flask-Login"""
return current_user.id if current_user.is_authenticated else None
@advanced_ux_bp.route('/search/suggestions', methods=['GET'])
@login_required
async def get_search_suggestions():
"""
Get intelligent search suggestions
Query Parameters:
- q: Search query
- context: Search context (general, discovery, download, playlist, offline, social)
- limit: Maximum suggestions to return (default: 10)
"""
try:
user_id = get_current_user_id()
query = request.args.get('q', '').strip()
context_str = request.args.get('context', 'general')
limit = min(request.args.get('limit', 10, type=int), 50)
# Validate inputs
validate_search_query(query)
context = validate_context(context_str)
# Get suggestions
suggestions = await advanced_ux_service.get_search_suggestions(user_id, query, context, limit)
# Format response
formatted_suggestions = []
for suggestion in suggestions:
formatted_suggestion = {
'id': suggestion.id,
'type': suggestion.type.value,
'title': suggestion.title,
'subtitle': suggestion.subtitle,
'image_url': suggestion.image_url,
'url': suggestion.url,
'metadata': suggestion.metadata,
'relevance_score': suggestion.relevance_score,
'context': suggestion.context.value
}
formatted_suggestions.append(formatted_suggestion)
return success_response({
'suggestions': formatted_suggestions,
'query': query,
'context': context.value,
'total_count': len(formatted_suggestions)
})
except Exception as e:
logger.error(f"Error getting search suggestions: {e}")
return error_response("Internal server error", 500)
@advanced_ux_bp.route('/discovery/recommendations', methods=['GET'])
@login_required
async def get_discovery_recommendations():
"""
Get personalized discovery recommendations
Query Parameters:
- type: Recommendation type (tracks, artists, albums, mixed)
- limit: Maximum recommendations to return (default: 20)
"""
try:
user_id = get_current_user_id()
recommendation_type = request.args.get('type', 'mixed')
limit = min(request.args.get('limit', 20, type=int), 100)
# Validate recommendation type
valid_types = ['tracks', 'artists', 'albums', 'mixed']
if recommendation_type not in valid_types:
return error_response(f"Invalid type. Must be one of: {valid_types}", 400)
# Get recommendations
recommendations = await advanced_ux_service.get_discovery_recommendations(user_id, recommendation_type, limit)
# Format response
formatted_recommendations = []
for recommendation in recommendations:
formatted_recommendation = {
'id': recommendation.id,
'type': recommendation.type.value,
'title': recommendation.title,
'subtitle': recommendation.subtitle,
'image_url': recommendation.image_url,
'url': recommendation.url,
'metadata': recommendation.metadata,
'relevance_score': recommendation.relevance_score
}
formatted_recommendations.append(formatted_recommendation)
return success_response({
'recommendations': formatted_recommendations,
'type': recommendation_type,
'total_count': len(formatted_recommendations)
})
except Exception as e:
logger.error(f"Error getting discovery recommendations: {e}")
return error_response("Internal server error", 500)
@advanced_ux_bp.route('/contextual/suggestions', methods=['GET'])
@login_required
async def get_contextual_suggestions():
"""
Get contextual suggestions based on current track
Query Parameters:
- track_id: Currently playing track ID
- context_type: Context type (similar, same_artist, same_genre, popular)
"""
try:
user_id = get_current_user_id()
track_id = request.args.get('track_id')
context_type = request.args.get('context_type', 'similar')
if not track_id:
return error_response("track_id is required", 400)
# Validate context type
valid_contexts = ['similar', 'same_artist', 'same_genre', 'popular']
if context_type not in valid_contexts:
return error_response(f"Invalid context_type. Must be one of: {valid_contexts}", 400)
# Get contextual suggestions
suggestions = await advanced_ux_service.get_contextual_suggestions(user_id, track_id, context_type)
# Format response
formatted_suggestions = []
for suggestion in suggestions:
formatted_suggestion = {
'id': suggestion.id,
'type': suggestion.type.value,
'title': suggestion.title,
'subtitle': suggestion.subtitle,
'image_url': suggestion.image_url,
'url': suggestion.url,
'metadata': suggestion.metadata,
'relevance_score': suggestion.relevance_score
}
formatted_suggestions.append(formatted_suggestion)
return success_response({
'suggestions': formatted_suggestions,
'track_id': track_id,
'context_type': context_type,
'total_count': len(formatted_suggestions)
})
except Exception as e:
logger.error(f"Error getting contextual suggestions: {e}")
return error_response("Internal server error", 500)
@advanced_ux_bp.route('/download/suggestions', methods=['GET'])
@login_required
async def get_download_suggestions():
"""
Get download-specific suggestions with universal downloader integration
Query Parameters:
- q: Search query (optional)
- limit: Maximum suggestions to return (default: 15)
"""
try:
user_id = get_current_user_id()
query = request.args.get('q', '').strip()
limit = min(request.args.get('limit', 15, type=int), 50)
# Get download suggestions
suggestions = await advanced_ux_service.get_download_suggestions(user_id, query, limit)
# Format response
formatted_suggestions = []
for suggestion in suggestions:
formatted_suggestion = {
'id': suggestion.id,
'type': suggestion.type.value,
'title': suggestion.title,
'subtitle': suggestion.subtitle,
'image_url': suggestion.image_url,
'url': suggestion.url,
'metadata': suggestion.metadata,
'relevance_score': suggestion.relevance_score
}
formatted_suggestions.append(formatted_suggestion)
return success_response({
'suggestions': formatted_suggestions,
'query': query,
'total_count': len(formatted_suggestions)
})
except Exception as e:
logger.error(f"Error getting download suggestions: {e}")
return error_response("Internal server error", 500)
@advanced_ux_bp.route('/search/filters', methods=['GET'])
@login_required
async def get_enhanced_search_filters():
"""
Get enhanced search filters with user personalization
"""
try:
user_id = get_current_user_id()
# Get enhanced filters
filters = await advanced_ux_service.get_enhanced_search_filters(user_id)
# Format response
formatted_filters = []
for filter_item in filters:
formatted_filter = {
'filter_id': filter_item.filter_id,
'name': filter_item.name,
'type': filter_item.type,
'options': filter_item.options,
'is_active': filter_item.is_active,
'is_multi_select': filter_item.is_multi_select
}
formatted_filters.append(formatted_filter)
return success_response({
'filters': formatted_filters,
'total_count': len(formatted_filters)
})
except Exception as e:
logger.error(f"Error getting enhanced search filters: {e}")
return error_response("Internal server error", 500)
@advanced_ux_bp.route('/behavior/track', methods=['POST'])
@login_required
async def track_user_behavior():
"""
Track user behavior for personalization
Request Body:
{
"type": "search|play|download|like",
"data": {
"query": "search query",
"track_id": "track_id",
"artist": "artist_name",
"timestamp": "ISO timestamp",
"context": "context information"
}
}
"""
try:
user_id = get_current_user_id()
data = request.get_json()
if not data:
return error_response("Request body is required", 400)
interaction_type = data.get('type')
interaction_data = data.get('data', {})
# Validate interaction type
valid_types = ['search', 'play', 'download', 'like']
if interaction_type not in valid_types:
return error_response(f"Invalid type. Must be one of: {valid_types}", 400)
# Add user ID and timestamp to interaction data
interaction_data['user_id'] = user_id
if 'timestamp' not in interaction_data:
interaction_data['timestamp'] = datetime.utcnow().isoformat()
# Update user behavior
await advanced_ux_service.update_user_behavior(user_id, interaction_data)
return success_response({
'message': 'User behavior tracked successfully'
})
except Exception as e:
logger.error(f"Error tracking user behavior: {e}")
return error_response("Internal server error", 500)
@advanced_ux_bp.route('/behavior/profile', methods=['GET'])
@login_required
async def get_user_behavior_profile():
"""
Get user behavior profile for personalization insights
"""
try:
user_id = get_current_user_id()
# Get user behavior
behavior = await advanced_ux_service._get_user_behavior(user_id)
# Format response
profile = {
'user_id': behavior.user_id,
'favorite_genres': behavior.favorite_genres,
'favorite_artists': behavior.favorite_artists,
'listening_patterns': behavior.listening_patterns,
'download_preferences': behavior.download_preferences,
'interaction_patterns': behavior.interaction_patterns,
'last_updated': behavior.last_updated.isoformat(),
'search_history_count': len(behavior.search_history),
'recent_searches': behavior.search_history[-5:] if behavior.search_history else []
}
return success_response({
'profile': profile
})
except Exception as e:
logger.error(f"Error getting user behavior profile: {e}")
return error_response("Internal server error", 500)
@advanced_ux_bp.route('/trending/content', methods=['GET'])
@login_required
async def get_trending_content():
"""
Get trending content based on user preferences and global trends
Query Parameters:
- type: Content type (tracks, artists, albums, mixed)
- limit: Maximum items to return (default: 20)
- timeframe: Timeframe for trends (day, week, month, all)
"""
try:
user_id = get_current_user_id()
content_type = request.args.get('type', 'mixed')
limit = min(request.args.get('limit', 20, type=int), 100)
timeframe = request.args.get('timeframe', 'week')
# Validate inputs
valid_types = ['tracks', 'artists', 'albums', 'mixed']
if content_type not in valid_types:
return error_response(f"Invalid type. Must be one of: {valid_types}", 400)
valid_timeframes = ['day', 'week', 'month', 'all']
if timeframe not in valid_timeframes:
return error_response(f"Invalid timeframe. Must be one of: {valid_timeframes}", 400)
# Get trending content (this would integrate with analytics)
# For now, return discovery recommendations as trending
trending = await advanced_ux_service.get_discovery_recommendations(user_id, content_type, limit)
# Format response
formatted_trending = []
for item in trending:
formatted_item = {
'id': item.id,
'type': item.type.value,
'title': item.title,
'subtitle': item.subtitle,
'image_url': item.image_url,
'url': item.url,
'metadata': item.metadata,
'relevance_score': item.relevance_score,
'trend_score': item.relevance_score # Would calculate actual trend score
}
formatted_trending.append(formatted_item)
return success_response({
'trending': formatted_trending,
'type': content_type,
'timeframe': timeframe,
'total_count': len(formatted_trending)
})
except Exception as e:
logger.error(f"Error getting trending content: {e}")
return error_response("Internal server error", 500)
@advanced_ux_bp.route('/search/advanced', methods=['POST'])
@login_required
async def advanced_search():
"""
Perform advanced search with filters and personalization
Request Body:
{
"query": "search query",
"filters": {
"genre": ["rock", "pop"],
"mood": "energetic",
"year": ["2020", "2021"],
"quality": "high",
"duration": "medium"
},
"sort_by": "relevance|popularity|date",
"sort_order": "asc|desc",
"limit": 20,
"offset": 0
}
"""
try:
user_id = get_current_user_id()
data = request.get_json()
if not data:
return error_response("Request body is required", 400)
query = data.get('query', '').strip()
filters = data.get('filters', {})
sort_by = data.get('sort_by', 'relevance')
sort_order = data.get('sort_order', 'desc')
limit = min(data.get('limit', 20, type=int), 100)
offset = max(data.get('offset', 0, type=int), 0)
# Validate inputs
validate_search_query(query)
valid_sort_by = ['relevance', 'popularity', 'date', 'title', 'artist']
if sort_by not in valid_sort_by:
return error_response(f"Invalid sort_by. Must be one of: {valid_sort_by}", 400)
valid_sort_order = ['asc', 'desc']
if sort_order not in valid_sort_order:
return error_response(f"Invalid sort_order. Must be one of: {valid_sort_order}", 400)
# Perform advanced search
# This would implement complex search logic with filters
# For now, use basic search suggestions as placeholder
context = SearchContext.GENERAL
if filters.get('quality') == 'lossless' or 'download' in query.lower():
context = SearchContext.DOWNLOAD
suggestions = await advanced_ux_service.get_search_suggestions(user_id, query, context, limit + offset)
# Apply filters (simplified)
filtered_suggestions = []
for suggestion in suggestions:
include = True
# Genre filter
if 'genre' in filters and filters['genre']:
if not any(genre.lower() in (suggestion.subtitle or '').lower() for genre in filters['genre']):
include = False
# Quality filter
if 'quality' in filters and filters['quality']:
if filters['quality'] not in (suggestion.subtitle or '').lower():
include = False
if include:
filtered_suggestions.append(suggestion)
# Sort results
if sort_by == 'relevance':
filtered_suggestions.sort(key=lambda x: x.relevance_score, reverse=(sort_order == 'desc'))
elif sort_by == 'title':
filtered_suggestions.sort(key=lambda x: x.title.lower(), reverse=(sort_order == 'desc'))
elif sort_by == 'artist':
filtered_suggestions.sort(key=lambda x: (x.subtitle or '').lower(), reverse=(sort_order == 'desc'))
# Apply pagination
paginated_suggestions = filtered_suggestions[offset:offset + limit]
# Format response
formatted_results = []
for suggestion in paginated_suggestions:
formatted_result = {
'id': suggestion.id,
'type': suggestion.type.value,
'title': suggestion.title,
'subtitle': suggestion.subtitle,
'image_url': suggestion.image_url,
'url': suggestion.url,
'metadata': suggestion.metadata,
'relevance_score': suggestion.relevance_score
}
formatted_results.append(formatted_result)
return success_response({
'results': formatted_results,
'query': query,
'filters': filters,
'sort_by': sort_by,
'sort_order': sort_order,
'total_count': len(filtered_suggestions),
'limit': limit,
'offset': offset
})
except Exception as e:
logger.error(f"Error performing advanced search: {e}")
return error_response("Internal server error", 500)
@advanced_ux_bp.route('/suggestions/quick', methods=['GET'])
@login_required
async def get_quick_suggestions():
"""
Get quick suggestions for UI components (autocomplete, etc.)
Query Parameters:
- type: Suggestion type (search, discovery, download)
- limit: Maximum suggestions (default: 5)
"""
try:
user_id = get_current_user_id()
suggestion_type = request.args.get('type', 'search')
limit = min(request.args.get('limit', 5, type=int), 20)
# Validate suggestion type
valid_types = ['search', 'discovery', 'download']
if suggestion_type not in valid_types:
return error_response(f"Invalid type. Must be one of: {valid_types}", 400)
suggestions = []
if suggestion_type == 'search':
# Get default search suggestions
suggestions = await advanced_ux_service._get_default_suggestions(user_id, SearchContext.GENERAL, limit)
elif suggestion_type == 'discovery':
# Get discovery recommendations
suggestions = await advanced_ux_service.get_discovery_recommendations(user_id, 'mixed', limit)
elif suggestion_type == 'download':
# Get download suggestions
suggestions = await advanced_ux_service.get_download_suggestions(user_id, '', limit)
# Format response for quick UI
formatted_suggestions = []
for suggestion in suggestions:
formatted_suggestion = {
'id': suggestion.id,
'type': suggestion.type.value,
'title': suggestion.title,
'subtitle': suggestion.subtitle,
'image_url': suggestion.image_url,
'url': suggestion.url
}
formatted_suggestions.append(formatted_suggestion)
return success_response({
'suggestions': formatted_suggestions,
'type': suggestion_type,
'total_count': len(formatted_suggestions)
})
except Exception as e:
logger.error(f"Error getting quick suggestions: {e}")
return error_response("Internal server error", 500)
@advanced_ux_bp.route('/personalization/preferences', methods=['GET', 'PUT'])
@login_required
async def personalization_preferences():
"""
Get or update personalization preferences
GET: Returns current preferences
PUT: Updates preferences
"""
try:
user_id = get_current_user_id()
if request.method == 'GET':
# Get user behavior profile
behavior = await advanced_ux_service._get_user_behavior(user_id)
preferences = {
'favorite_genres': behavior.favorite_genres,
'favorite_artists': behavior.favorite_artists,
'download_preferences': behavior.download_preferences,
'interaction_patterns': behavior.interaction_patterns
}
return success_response({
'preferences': preferences
})
elif request.method == 'PUT':
# Update preferences
data = request.get_json()
if not data:
return error_response("Request body is required", 400)
# Update user behavior with preferences
interaction_data = {
'type': 'preferences_update',
'data': data
}
await advanced_ux_service.update_user_behavior(user_id, interaction_data)
return success_response({
'message': 'Preferences updated successfully'
})
except Exception as e:
logger.error(f"Error handling personalization preferences: {e}")
return error_response("Internal server error", 500)
+805
View File
@@ -0,0 +1,805 @@
"""
Audio Quality Management API Endpoints
This module provides REST API endpoints for the advanced audio quality control system,
including adaptive streaming, audio enhancement, quality analysis, and user preferences.
"""
import logging
from typing import Dict, List, Optional, Any
from flask import Blueprint, request, jsonify, send_file
from flask_login import login_required, current_user
from swingmusic.db import db
from swingmusic.services.audio_quality_manager import (
audio_quality_manager, AudioQualitySettings, AudioFormat, QualityLevel,
SampleRate, BitDepth, SpatialAudioFormat
)
from swingmusic.utils.request import APIError, success_response, error_response
from swingmusic.utils.validators import validate_audio_file
logger = logging.getLogger(__name__)
audio_quality_bp = Blueprint('audio_quality', __name__, url_prefix='/api/audio-quality')
def get_current_user_id() -> int:
"""Get current user ID from Flask-Login"""
return current_user.id if current_user.is_authenticated else None
@audio_quality_bp.route('/settings', methods=['GET'])
@login_required
async def get_quality_settings():
"""
Get user's audio quality settings
"""
try:
settings = await audio_quality_manager._get_user_settings(get_current_user_id())
return success_response({
'settings': {
'streaming_quality': settings.streaming_quality.value,
'adaptive_quality': settings.adaptive_quality,
'network_aware_quality': settings.network_aware_quality,
'device_specific_quality': settings.device_specific_quality,
'download_format': settings.download_format.value,
'download_bitrate': settings.download_bitrate,
'download_sample_rate': settings.download_sample_rate.value,
'download_bit_depth': settings.download_bit_depth.value,
'enable_dolby_atmos': settings.enable_dolby_atmos,
'enable_360_audio': settings.enable_360_audio,
'spatial_audio_format': settings.spatial_audio_format.value,
'enable_adaptive_eq': settings.enable_adaptive_eq,
'enable_spatial_audio_processing': settings.enable_spatial_audio_processing,
'enable_loudness_normalization': settings.enable_loudness_normalization,
'target_loudness': settings.target_loudness,
'enable_crossfade': settings.enable_crossfade,
'crossfade_duration': settings.crossfade_duration,
'enable_gapless_playback': settings.enable_gapless_playback,
'enable_replaygain': settings.enable_replaygain,
'prioritize_fidelity': settings.prioritize_fidelity,
'prioritize_file_size': settings.prioritize_file_size,
'prioritize_compatibility': settings.prioritize_compatibility,
'custom_ffmpeg_params': settings.custom_ffmpeg_params or {},
'enable_experimental_codecs': settings.enable_experimental_codecs,
'cache_transcoded_files': settings.cache_transcoded_files
}
})
except Exception as e:
logger.error(f"Error getting quality settings: {e}")
return error_response("Internal server error", 500)
@audio_quality_bp.route('/settings', methods=['POST'])
@login_required
async def update_quality_settings():
"""
Update user's audio quality settings
Request Body:
{
"streaming_quality": "lossless|high|medium|low|data_saver",
"adaptive_quality": true,
"network_aware_quality": true,
"device_specific_quality": true,
"download_format": "flac|mp3_320|mp3_256|aac_256|...",
"download_bitrate": 320,
"download_sample_rate": "44.1kHz|48kHz|96kHz|192kHz",
"download_bit_depth": "16bit|24bit|32bit",
"enable_dolby_atmos": false,
"enable_360_audio": false,
"spatial_audio_format": "stereo|binaural|dolby_atmos|...",
"enable_adaptive_eq": true,
"enable_spatial_audio_processing": false,
"enable_loudness_normalization": true,
"target_loudness": -14.0,
"enable_crossfade": false,
"crossfade_duration": 2.0,
"enable_gapless_playback": true,
"enable_replaygain": true,
"prioritize_fidelity": true,
"prioritize_file_size": false,
"prioritize_compatibility": false,
"custom_ffmpeg_params": {},
"enable_experimental_codecs": false,
"cache_transcoded_files": true
}
"""
try:
data = request.get_json()
if not data:
return error_response("Request body is required", 400)
# Validate and convert settings
settings = AudioQualitySettings()
# Streaming quality
if 'streaming_quality' in data:
try:
settings.streaming_quality = QualityLevel(data['streaming_quality'])
except ValueError:
return error_response("Invalid streaming quality", 400)
# Boolean settings
for key in ['adaptive_quality', 'network_aware_quality', 'device_specific_quality',
'enable_dolby_atmos', 'enable_360_audio', 'enable_adaptive_eq',
'enable_spatial_audio_processing', 'enable_loudness_normalization',
'enable_crossfade', 'enable_gapless_playback', 'enable_replaygain',
'prioritize_fidelity', 'prioritize_file_size', 'prioritize_compatibility',
'enable_experimental_codecs', 'cache_transcoded_files']:
if key in data:
setattr(settings, key, bool(data[key]))
# Download format
if 'download_format' in data:
try:
settings.download_format = AudioFormat(data['download_format'])
except ValueError:
return error_response("Invalid download format", 400)
# Numeric settings
if 'download_bitrate' in data:
bitrate = data['download_bitrate']
if bitrate is not None and (not isinstance(bitrate, int) or bitrate < 0 or bitrate > 1000):
return error_response("Invalid download bitrate", 400)
settings.download_bitrate = bitrate
if 'target_loudness' in data:
loudness = data['target_loudness']
if not isinstance(loudness, (int, float)) or loudness < -70 or loudness > 0:
return error_response("Invalid target loudness", 400)
settings.target_loudness = float(loudness)
if 'crossfade_duration' in data:
duration = data['crossfade_duration']
if not isinstance(duration, (int, float)) or duration < 0 or duration > 10:
return error_response("Invalid crossfade duration", 400)
settings.crossfade_duration = float(duration)
# Enum settings
if 'download_sample_rate' in data:
try:
settings.download_sample_rate = SampleRate(data['download_sample_rate'])
except ValueError:
return error_response("Invalid download sample rate", 400)
if 'download_bit_depth' in data:
try:
settings.download_bit_depth = BitDepth(data['download_bit_depth'])
except ValueError:
return error_response("Invalid download bit depth", 400)
if 'spatial_audio_format' in data:
try:
settings.spatial_audio_format = SpatialAudioFormat(data['spatial_audio_format'])
except ValueError:
return error_response("Invalid spatial audio format", 400)
# Custom FFmpeg params
if 'custom_ffmpeg_params' in data:
if not isinstance(data['custom_ffmpeg_params'], dict):
return error_response("Custom FFmpeg params must be an object", 400)
settings.custom_ffmpeg_params = data['custom_ffmpeg_params']
# Update settings
success = await audio_quality_manager.update_user_settings(get_current_user_id(), settings)
if success:
return success_response({
'message': 'Audio quality settings updated successfully',
'settings': data
})
else:
return error_response("Failed to update settings", 500)
except Exception as e:
logger.error(f"Error updating quality settings: {e}")
return error_response("Internal server error", 500)
@audio_quality_bp.route('/optimal-streaming', methods=['GET'])
@login_required
async def get_optimal_streaming_quality():
"""
Get optimal streaming quality based on current conditions
Query Parameters:
- context: JSON string with additional context (battery, network, etc.)
"""
try:
context_str = request.args.get('context', '{}')
try:
context = json.loads(context_str) if context_str else {}
except json.JSONDecodeError:
context = {}
optimal = await audio_quality_manager.get_optimal_streaming_quality(
get_current_user_id(), context
)
return success_response({
'optimal_quality': optimal,
'context': context
})
except Exception as e:
logger.error(f"Error getting optimal streaming quality: {e}")
return error_response("Internal server error", 500)
@audio_quality_bp.route('/transcode', methods=['POST'])
@login_required
async def transcode_for_streaming():
"""
Transcode audio file for optimal streaming
Request Body:
{
"file_path": "/path/to/audio/file",
"context": {}
}
"""
try:
data = request.get_json()
if not data or not data.get('file_path'):
return error_response("file_path is required", 400)
file_path = data['file_path']
context = data.get('context', {})
# Validate file
if not validate_audio_file(file_path):
return error_response("Invalid audio file", 400)
# Transcode for streaming
transcoded_path = await audio_quality_manager.transcode_for_streaming(
file_path, get_current_user_id(), context
)
if transcoded_path:
return success_response({
'transcoded_path': transcoded_path,
'original_path': file_path
})
else:
return error_response("Transcoding failed", 500)
except Exception as e:
logger.error(f"Error transcoding for streaming: {e}")
return error_response("Internal server error", 500)
@audio_quality_bp.route('/analyze', methods=['POST'])
@login_required
async def analyze_audio_file():
"""
Analyze audio file for quality metrics
Request Body:
{
"file_path": "/path/to/audio/file"
}
"""
try:
data = request.get_json()
if not data or not data.get('file_path'):
return error_response("file_path is required", 400)
file_path = data['file_path']
# Validate file
if not validate_audio_file(file_path):
return error_response("Invalid audio file", 400)
# Analyze file
analysis = await audio_quality_manager.analyze_audio_file(file_path)
return success_response({
'analysis': {
'file_path': analysis.file_path,
'format': analysis.format,
'duration': analysis.duration,
'sample_rate': analysis.sample_rate,
'bit_depth': analysis.bit_depth,
'bitrate': analysis.bitrate,
'channels': analysis.channels,
'codec': analysis.codec,
'dynamic_range': analysis.dynamic_range,
'peak_level': analysis.peak_level,
'rms_level': analysis.rms_level,
'loudness': analysis.loudness,
'frequency_response': analysis.frequency_response,
'spectral_centroid': analysis.spectral_centroid,
'spectral_rolloff': analysis.spectral_rolloff,
'signal_to_noise_ratio': analysis.signal_to_noise_ratio,
'total_harmonic_distortion': analysis.total_harmonic_distortion,
'detected_genre': analysis.detected_genre,
'acoustic_features': analysis.acoustic_features or {}
}
})
except Exception as e:
logger.error(f"Error analyzing audio file: {e}")
return error_response("Internal server error", 500)
@audio_quality_bp.route('/compare', methods=['POST'])
@login_required
async def compare_quality_formats():
"""
Compare quality across different audio formats
Request Body:
{
"file_path": "/path/to/audio/file",
"formats": ["flac", "mp3_320", "mp3_256", "aac_256"]
}
"""
try:
data = request.get_json()
if not data or not data.get('file_path'):
return error_response("file_path is required", 400)
file_path = data['file_path']
formats = data.get('formats', ['flac', 'mp3_320'])
# Validate file
if not validate_audio_file(file_path):
return error_response("Invalid audio file", 400)
# Convert format strings to enum
format_enums = []
for format_str in formats:
try:
format_enums.append(AudioFormat(format_str))
except ValueError:
return error_response(f"Invalid format: {format_str}", 400)
# Compare formats
comparison = await audio_quality_manager.compare_quality_formats(
file_path, format_enums
)
return success_response({
'comparison': {
'original_file': comparison.original_file,
'formats': comparison.formats,
'size_difference': comparison.size_difference,
'quality_score': comparison.quality_score,
'transparency_score': comparison.transparency_score,
'recommended_format': comparison.recommended_format,
'recommended_reason': comparison.recommended_reason
}
})
except Exception as e:
logger.error(f"Error comparing quality formats: {e}")
return error_response("Internal server error", 500)
@audio_quality_bp.route('/enhance', methods=['POST'])
@login_required
async def enhance_audio():
"""
Apply audio enhancements to a file
Request Body:
{
"input_path": "/path/to/input/file",
"output_path": "/path/to/output/file",
"enhancements": {
"enable_loudness_normalization": true,
"target_loudness": -14.0,
"enable_adaptive_eq": true,
"enable_spatial_audio_processing": false,
"spatial_audio_format": "stereo"
}
}
"""
try:
data = request.get_json()
if not data or not data.get('input_path') or not data.get('output_path'):
return error_response("input_path and output_path are required", 400)
input_path = data['input_path']
output_path = data['output_path']
enhancements = data.get('enhancements', {})
# Validate files
if not validate_audio_file(input_path):
return error_response("Invalid input audio file", 400)
# Build settings
settings = AudioQualitySettings()
# Apply enhancement settings
for key, value in enhancements.items():
if hasattr(settings, key):
setattr(settings, key, value)
# Apply enhancements
success = await audio_quality_manager.enhancement_service.apply_enhancements(
input_path, output_path, settings
)
if success:
return success_response({
'message': 'Audio enhancements applied successfully',
'input_path': input_path,
'output_path': output_path,
'enhancements': enhancements
})
else:
return error_response("Audio enhancement failed", 500)
except Exception as e:
logger.error(f"Error enhancing audio: {e}")
return error_response("Internal server error", 500)
@audio_quality_bp.route('/formats', methods=['GET'])
@login_required
async def get_supported_formats():
"""
Get list of supported audio formats and their capabilities
"""
try:
formats = {
'lossless': {
'flac': {
'name': 'FLAC',
'description': 'Free Lossless Audio Codec',
'extension': '.flac',
'max_bitrate': None,
'sample_rates': ['44.1kHz', '48kHz', '96kHz', '192kHz'],
'bit_depths': ['16bit', '24bit'],
'channels': ['mono', 'stereo', '5.1', '7.1'],
'compression': 'lossless',
'compatibility': 'high'
},
'alac': {
'name': 'ALAC',
'description': 'Apple Lossless Audio Codec',
'extension': '.m4a',
'max_bitrate': None,
'sample_rates': ['44.1kHz', '48kHz', '96kHz'],
'bit_depths': ['16bit', '24bit'],
'channels': ['mono', 'stereo', '5.1'],
'compression': 'lossless',
'compatibility': 'medium' # Apple ecosystem
},
'wav': {
'name': 'WAV',
'description': 'Waveform Audio File Format',
'extension': '.wav',
'max_bitrate': None,
'sample_rates': ['44.1kHz', '48kHz', '96kHz', '192kHz'],
'bit_depths': ['16bit', '24bit', '32bit'],
'channels': ['mono', 'stereo', '5.1', '7.1'],
'compression': 'none',
'compatibility': 'high'
}
},
'lossy': {
'mp3_320': {
'name': 'MP3 320kbps',
'description': 'MPEG Audio Layer 3 at 320kbps',
'extension': '.mp3',
'max_bitrate': 320,
'sample_rates': ['44.1kHz', '48kHz'],
'bit_depths': ['16bit'],
'channels': ['stereo'],
'compression': 'lossy',
'compatibility': 'very_high'
},
'mp3_256': {
'name': 'MP3 256kbps',
'description': 'MPEG Audio Layer 3 at 256kbps',
'extension': '.mp3',
'max_bitrate': 256,
'sample_rates': ['44.1kHz', '48kHz'],
'bit_depths': ['16bit'],
'channels': ['stereo'],
'compression': 'lossy',
'compatibility': 'very_high'
},
'mp3_192': {
'name': 'MP3 192kbps',
'description': 'MPEG Audio Layer 3 at 192kbps',
'extension': '.mp3',
'max_bitrate': 192,
'sample_rates': ['44.1kHz', '48kHz'],
'bit_depths': ['16bit'],
'channels': ['stereo'],
'compression': 'lossy',
'compatibility': 'very_high'
},
'mp3_128': {
'name': 'MP3 128kbps',
'description': 'MPEG Audio Layer 3 at 128kbps',
'extension': '.mp3',
'max_bitrate': 128,
'sample_rates': ['44.1kHz', '48kHz'],
'bit_depths': ['16bit'],
'channels': ['stereo'],
'compression': 'lossy',
'compatibility': 'very_high'
},
'aac_256': {
'name': 'AAC 256kbps',
'description': 'Advanced Audio Coding at 256kbps',
'extension': '.m4a',
'max_bitrate': 256,
'sample_rates': ['44.1kHz', '48kHz'],
'bit_depths': ['16bit'],
'channels': ['stereo'],
'compression': 'lossy',
'compatibility': 'high'
},
'aac_192': {
'name': 'AAC 192kbps',
'description': 'Advanced Audio Coding at 192kbps',
'extension': '.m4a',
'max_bitrate': 192,
'sample_rates': ['44.1kHz', '48kHz'],
'bit_depths': ['16bit'],
'channels': ['stereo'],
'compression': 'lossy',
'compatibility': 'high'
},
'aac_128': {
'name': 'AAC 128kbps',
'description': 'Advanced Audio Coding at 128kbps',
'extension': '.m4a',
'max_bitrate': 128,
'sample_rates': ['44.1kHz', '48kHz'],
'bit_depths': ['16bit'],
'channels': ['stereo'],
'compression': 'lossy',
'compatibility': 'high'
},
'ogg_vorbis': {
'name': 'Ogg Vorbis',
'description': 'Ogg Vorbis compressed audio',
'extension': '.ogg',
'max_bitrate': 500,
'sample_rates': ['44.1kHz', '48kHz', '96kHz'],
'bit_depths': ['16bit', '24bit'],
'channels': ['mono', 'stereo', '5.1'],
'compression': 'lossy',
'compatibility': 'medium'
},
'ogg_opus': {
'name': 'Opus',
'description': 'Opus audio codec',
'extension': '.opus',
'max_bitrate': 510,
'sample_rates': ['48kHz'],
'bit_depths': ['16bit'],
'channels': ['mono', 'stereo'],
'compression': 'lossy',
'compatibility': 'medium'
}
}
}
return success_response({'formats': formats})
except Exception as e:
logger.error(f"Error getting supported formats: {e}")
return error_response("Internal server error", 500)
@audio_quality_bp.route('/quality-presets', methods=['GET'])
@login_required
async def get_quality_presets():
"""
Get predefined quality presets for different use cases
"""
try:
presets = {
'audiophile': {
'name': 'Audiophile',
'description': 'Maximum quality for critical listening',
'settings': {
'streaming_quality': 'lossless',
'download_format': 'flac',
'download_sample_rate': '96kHz',
'download_bit_depth': '24bit',
'enable_loudness_normalization': false,
'prioritize_fidelity': true
}
},
'portable': {
'name': 'Portable',
'description': 'Balanced quality for mobile devices',
'settings': {
'streaming_quality': 'high',
'download_format': 'aac_256',
'adaptive_quality': true,
'network_aware_quality': true,
'device_specific_quality': true,
'enable_loudness_normalization': true,
'prioritize_compatibility': true
}
},
'data_saver': {
'name': 'Data Saver',
'description': 'Minimal bandwidth usage',
'settings': {
'streaming_quality': 'data_saver',
'download_format': 'mp3_128',
'adaptive_quality': true,
'network_aware_quality': true,
'enable_loudness_normalization': true,
'prioritize_file_size': true
}
},
'studio': {
'name': 'Studio',
'description': 'Professional quality for production',
'settings': {
'streaming_quality': 'lossless',
'download_format': 'wav',
'download_sample_rate': '192kHz',
'download_bit_depth': '32bit',
'enable_loudness_normalization': false,
'prioritize_fidelity': true,
'cache_transcoded_files': false
}
},
'gaming': {
'name': 'Gaming',
'description': 'Low latency with good quality',
'settings': {
'streaming_quality': 'medium',
'download_format': 'mp3_256',
'enable_crossfade': false,
'enable_gapless_playback': true,
'cache_transcoded_files': true
}
},
'podcast': {
'name': 'Podcast',
'description': 'Optimized for speech content',
'settings': {
'streaming_quality': 'medium',
'download_format': 'aac_128',
'enable_loudness_normalization': true,
'target_loudness': -16.0,
'enable_adaptive_eq': true,
'prioritize_file_size': true
}
}
}
return success_response({'presets': presets})
except Exception as e:
logger.error(f"Error getting quality presets: {e}")
return error_response("Internal server error", 500)
@audio_quality_bp.route('/apply-preset', methods=['POST'])
@login_required
async def apply_quality_preset():
"""
Apply a quality preset to user settings
Request Body:
{
"preset_name": "audiophile|portable|data_saver|studio|gaming|podcast"
}
"""
try:
data = request.get_json()
if not data or not data.get('preset_name'):
return error_response("preset_name is required", 400)
preset_name = data['preset_name']
# Get presets
presets_response = await get_quality_presets()
presets = presets_response[1].get_json()['presets']
if preset_name not in presets:
return error_response(f"Unknown preset: {preset_name}", 400)
preset = presets[preset_name]
# Apply preset settings
success = await audio_quality_manager.update_user_settings(
get_current_user_id(),
AudioQualitySettings(**preset['settings'])
)
if success:
return success_response({
'message': f'Applied {preset["name"]} preset successfully',
'preset': preset,
'settings': preset['settings']
})
else:
return error_response("Failed to apply preset", 500)
except Exception as e:
logger.error(f"Error applying quality preset: {e}")
return error_response("Internal server error", 500)
@audio_quality_bp.route('/cache/clear', methods=['POST'])
@login_required
async def clear_quality_cache():
"""
Clear audio quality analysis and transcoding cache
"""
try:
audio_quality_manager.clear_cache()
return success_response({
'message': 'Audio quality cache cleared successfully'
})
except Exception as e:
logger.error(f"Error clearing quality cache: {e}")
return error_response("Internal server error", 500)
@audio_quality_bp.route('/network/status', methods=['GET'])
@login_required
async def get_network_status():
"""
Get current network status for quality optimization
"""
try:
from swingmusic.services.audio_quality_manager import NetworkMonitor
network_monitor = NetworkMonitor()
status = await network_monitor.get_network_status()
return success_response({
'network_status': status
})
except Exception as e:
logger.error(f"Error getting network status: {e}")
return error_response("Internal server error", 500)
@audio_quality_bp.route('/device/info', methods=['GET'])
@login_required
async def get_device_info():
"""
Get device information for quality optimization
"""
try:
from swingmusic.services.audio_quality_manager import DeviceDetector
device_detector = DeviceDetector()
device_info = device_detector.get_device_info()
return success_response({
'device_info': device_info
})
except Exception as e:
logger.error(f"Error getting device info: {e}")
return error_response("Internal server error", 500)
# Error handlers
@audio_quality_bp.errorhandler(404)
def not_found(error):
return error_response("Endpoint not found", 404)
@audio_quality_bp.errorhandler(500)
def internal_error(error):
return error_response("Internal server error", 500)
+463
View File
@@ -0,0 +1,463 @@
"""
Enhanced Search API for SwingMusic
Integrates global music catalog search with existing local search
"""
from flask import Blueprint, request, jsonify
from typing import Dict, List, Any, Optional
import asyncio
from swingmusic.services.music_catalog import music_catalog_service
from swingmusic.api.search import search as local_search
from swingmusic import logger
from swingmusic.db.spotify import UserCatalogPreferencesTable
# Create blueprint
enhanced_search_bp = Blueprint('enhanced_search', __name__, url_prefix='/api/search')
@enhanced_search_bp.route('/global', methods=['POST'])
def global_search():
"""
Search across global music catalog (Spotify)
Request body:
{
"query": "search query",
"type": "all|tracks|albums|artists|playlists",
"limit": 20,
"user_id": 1
}
"""
try:
data = request.get_json()
if not data or not data.get('query'):
return jsonify({'error': 'Search query is required'}), 400
query = data['query'].strip()
search_type = data.get('type', 'all')
limit = min(data.get('limit', 20), 50) # Cap at 50
user_id = data.get('user_id')
# Get user preferences if available
user_prefs = None
if user_id:
user_prefs = UserCatalogPreferencesTable.get_or_create(user_id)
limit = min(limit, user_prefs.max_search_results)
# Run async search
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result = loop.run_until_complete(
music_catalog_service.search_global_catalog(query, search_type, limit)
)
finally:
loop.close()
# Filter based on user preferences
if user_prefs and not user_prefs.show_explicit:
result.tracks = [track for track in result.tracks if not track.explicit]
result.albums = [album for album in result.albums if not album.explicit]
# Convert to dict for JSON response
response_data = {
'query': result.query,
'total': result.total,
'tracks': [_catalog_item_to_dict(track) for track in result.tracks],
'albums': [_catalog_item_to_dict(album) for album in result.albums],
'artists': [_catalog_item_to_dict(artist) for artist in result.artists],
'playlists': [_catalog_item_to_dict(playlist) for playlist in result.playlists],
'source': 'global_catalog',
'cache_info': {
'from_cache': True, # TODO: Implement cache detection
'expires_at': None
}
}
return jsonify(response_data)
except Exception as e:
logger.error(f"Error in global search: {e}")
return jsonify({'error': 'Search failed'}), 500
@enhanced_search_bp.route('/combined', methods=['POST'])
def combined_search():
"""
Search both local library and global catalog
Request body:
{
"query": "search query",
"include_local": true,
"include_global": true,
"type": "all|tracks|albums|artists",
"limit": 20,
"user_id": 1
}
"""
try:
data = request.get_json()
if not data or not data.get('query'):
return jsonify({'error': 'Search query is required'}), 400
query = data['query'].strip()
include_local = data.get('include_local', True)
include_global = data.get('include_global', True)
search_type = data.get('type', 'all')
limit = min(data.get('limit', 20), 50)
user_id = data.get('user_id')
results = {
'query': query,
'local': {'tracks': [], 'albums': [], 'artists': []},
'global': {'tracks': [], 'albums': [], 'artists': [], 'playlists': []},
'total': 0
}
# Search local library
if include_local:
try:
# Use existing local search
local_results = local_search(query, search_type)
results['local'] = local_results if local_results else {'tracks': [], 'albums': [], 'artists': []}
except Exception as e:
logger.error(f"Error in local search: {e}")
# Search global catalog
if include_global:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
global_results = loop.run_until_complete(
music_catalog_service.search_global_catalog(query, search_type, limit)
)
# Filter based on user preferences
user_prefs = None
if user_id:
user_prefs = UserCatalogPreferencesTable.get_or_create(user_id)
if not user_prefs.show_explicit:
global_results.tracks = [track for track in global_results.tracks if not track.explicit]
global_results.albums = [album for album in global_results.albums if not album.explicit]
results['global'] = {
'tracks': [_catalog_item_to_dict(track) for track in global_results.tracks],
'albums': [_catalog_item_to_dict(album) for album in global_results.albums],
'artists': [_catalog_item_to_dict(artist) for artist in global_results.artists],
'playlists': [_catalog_item_to_dict(playlist) for playlist in global_results.playlists]
}
finally:
loop.close()
# Calculate total
results['total'] = (
len(results['local'].get('tracks', [])) +
len(results['local'].get('albums', [])) +
len(results['local'].get('artists', [])) +
len(results['global'].get('tracks', [])) +
len(results['global'].get('albums', [])) +
len(results['global'].get('artists', [])) +
len(results['global'].get('playlists', []))
)
return jsonify(results)
except Exception as e:
logger.error(f"Error in combined search: {e}")
return jsonify({'error': 'Search failed'}), 500
@enhanced_search_bp.route('/suggestions', methods=['GET'])
def search_suggestions():
"""
Get search suggestions based on query and user preferences
Query parameters:
- q: search query
- type: tracks|albums|artists|all
- limit: number of suggestions (default 10)
- user_id: user ID for preferences
"""
try:
query = request.args.get('q', '').strip()
if not query or len(query) < 2:
return jsonify({'suggestions': []})
search_type = request.args.get('type', 'all')
limit = min(int(request.args.get('limit', 10)), 20)
user_id = request.args.get('user_id')
# Get user preferences
user_prefs = None
if user_id:
user_prefs = UserCatalogPreferencesTable.get_or_create(user_id)
limit = min(limit, user_prefs.max_search_results)
# Search cached items for fast suggestions
item_types = None
if search_type != 'all':
item_types = [search_type]
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
# For suggestions, search both cache and live
suggestions = []
# Search cached items first (fast)
from swingmusic.db.spotify import GlobalCatalogCacheTable
cached_items = GlobalCatalogCacheTable.search_cached(query, item_types, limit)
for item in cached_items:
if user_prefs and not user_prefs.show_explicit and item.explicit:
continue
suggestion = {
'id': item.spotify_id,
'type': item.item_type,
'title': item.title,
'artist': item.artist,
'album': item.album,
'image_url': item.image_url,
'popularity': item.popularity,
'source': 'cache'
}
suggestions.append(suggestion)
# If we need more suggestions, search global catalog
if len(suggestions) < limit:
remaining = limit - len(suggestions)
global_results = loop.run_until_complete(
music_catalog_service.search_global_catalog(query, search_type, remaining)
)
for track in global_results.tracks[:remaining]:
if user_prefs and not user_prefs.show_explicit and track.explicit:
continue
suggestion = {
'id': track.spotify_id,
'type': 'track',
'title': track.title,
'artist': track.artist,
'album': track.album,
'image_url': track.image_url,
'popularity': track.popularity,
'source': 'global'
}
suggestions.append(suggestion)
return jsonify({'suggestions': suggestions[:limit]})
finally:
loop.close()
except Exception as e:
logger.error(f"Error in search suggestions: {e}")
return jsonify({'suggestions': []})
@enhanced_search_bp.route('/artist/<artist_id>', methods=['GET'])
def get_artist_info(artist_id: str):
"""
Get comprehensive artist information including top tracks and albums
Path parameters:
- artist_id: Spotify artist ID
Query parameters:
- user_id: user ID for preferences
"""
try:
user_id = request.args.get('user_id')
# Get user preferences
user_prefs = None
if user_id:
user_prefs = UserCatalogPreferencesTable.get_or_create(user_id)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
artist_info = loop.run_until_complete(
music_catalog_service.get_artist_info(artist_id)
)
if not artist_info:
return jsonify({'error': 'Artist not found'}), 404
# Filter based on user preferences
if user_prefs and not user_prefs.show_explicit:
artist_info.top_tracks = [
track for track in artist_info.top_tracks or [] if not track.explicit
]
artist_info.albums = [
album for album in artist_info.albums or [] if not album.explicit
]
response_data = {
'spotify_id': artist_info.spotify_id,
'name': artist_info.name,
'image_url': artist_info.image_url,
'followers': artist_info.followers,
'popularity': artist_info.popularity,
'genres': artist_info.genres or [],
'top_tracks': [_catalog_item_to_dict(track) for track in (artist_info.top_tracks or [])],
'albums': [_catalog_item_to_dict(album) for album in (artist_info.albums or [])],
'related_artists': artist_info.related_artists or []
}
return jsonify(response_data)
finally:
loop.close()
except Exception as e:
logger.error(f"Error getting artist info: {e}")
return jsonify({'error': 'Failed to get artist info'}), 500
@enhanced_search_bp.route('/album/<album_id>', methods=['GET'])
def get_album_details(album_id: str):
"""
Get detailed album information with tracklist
Path parameters:
- album_id: Spotify album ID
Query parameters:
- user_id: user ID for preferences
"""
try:
user_id = request.args.get('user_id')
# Get user preferences
user_prefs = None
if user_id:
user_prefs = UserCatalogPreferencesTable.get_or_create(user_id)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
album = loop.run_until_complete(
music_catalog_service.get_album_details(album_id)
)
if not album:
return jsonify({'error': 'Album not found'}), 404
# Filter based on user preferences
if user_prefs and not user_prefs.show_explicit and album.explicit:
return jsonify({'error': 'Explicit content filtered'}), 403
response_data = _catalog_item_to_dict(album)
# Add tracklist if available in data
if album.data and 'tracks' in album.data:
response_data['tracks'] = [
_catalog_item_to_dict(track) for track in album.data['tracks']
]
return jsonify(response_data)
finally:
loop.close()
except Exception as e:
logger.error(f"Error getting album details: {e}")
return jsonify({'error': 'Failed to get album details'}), 500
@enhanced_search_bp.route('/preferences/<int:user_id>', methods=['GET', 'POST'])
def user_preferences(user_id: int):
"""Get or update user catalog search preferences"""
try:
if request.method == 'GET':
prefs = UserCatalogPreferencesTable.get_or_create(user_id)
return jsonify({
'user_id': prefs.user_id,
'show_explicit': prefs.show_explicit,
'default_quality': prefs.default_quality,
'auto_download': prefs.auto_download,
'show_suggestions': prefs.show_suggestions,
'preferred_genres': prefs.preferred_genres or [],
'excluded_genres': prefs.excluded_genres or [],
'max_search_results': prefs.max_search_results,
'cache_ttl_preference': prefs.cache_ttl_preference
})
elif request.method == 'POST':
data = request.get_json()
if not data:
return jsonify({'error': 'No data provided'}), 400
# Update only provided fields
update_data = {}
allowed_fields = [
'show_explicit', 'default_quality', 'auto_download',
'show_suggestions', 'preferred_genres', 'excluded_genres',
'max_search_results', 'cache_ttl_preference'
]
for field in allowed_fields:
if field in data:
update_data[field] = data[field]
if update_data:
UserCatalogPreferencesTable.update_preferences(user_id, update_data)
return jsonify({'message': 'Preferences updated successfully'})
except Exception as e:
logger.error(f"Error handling user preferences: {e}")
return jsonify({'error': 'Failed to handle preferences'}), 500
def _catalog_item_to_dict(item) -> Dict[str, Any]:
"""Convert CatalogItem to dictionary for JSON response"""
if hasattr(item, '__dict__'):
# It's a dataclass instance
return {
'spotify_id': item.spotify_id,
'type': item.item_type.value if hasattr(item.item_type, 'value') else str(item.item_type),
'title': item.title,
'artist': item.artist,
'album': item.album,
'duration_ms': item.duration_ms,
'popularity': item.popularity,
'preview_url': item.preview_url,
'image_url': item.image_url,
'release_date': item.release_date,
'explicit': item.explicit,
'data': item.data
}
else:
# It's likely a database model
return {
'spotify_id': getattr(item, 'spotify_id', None),
'type': getattr(item, 'item_type', None),
'title': getattr(item, 'title', None),
'artist': getattr(item, 'artist', None),
'album': getattr(item, 'album', None),
'duration_ms': getattr(item, 'duration_ms', None),
'popularity': getattr(item, 'popularity', None),
'preview_url': getattr(item, 'preview_url', None),
'image_url': getattr(item, 'image_url', None),
'release_date': getattr(item, 'release_date', None),
'explicit': getattr(item, 'explicit', False),
'data': getattr(item, 'data', None)
}
def register_enhanced_search_api(app):
"""Register enhanced search API with Flask app"""
app.register_blueprint(enhanced_search_bp)
logger.info("Enhanced search API registered")
+47
View File
@@ -8,7 +8,9 @@ from swingmusic.lib.lyrics import (
get_lyrics_file,
get_lyrics_from_duplicates,
get_lyrics_from_tags,
Lyrics as Lyrics_class,
)
from swingmusic.plugins.lyrics import Lyrics
bp_tag = Tag(name="Lyrics", description="Get lyrics")
api = APIBlueprint("lyrics", __name__, url_prefix="/lyrics", abp_tags=[bp_tag])
@@ -50,6 +52,51 @@ def send_lyrics(body: SendLyricsBody):
# check lyrics plugins
if not lyrics:
try:
# Get track metadata for plugin search
entry = TrackStore.trackhashmap.get(trackhash, None)
if entry and len(entry.tracks) > 0:
track = entry.tracks[0] # Use first track for metadata
title = getattr(track, 'title', '') or ''
artist = ''
if hasattr(track, 'artists') and track.artists:
artist = track.artists[0].name if hasattr(track.artists[0], 'name') else str(track.artists[0])
album = ''
if hasattr(track, 'album') and track.album:
album = track.album.name if hasattr(track.album, 'name') else str(track.album)
# Only proceed if we have basic metadata
if title and artist:
# Initialize lyrics plugin
lyrics_plugin = Lyrics()
if lyrics_plugin.enabled:
# Search for lyrics using plugin
search_results = lyrics_plugin.search_lyrics_by_title_and_artist(title, artist)
if search_results and len(search_results) > 0:
# Use first result or perfect match
perfect_match = search_results[0]
# Try to find perfect match by comparing title and album
if album:
for result in search_results:
result_title = result.get("title", "").lower()
result_album = result.get("album", "").lower()
if (result_title == title.lower() and
result_album == album.lower()):
perfect_match = result
break
# Download lyrics using track ID
track_id = perfect_match.get("track_id")
if track_id:
lrc_content = lyrics_plugin.download_lyrics(track_id, filepath)
if lrc_content and len(lrc_content.strip()) > 0:
lyrics = Lyrics_class(lrc_content)
except Exception as e:
# Log error but don't break the lyrics fetching process
# In production, you might want to log this error
pass
if not lyrics:
return {"error": "No lyrics found"}
+621
View File
@@ -0,0 +1,621 @@
"""
Mobile Offline Mode API Endpoints
This module provides REST API endpoints for mobile offline functionality,
including device management, sync operations, and offline library access.
"""
import logging
from datetime import datetime
from typing import Dict, List, Optional, Any
from flask import Blueprint, request, jsonify
from flask_login import login_required, current_user
from swingmusic.db import db
from swingmusic.services.mobile_offline_service import mobile_offline_service, OfflineQuality, SyncStatus
from swingmusic.utils.request import APIError, success_response, error_response
from swingmusic.utils.validators import validate_device_info, validate_track_ids
logger = logging.getLogger(__name__)
mobile_offline_bp = Blueprint('mobile_offline', __name__, url_prefix='/api/mobile-offline')
def get_current_user_id() -> int:
"""Get current user ID from Flask-Login"""
return current_user.id if current_user.is_authenticated else None
@mobile_offline_bp.route('/devices/register', methods=['POST'])
@login_required
async def register_device():
"""
Register a new mobile device for offline sync
Request Body:
{
"name": "iPhone 14 Pro",
"type": "ios",
"storage_capacity": 256000000000,
"available_storage": 128000000000,
"preferences": {
"auto_sync": true,
"wifi_only": true,
"quality": "balanced"
}
}
"""
try:
user_id = get_current_user_id()
data = request.get_json()
if not data:
return error_response("Request body is required", 400)
# Validate device information
device_info = validate_device_info(data)
# Register device
device = await mobile_offline_service.register_device(user_id, device_info)
return success_response({
'message': 'Device registered successfully',
'device': {
'device_id': device.device_id,
'name': device.device_name,
'type': device.device_type,
'storage_capacity': device.storage_capacity,
'available_storage': device.available_storage,
'offline_quality': device.offline_quality.value,
'auto_sync_enabled': device.auto_sync_enabled,
'sync_status': device.sync_status.value,
'created_at': device.created_at.isoformat()
}
})
except Exception as e:
logger.error(f"Error registering device: {e}")
return error_response("Internal server error", 500)
@mobile_offline_bp.route('/devices', methods=['GET'])
@login_required
async def get_user_devices():
"""
Get all registered devices for the current user
"""
try:
user_id = get_current_user_id()
# This would get all devices for the user from database
# For now, return empty list as placeholder
devices = []
return success_response({
'devices': devices,
'total_count': len(devices)
})
except Exception as e:
logger.error(f"Error getting user devices: {e}")
return error_response("Internal server error", 500)
@mobile_offline_bp.route('/devices/<device_id>', methods=['GET'])
@login_required
async def get_device_info(device_id: str):
"""
Get specific device information
Path Parameters:
- device_id: Device ID
"""
try:
user_id = get_current_user_id()
device = await mobile_offline_service._get_device(device_id, user_id)
if not device:
return error_response("Device not found", 404)
return success_response({
'device': {
'device_id': device.device_id,
'name': device.device_name,
'type': device.device_type,
'storage_capacity': device.storage_capacity,
'available_storage': device.available_storage,
'last_sync': device.last_sync.isoformat() if device.last_sync else None,
'sync_status': device.sync_status.value,
'offline_quality': device.offline_quality.value,
'auto_sync_enabled': device.auto_sync_enabled,
'sync_preferences': device.sync_preferences,
'created_at': device.created_at.isoformat(),
'updated_at': device.updated_at.isoformat()
}
})
except Exception as e:
logger.error(f"Error getting device info: {e}")
return error_response("Internal server error", 500)
@mobile_offline_bp.route('/devices/<device_id>/settings', methods=['PUT'])
@login_required
async def update_device_settings(device_id: str):
"""
Update device settings
Path Parameters:
- device_id: Device ID
Request Body:
{
"offline_quality": "high_quality",
"auto_sync_enabled": true,
"sync_preferences": {
"wifi_only": true,
"auto_cleanup": true
},
"available_storage": 120000000000
}
"""
try:
user_id = get_current_user_id()
data = request.get_json()
if not data:
return error_response("Request body is required", 400)
# Validate settings
if 'offline_quality' in data:
try:
OfflineQuality(data['offline_quality'])
except ValueError:
return error_response("Invalid offline quality", 400)
# Update settings
success = await mobile_offline_service.update_device_settings(user_id, device_id, data)
if not success:
return error_response("Failed to update device settings", 500)
return success_response({
'message': 'Device settings updated successfully'
})
except Exception as e:
logger.error(f"Error updating device settings: {e}")
return error_response("Internal server error", 500)
@mobile_offline_bp.route('/devices/<device_id>/offline-library', methods=['GET'])
@login_required
async def get_offline_library(device_id: str):
"""
Get offline library for device
Path Parameters:
- device_id: Device ID
Query Parameters:
- include_tracks: Include track details (default: true)
- include_queue: Include sync queue status (default: true)
- include_storage: Include storage usage (default: true)
"""
try:
user_id = get_current_user_id()
# Parse include flags
include_flags = {
'tracks': request.args.get('include_tracks', 'true').lower() == 'true',
'queue': request.args.get('include_queue', 'true').lower() == 'true',
'storage': request.args.get('include_storage', 'true').lower() == 'true'
}
# Get offline library
library_data = await mobile_offline_service.get_offline_library(user_id, device_id)
# Build response based on include flags
response_data = {
'device': library_data['device'],
'last_sync': library_data['last_sync'],
'sync_status': library_data['sync_status']
}
if include_flags['tracks']:
response_data['offline_tracks'] = library_data['offline_tracks']
if include_flags['queue']:
response_data['sync_queue'] = library_data['sync_queue']
if include_flags['storage']:
response_data['storage_usage'] = library_data['storage_usage']
return success_response({
'offline_library': response_data
})
except Exception as e:
logger.error(f"Error getting offline library: {e}")
return error_response("Internal server error", 500)
@mobile_offline_bp.route('/devices/<device_id>/add-tracks', methods=['POST'])
@login_required
async def add_tracks_to_offline(device_id: str):
"""
Add tracks to offline library
Path Parameters:
- device_id: Device ID
Request Body:
{
"track_ids": ["track1", "track2", "track3"],
"quality": "high_quality"
}
"""
try:
user_id = get_current_user_id()
data = request.get_json()
if not data:
return error_response("Request body is required", 400)
track_ids = data.get('track_ids', [])
if not track_ids:
return error_response("track_ids are required", 400)
# Validate track IDs
validate_track_ids(track_ids)
# Parse quality
quality = None
if 'quality' in data:
try:
quality = OfflineQuality(data['quality'])
except ValueError:
return error_response("Invalid quality", 400)
# Add tracks to offline library
queue_items = await mobile_offline_service.add_to_offline_library(
user_id, device_id, track_ids, quality
)
return success_response({
'message': f'Added {len(queue_items)} tracks to offline library',
'queue_items': [
{
'queue_id': item.queue_id,
'track_id': item.track_id,
'priority': item.priority,
'quality': item.quality,
'status': item.status,
'added_at': item.added_at.isoformat()
}
for item in queue_items
]
})
except Exception as e:
logger.error(f"Error adding tracks to offline library: {e}")
return error_response("Internal server error", 500)
@mobile_offline_bp.route('/devices/<device_id>/sync-playlist/<playlist_id>', methods=['POST'])
@login_required
async def sync_playlist_offline(device_id: str, playlist_id: str):
"""
Sync entire playlist for offline playback
Path Parameters:
- device_id: Device ID
- playlist_id: Playlist ID
Request Body:
{
"quality": "balanced"
}
"""
try:
user_id = get_current_user_id()
data = request.get_json() or {}
# Parse quality
quality = None
if 'quality' in data:
try:
quality = OfflineQuality(data['quality'])
except ValueError:
return error_response("Invalid quality", 400)
# Sync playlist
queue_items = await mobile_offline_service.sync_playlist_offline(
user_id, device_id, playlist_id, quality
)
return success_response({
'message': f'Playlist sync started with {len(queue_items)} tracks',
'queue_items': [
{
'queue_id': item.queue_id,
'track_id': item.track_id,
'priority': item.priority,
'quality': item.quality,
'status': item.status,
'added_at': item.added_at.isoformat()
}
for item in queue_items
]
})
except Exception as e:
logger.error(f"Error syncing playlist offline: {e}")
return error_response("Internal server error", 500)
@mobile_offline_bp.route('/devices/<device_id>/remove-tracks', methods=['POST'])
@login_required
async def remove_tracks_from_offline(device_id: str):
"""
Remove tracks from offline library
Path Parameters:
- device_id: Device ID
Request Body:
{
"track_ids": ["track1", "track2", "track3"]
}
"""
try:
user_id = get_current_user_id()
data = request.get_json()
if not data:
return error_response("Request body is required", 400)
track_ids = data.get('track_ids', [])
if not track_ids:
return error_response("track_ids are required", 400)
# Validate track IDs
validate_track_ids(track_ids)
# Remove tracks
success = await mobile_offline_service.remove_from_offline_library(
user_id, device_id, track_ids
)
if not success:
return error_response("Failed to remove tracks from offline library", 500)
return success_response({
'message': f'Removed {len(track_ids)} tracks from offline library'
})
except Exception as e:
logger.error(f"Error removing tracks from offline library: {e}")
return error_response("Internal server error", 500)
@mobile_offline_bp.route('/devices/<device_id>/sync-progress', methods=['GET'])
@login_required
async def get_sync_progress(device_id: str):
"""
Get sync progress for device
Path Parameters:
- device_id: Device ID
"""
try:
user_id = get_current_user_id()
progress_data = await mobile_offline_service.get_sync_progress(user_id, device_id)
return success_response({
'sync_progress': progress_data
})
except Exception as e:
logger.error(f"Error getting sync progress: {e}")
return error_response("Internal server error", 500)
@mobile_offline_bp.route('/devices/<device_id>/force-sync', methods=['POST'])
@login_required
async def force_sync_now(device_id: str):
"""
Force immediate sync for device
Path Parameters:
- device_id: Device ID
"""
try:
user_id = get_current_user_id()
success = await mobile_offline_service.force_sync_now(user_id, device_id)
if not success:
return error_response("Failed to force sync", 500)
return success_response({
'message': 'Sync started successfully'
})
except Exception as e:
logger.error(f"Error forcing sync: {e}")
return error_response("Internal server error", 500)
@mobile_offline_bp.route('/devices/<device_id>/storage-info', methods=['GET'])
@login_required
async def get_storage_info(device_id: str):
"""
Get detailed storage information for device
Path Parameters:
- device_id: Device ID
"""
try:
user_id = get_current_user_id()
# Get device info
device = await mobile_offline_service._get_device(device_id, user_id)
if not device:
return error_response("Device not found", 404)
# Get storage usage
storage_usage = await mobile_offline_service._get_storage_usage(device_id)
# Calculate additional info
usage_percentage = (storage_usage.used_space / storage_usage.total_capacity * 100) if storage_usage.total_capacity > 0 else 0
return success_response({
'storage_info': {
'total_capacity': storage_usage.total_capacity,
'used_space': storage_usage.used_space,
'available_space': storage_usage.available_space,
'usage_percentage': round(usage_percentage, 2),
'offline_tracks_count': storage_usage.offline_tracks_count,
'offline_tracks_size': storage_usage.offline_tracks_size,
'other_data_size': storage_usage.other_data_size,
'quality_breakdown': storage_usage.quality_breakdown,
'needs_cleanup': usage_percentage > 90,
'recommendations': _get_storage_recommendations(usage_percentage, storage_usage)
}
})
except Exception as e:
logger.error(f"Error getting storage info: {e}")
return error_response("Internal server error", 500)
@mobile_offline_bp.route('/devices/<device_id>/cleanup', methods=['POST'])
@login_required
async def cleanup_storage(device_id: str):
"""
Cleanup storage by removing old/unused content
Path Parameters:
- device_id: Device ID
Request Body:
{
"strategy": "least_played|oldest|all",
"free_space_bytes": 1000000000
}
"""
try:
user_id = get_current_user_id()
data = request.get_json()
if not data:
return error_response("Request body is required", 400)
strategy = data.get('strategy', 'least_played')
free_space_bytes = data.get('free_space_bytes', 0)
# Validate strategy
valid_strategies = ['least_played', 'oldest', 'all']
if strategy not in valid_strategies:
return error_response(f"Invalid strategy. Must be one of: {valid_strategies}", 400)
# Perform cleanup
# This would implement the actual cleanup logic
freed_space = await mobile_offline_service._cleanup_old_content(device_id, free_space_bytes)
return success_response({
'message': f'Cleanup completed',
'freed_space': freed_space,
'strategy_used': strategy
})
except Exception as e:
logger.error(f"Error during cleanup: {e}")
return error_response("Internal server error", 500)
@mobile_offline_bp.route('/quality-presets', methods=['GET'])
@login_required
async def get_quality_presets():
"""
Get available quality presets for offline downloads
"""
try:
presets = {
'space_saver': {
'name': 'Space Saver',
'description': 'Low quality, maximum storage efficiency',
'estimated_size_per_track': '3MB',
'recommended_for': 'Limited storage, large libraries',
'formats': ['MP3 128kbps', 'AAC 128kbps']
},
'balanced': {
'name': 'Balanced',
'description': 'Medium quality, good balance',
'estimated_size_per_track': '6MB',
'recommended_for': 'Most users, good quality',
'formats': ['MP3 256kbps', 'AAC 256kbps']
},
'high_quality': {
'name': 'High Quality',
'description': 'High quality, more storage usage',
'estimated_size_per_track': '12MB',
'recommended_for': 'Audiophiles, premium headphones',
'formats': ['MP3 320kbps', 'AAC 320kbps', 'OGG Vorbis']
},
'lossless': {
'name': 'Lossless',
'description': 'Lossless quality, maximum storage usage',
'estimated_size_per_track': '30MB',
'recommended_for': 'Critical listening, unlimited storage',
'formats': ['FLAC', 'ALAC', 'WAV']
}
}
return success_response({
'quality_presets': presets
})
except Exception as e:
logger.error(f"Error getting quality presets: {e}")
return error_response("Internal server error", 500)
def _get_storage_recommendations(usage_percentage: float, storage_usage) -> List[str]:
"""Get storage recommendations based on usage"""
recommendations = []
if usage_percentage > 95:
recommendations.extend([
"Critical: Storage almost full",
"Remove least played tracks immediately",
"Consider upgrading to higher capacity device"
])
elif usage_percentage > 90:
recommendations.extend([
"Storage nearly full",
"Enable auto-cleanup settings",
"Remove old or rarely played tracks"
])
elif usage_percentage > 80:
recommendations.extend([
"Storage getting full",
"Consider using space saver quality",
"Review offline library regularly"
])
elif usage_percentage > 70:
recommendations.extend([
"Moderate storage usage",
"Monitor storage regularly",
"Consider quality adjustments"
])
else:
recommendations.extend([
"Storage usage is healthy",
"Continue current settings",
"Consider adding more content if desired"
])
return recommendations
+467
View File
@@ -0,0 +1,467 @@
"""
Music Catalog API for SwingMusic
Provides Spotify-like browsing of global music catalog with download capabilities
"""
from flask import Blueprint, request, jsonify
from typing import Dict, List, Any, Optional
import asyncio
from swingmusic.services.music_catalog import music_catalog_service
from swingmusic import logger
from swingmusic.db.spotify import UserCatalogPreferencesTable
# Create blueprint
music_catalog_bp = Blueprint('music_catalog', __name__, url_prefix='/api/catalog')
@music_catalog_bp.route('/artist/<artist_id>/top-tracks', methods=['GET'])
def get_artist_top_tracks(artist_id: str):
"""
Get artist's most popular tracks
Query parameters:
- limit: Maximum number of tracks (default: 15, max: 50)
- user_id: User ID for preferences
"""
try:
limit = min(request.args.get('limit', 15, type=int), 50)
user_id = request.args.get('user_id', type=int)
# Get user preferences if available
if user_id:
user_prefs = UserCatalogPreferencesTable.get_or_create(user_id)
limit = min(limit, user_prefs.max_top_tracks)
# Run async operation
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
tracks = loop.run_until_complete(
music_catalog_service.get_artist_top_tracks(artist_id, limit)
)
return jsonify({
'tracks': [track.__dict__ for track in tracks],
'total': len(tracks)
})
except Exception as e:
logger.error(f"Error getting artist top tracks: {e}")
return jsonify({'error': 'Failed to get artist top tracks'}), 500
@music_catalog_bp.route('/artist/<artist_id>/albums', methods=['GET'])
def get_artist_discography(artist_id: str):
"""
Get complete artist discography with albums
Query parameters:
- limit: Maximum number of albums (default: 20, max: 50)
- user_id: User ID for preferences
"""
try:
limit = min(request.args.get('limit', 20, type=int), 50)
user_id = request.args.get('user_id', type=int)
# Get user preferences if available
if user_id:
user_prefs = UserCatalogPreferencesTable.get_or_create(user_id)
limit = min(limit, user_prefs.max_albums_per_artist)
# Run async operation
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
albums = loop.run_until_complete(
music_catalog_service.get_artist_discography(artist_id)
)
# Apply limit
albums = albums[:limit]
return jsonify({
'albums': [album.__dict__ for album in albums],
'total': len(albums)
})
except Exception as e:
logger.error(f"Error getting artist discography: {e}")
return jsonify({'error': 'Failed to get artist discography'}), 500
@music_catalog_bp.route('/artist/<artist_id>', methods=['GET'])
def get_artist_info(artist_id: str):
"""
Get comprehensive artist information including top tracks and albums
Query parameters:
- user_id: User ID for preferences
"""
try:
user_id = request.args.get('user_id', type=int)
# Run async operation
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
artist_info = loop.run_until_complete(
music_catalog_service.get_artist_info(artist_id)
)
if not artist_info:
return jsonify({'error': 'Artist not found'}), 404
return jsonify({
'spotify_id': artist_info.spotify_id,
'name': artist_info.name,
'image_url': artist_info.image_url,
'followers': artist_info.followers,
'popularity': artist_info.popularity,
'genres': artist_info.genres or [],
'top_tracks': [track.__dict__ for track in (artist_info.top_tracks or [])],
'albums': [album.__dict__ for album in (artist_info.albums or [])],
'related_artists': artist_info.related_artists or []
})
except Exception as e:
logger.error(f"Error getting artist info: {e}")
return jsonify({'error': 'Failed to get artist info'}), 500
@music_catalog_bp.route('/album/<album_id>', methods=['GET'])
def get_album_details(album_id: str):
"""
Get full album information with tracklist
Query parameters:
- user_id: User ID for preferences
"""
try:
user_id = request.args.get('user_id', type=int)
# Run async operation
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
album = loop.run_until_complete(
music_catalog_service.get_album_details(album_id)
)
if not album:
return jsonify({'error': 'Album not found'}), 404
return jsonify(album.__dict__)
except Exception as e:
logger.error(f"Error getting album details: {e}")
return jsonify({'error': 'Failed to get album details'}), 500
@music_catalog_bp.route('/search', methods=['POST'])
def search_catalog():
"""
Search across global music catalog
Request body:
{
"query": "search query",
"type": "all|tracks|albums|artists|playlists",
"limit": 20,
"user_id": 1
}
"""
try:
data = request.get_json()
if not data or not data.get('query'):
return jsonify({'error': 'Search query is required'}), 400
query = data['query'].strip()
search_type = data.get('type', 'all')
limit = min(data.get('limit', 20), 50) # Cap at 50
user_id = data.get('user_id')
# Validate search type
valid_types = ['all', 'tracks', 'albums', 'artists', 'playlists']
if search_type not in valid_types:
return jsonify({'error': f'Invalid search type. Must be one of: {valid_types}'}), 400
# Get user preferences if available
if user_id:
user_prefs = UserCatalogPreferencesTable.get_or_create(user_id)
limit = min(limit, user_prefs.max_search_results)
# Run async search
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
result = loop.run_until_complete(
music_catalog_service.search_global_catalog(query, search_type, limit)
)
return jsonify({
'tracks': [track.__dict__ for track in result.tracks],
'albums': [album.__dict__ for album in result.albums],
'artists': [artist.__dict__ for artist in result.artists],
'playlists': [playlist.__dict__ for playlist in result.playlists],
'total': result.total,
'query': result.query
})
except Exception as e:
logger.error(f"Error searching catalog: {e}")
return jsonify({'error': 'Failed to search catalog'}), 500
@music_catalog_bp.route('/trending', methods=['GET'])
def get_trending_content():
"""
Get trending/popular content from global catalog
Query parameters:
- type: "tracks|albums|artists" (default: "tracks")
- limit: Maximum results (default: 20, max: 50)
- user_id: User ID for preferences
"""
try:
content_type = request.args.get('type', 'tracks')
limit = min(request.args.get('limit', 20, type=int), 50)
user_id = request.args.get('user_id', type=int)
# Validate content type
valid_types = ['tracks', 'albums', 'artists']
if content_type not in valid_types:
return jsonify({'error': f'Invalid type. Must be one of: {valid_types}'}), 400
# Get user preferences if available
if user_id:
user_prefs = UserCatalogPreferencesTable.get_or_create(user_id)
limit = min(limit, user_prefs.max_trending_results)
# For now, search for popular content with generic queries
# This could be enhanced with actual trending data from Spotify API
trending_queries = {
'tracks': 'popular hits 2024',
'albums': 'new releases 2024',
'artists': 'popular artists'
}
query = trending_queries.get(content_type, 'popular')
# Run async search
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
result = loop.run_until_complete(
music_catalog_service.search_global_catalog(query, content_type, limit)
)
# Return only the requested type
response = {
'type': content_type,
'total': len(getattr(result, content_type)),
'query': query
}
if content_type == 'tracks':
response['tracks'] = [track.__dict__ for track in result.tracks]
elif content_type == 'albums':
response['albums'] = [album.__dict__ for album in result.albums]
elif content_type == 'artists':
response['artists'] = [artist.__dict__ for artist in result.artists]
return jsonify(response)
except Exception as e:
logger.error(f"Error getting trending content: {e}")
return jsonify({'error': 'Failed to get trending content'}), 500
@music_catalog_bp.route('/recommendations', methods=['POST'])
def get_recommendations():
"""
Get personalized recommendations based on seeds
Request body:
{
"seed_artists": ["artist_id1", "artist_id2"],
"seed_tracks": ["track_id1", "track_id2"],
"seed_genres": ["rock", "pop"],
"limit": 20,
"user_id": 1
}
"""
try:
data = request.get_json()
if not data:
return jsonify({'error': 'Request body is required'}), 400
seed_artists = data.get('seed_artists', [])
seed_tracks = data.get('seed_tracks', [])
seed_genres = data.get('seed_genres', [])
limit = min(data.get('limit', 20), 50)
user_id = data.get('user_id')
# Validate at least one seed type
if not any([seed_artists, seed_tracks, seed_genres]):
return jsonify({'error': 'At least one seed type must be provided'}), 400
# Get user preferences if available
if user_id:
user_prefs = UserCatalogPreferencesTable.get_or_create(user_id)
limit = min(limit, user_prefs.max_recommendations)
# For now, generate recommendations based on seed artists
# This could be enhanced with Spotify's recommendations API
recommendations = []
if seed_artists:
# Get top tracks from seed artists
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
for artist_id in seed_artists[:5]: # Limit to 5 artists
try:
tracks = loop.run_until_complete(
music_catalog_service.get_artist_top_tracks(artist_id, 5)
)
recommendations.extend(tracks)
except Exception as e:
logger.warning(f"Failed to get tracks for artist {artist_id}: {e}")
# Remove duplicates and apply limit
seen_ids = set()
unique_recommendations = []
for track in recommendations:
if track.spotify_id not in seen_ids:
seen_ids.add(track.spotify_id)
unique_recommendations.append(track)
if len(unique_recommendations) >= limit:
break
return jsonify({
'tracks': [track.__dict__ for track in unique_recommendations],
'total': len(unique_recommendations),
'seeds': {
'artists': seed_artists,
'tracks': seed_tracks,
'genres': seed_genres
}
})
except Exception as e:
logger.error(f"Error getting recommendations: {e}")
return jsonify({'error': 'Failed to get recommendations'}), 500
@music_catalog_bp.route('/preferences/<int:user_id>', methods=['GET', 'POST'])
def user_catalog_preferences(user_id: int):
"""
Get or update user's catalog preferences
GET: Returns user preferences
POST: Updates user preferences
POST request body:
{
"max_search_results": 50,
"max_top_tracks": 15,
"max_albums_per_artist": 20,
"max_trending_results": 20,
"max_recommendations": 20,
"show_explicit": true,
"preferred_markets": ["US", "GB", "DE"]
}
"""
try:
if request.method == 'GET':
user_prefs = UserCatalogPreferencesTable.get_or_create(user_id)
return jsonify({
'user_id': user_id,
'max_search_results': user_prefs.max_search_results,
'max_top_tracks': user_prefs.max_top_tracks,
'max_albums_per_artist': user_prefs.max_albums_per_artist,
'max_trending_results': user_prefs.max_trending_results,
'max_recommendations': user_prefs.max_recommendations,
'show_explicit': user_prefs.show_explicit,
'preferred_markets': user_prefs.preferred_markets or []
})
else: # POST
data = request.get_json()
if not data:
return jsonify({'error': 'Request body is required'}), 400
user_prefs = UserCatalogPreferencesTable.get_or_create(user_id)
# Update preferences
if 'max_search_results' in data:
user_prefs.max_search_results = min(data['max_search_results'], 100)
if 'max_top_tracks' in data:
user_prefs.max_top_tracks = min(data['max_top_tracks'], 50)
if 'max_albums_per_artist' in data:
user_prefs.max_albums_per_artist = min(data['max_albums_per_artist'], 100)
if 'max_trending_results' in data:
user_prefs.max_trending_results = min(data['max_trending_results'], 100)
if 'max_recommendations' in data:
user_prefs.max_recommendations = min(data['max_recommendations'], 100)
if 'show_explicit' in data:
user_prefs.show_explicit = bool(data['show_explicit'])
if 'preferred_markets' in data:
user_prefs.preferred_markets = data['preferred_markets']
user_prefs.save()
return jsonify({'message': 'Preferences updated successfully'})
except Exception as e:
logger.error(f"Error handling catalog preferences: {e}")
return jsonify({'error': 'Failed to handle preferences'}), 500
@music_catalog_bp.route('/cache/cleanup', methods=['POST'])
def cleanup_cache():
"""
Clean up expired cache entries
This is typically called by a background job
"""
try:
music_catalog_service.cleanup_expired_cache()
return jsonify({'message': 'Cache cleanup completed'})
except Exception as e:
logger.error(f"Error cleaning up cache: {e}")
return jsonify({'error': 'Failed to cleanup cache'}), 500
@music_catalog_bp.route('/health', methods=['GET'])
def health_check():
"""
Health check endpoint for the music catalog service
"""
try:
# Check if the service is accessible
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# Simple test - try to get a popular artist's top tracks
# Using a well-known artist ID for testing
test_result = loop.run_until_complete(
music_catalog_service.get_artist_top_tracks("4q3ewHC7JlriWjjK2XsvrO", 1) # Daft Punk
)
return jsonify({
'status': 'healthy',
'service': 'music_catalog',
'test_query_success': len(test_result) > 0
})
except Exception as e:
logger.error(f"Health check failed: {e}")
return jsonify({
'status': 'unhealthy',
'service': 'music_catalog',
'error': str(e)
}), 500
+6 -2
View File
@@ -55,6 +55,10 @@ def search_lyrics(body: LyricsSearchBody):
if lrc is not None:
lyrics = Lyrics_class(lrc)
return {"trackhash": trackhash, "lyrics": lyrics.format_synced_lyrics()}, 200
if lyrics.is_synced:
formatted_lyrics = lyrics.format_synced_lyrics()
else:
formatted_lyrics = lyrics.format_unsynced_lyrics()
return {"trackhash": trackhash, "lyrics": formatted_lyrics, "synced": lyrics.is_synced}, 200
return {"trackhash": trackhash, "lyrics": lrc}, 200
return {"trackhash": trackhash, "lyrics": None, "synced": False}, 200
+435
View File
@@ -0,0 +1,435 @@
"""
Year-in-Review API Endpoints
This module provides REST API endpoints for the year-in-review experience,
including recap generation, summary retrieval, and video generation.
"""
import logging
from datetime import datetime
from typing import Dict, List, Optional, Any
from flask import Blueprint, request, jsonify
from flask_login import login_required, current_user
from swingmusic.db import db
from swingmusic.services.recap_service import recap_service, RecapTheme
from swingmusic.utils.request import APIError, success_response, error_response
logger = logging.getLogger(__name__)
recap_bp = Blueprint('recap', __name__, url_prefix='/api/recap')
def get_current_user_id() -> int:
"""Get current user ID from Flask-Login"""
return current_user.id if current_user.is_authenticated else None
@recap_bp.route('/generate/<int:year>', methods=['POST'])
@login_required
async def generate_recap(year: int):
"""
Generate year-in-review for a specific year
Path Parameters:
- year: Year to generate recap for
Query Parameters:
- force: Force regeneration even if recap exists (default: false)
"""
try:
user_id = get_current_user_id()
force_regeneration = request.args.get('force', 'false').lower() == 'true'
# Check if recap already exists
if not force_regeneration:
existing_recap = await recap_service.get_recap_summary(user_id, year)
if existing_recap:
return success_response({
'message': f'Recap for {year} already exists',
'recap': existing_recap
})
# Generate new recap
recap_data = await recap_service.generate_year_recap(user_id, year)
return success_response({
'message': f'Successfully generated recap for {year}',
'recap_id': f"{user_id}_{year}",
'year': recap_data.year,
'stats': {
'total_minutes': recap_data.stats.total_minutes,
'total_tracks': recap_data.stats.total_tracks,
'total_artists': recap_data.stats.total_artists,
'unique_tracks': recap_data.stats.unique_tracks,
'listening_streak': recap_data.stats.listening_streak,
'personality_type': recap_data.personality.personality_type
}
})
except Exception as e:
logger.error(f"Error generating recap: {e}")
return error_response("Internal server error", 500)
@recap_bp.route('/summary/<int:year>', methods=['GET'])
@login_required
async def get_recap_summary(year: int):
"""
Get recap summary for a specific year
Path Parameters:
- year: Year to get recap summary for
"""
try:
user_id = get_current_user_id()
recap_summary = await recap_service.get_recap_summary(user_id, year)
if not recap_summary:
return error_response(f"No recap found for year {year}", 404)
return success_response({
'recap': recap_summary
})
except Exception as e:
logger.error(f"Error getting recap summary: {e}")
return error_response("Internal server error", 500)
@recap_bp.route('/details/<int:year>', methods=['GET'])
@login_required
async def get_recap_details(year: int):
"""
Get detailed recap data for a specific year
Path Parameters:
- year: Year to get recap details for
Query Parameters:
- include_top_tracks: Include top tracks data (default: true)
- include_top_artists: Include top artists data (default: true)
- include_top_albums: Include top albums data (default: true)
- include_discoveries: Include discoveries data (default: true)
- include_milestones: Include milestones data (default: true)
"""
try:
user_id = get_current_user_id()
# Get recap summary first
recap_summary = await recap_service.get_recap_summary(user_id, year)
if not recap_summary:
return error_response(f"No recap found for year {year}", 404)
# Parse include flags
include_flags = {
'top_tracks': request.args.get('include_top_tracks', 'true').lower() == 'true',
'top_artists': request.args.get('include_top_artists', 'true').lower() == 'true',
'top_albums': request.args.get('include_top_albums', 'true').lower() == 'true',
'discoveries': request.args.get('include_discoveries', 'true').lower() == 'true',
'milestones': request.args.get('include_milestones', 'true').lower() == 'true'
}
# Load full recap data from file
import json
from pathlib import Path
recap_file = Path(recap_service.recap_dir) / f"recap_{user_id}_{year}.json"
if not recap_file.exists():
return error_response(f"Recap data not found for year {year}", 404)
with open(recap_file, 'r') as f:
full_recap_data = json.load(f)
# Build response based on include flags
response_data = {
'year': full_recap_data['year'],
'stats': full_recap_data['stats'],
'personality': full_recap_data['personality'],
'monthly_breakdown': full_recap_data['monthly_breakdown'],
'created_at': full_recap_data['created_at']
}
if include_flags['top_tracks']:
response_data['top_tracks'] = full_recap_data['top_tracks']
if include_flags['top_artists']:
response_data['top_artists'] = full_recap_data['top_artists']
if include_flags['top_albums']:
response_data['top_albums'] = full_recap_data['top_albums']
if include_flags['discoveries']:
response_data['discoveries'] = full_recap_data['discoveries']
if include_flags['milestones']:
response_data['milestones'] = full_recap_data['milestones']
return success_response({
'recap': response_data
})
except Exception as e:
logger.error(f"Error getting recap details: {e}")
return error_response("Internal server error", 500)
@recap_bp.route('/video/<int:year>', methods=['POST'])
@login_required
async def generate_recap_video(year: int):
"""
Generate recap video for a specific year
Path Parameters:
- year: Year to generate video for
Request Body:
{
"theme": "modern|retro|minimal|vibrant|dark|light",
"include_audio": true,
"duration_limit": 180 // Optional: max duration in seconds
}
"""
try:
user_id = get_current_user_id()
# Get request data
data = request.get_json() or {}
# Validate theme
theme_name = data.get('theme', 'modern')
try:
theme = RecapTheme(theme_name)
except ValueError:
return error_response(f"Invalid theme: {theme_name}. Must be one of: {[t.value for t in RecapTheme]}", 400)
# Check if recap exists
recap_summary = await recap_service.get_recap_summary(user_id, year)
if not recap_summary:
return error_response(f"No recap found for year {year}. Generate recap first.", 404)
# Generate video (this is a placeholder - would integrate with Remotion service)
video_path = await recap_service.generate_recap_video(
# This would need to load the full recap data
None, # recap_data would be loaded here
theme
)
return success_response({
'message': f'Video generation started for {year}',
'video_path': video_path,
'theme': theme.value,
'estimated_completion': '2-5 minutes'
})
except Exception as e:
logger.error(f"Error generating recap video: {e}")
return error_response("Internal server error", 500)
@recap_bp.route('/available-years', methods=['GET'])
@login_required
async def get_available_years():
"""
Get list of years for which recaps are available
"""
try:
user_id = get_current_user_id()
# Scan recap directory for user's recaps
import os
from pathlib import Path
recap_dir = Path(recap_service.recap_dir)
available_years = []
if recap_dir.exists():
for file_path in recap_dir.glob(f"recap_{user_id}_*.json"):
# Extract year from filename
parts = file_path.stem.split('_')
if len(parts) >= 3:
year = parts[2]
try:
year_int = int(year)
available_years.append(year_int)
except ValueError:
continue
# Sort years in descending order
available_years.sort(reverse=True)
return success_response({
'available_years': available_years,
'total_recaps': len(available_years)
})
except Exception as e:
logger.error(f"Error getting available years: {e}")
return error_response("Internal server error", 500)
@recap_bp.route('/share/<int:year>', methods=['POST'])
@login_required
async def create_shareable_link(year: int):
"""
Create a shareable link for recap
Path Parameters:
- year: Year to create shareable link for
Request Body:
{
"include_personal_data": false,
"expires_in_days": 30
}
"""
try:
user_id = get_current_user_id()
# Get request data
data = request.get_json() or {}
include_personal_data = data.get('include_personal_data', False)
expires_in_days = data.get('expires_in_days', 30)
# Check if recap exists
recap_summary = await recap_service.get_recap_summary(user_id, year)
if not recap_summary:
return error_response(f"No recap found for year {year}", 404)
# Generate shareable link (this is a placeholder implementation)
import secrets
import hashlib
# Generate unique token
token_data = f"{user_id}_{year}_{datetime.utcnow().timestamp()}"
share_token = hashlib.sha256(token_data.encode()).hexdigest()[:16]
# Create shareable data
shareable_data = {
'year': year,
'stats': {
'total_minutes': recap_summary['total_minutes'],
'total_tracks': recap_summary['total_tracks'],
'personality_type': recap_summary['personality_type']
},
'top_track': recap_summary.get('top_track'),
'top_artist': recap_summary.get('top_artist'),
'created_at': recap_summary['created_at']
}
# Save shareable data (in a real implementation, this would go to database)
share_file = Path(recap_service.recap_dir) / f"share_{share_token}.json"
import json
with open(share_file, 'w') as f:
json.dump({
'user_id': user_id,
'year': year,
'data': shareable_data,
'expires_at': (datetime.utcnow() + datetime.timedelta(days=expires_in_days)).isoformat(),
'created_at': datetime.utcnow().isoformat()
}, f)
share_url = f"/recap/shared/{share_token}"
return success_response({
'share_url': share_url,
'share_token': share_token,
'expires_in_days': expires_in_days,
'includes_personal_data': include_personal_data
})
except Exception as e:
logger.error(f"Error creating shareable link: {e}")
return error_response("Internal server error", 500)
@recap_bp.route('/shared/<token>', methods=['GET'])
async def get_shared_recap(token: str):
"""
Get shared recap by token (public endpoint)
Path Parameters:
- token: Share token
"""
try:
# Load share data
share_file = Path(recap_service.recap_dir) / f"share_{token}.json"
if not share_file.exists():
return error_response("Shared recap not found or expired", 404)
import json
with open(share_file, 'r') as f:
share_data = json.load(f)
# Check if expired
expires_at = datetime.fromisoformat(share_data['expires_at'])
if datetime.utcnow() > expires_at:
share_file.unlink() # Clean up expired share
return error_response("Shared recap has expired", 410)
return success_response({
'shared_recap': share_data['data'],
'year': share_data['year'],
'created_at': share_data['created_at']
})
except Exception as e:
logger.error(f"Error getting shared recap: {e}")
return error_response("Internal server error", 500)
@recap_bp.route('/compare/<int:year1>/<int:year2>', methods=['GET'])
@login_required
async def compare_years(year1: int, year2: int):
"""
Compare recaps between two years
Path Parameters:
- year1: First year to compare
- year2: Second year to compare
"""
try:
user_id = get_current_user_id()
# Get both recaps
recap1 = await recap_service.get_recap_summary(user_id, year1)
recap2 = await recap_service.get_recap_summary(user_id, year2)
if not recap1:
return error_response(f"No recap found for year {year1}", 404)
if not recap2:
return error_response(f"No recap found for year {year2}", 404)
# Calculate comparisons
comparison = {
'year1': year1,
'year2': year2,
'listening_time_change': {
'absolute': recap2['total_minutes'] - recap1['total_minutes'],
'percentage': ((recap2['total_minutes'] - recap1['total_minutes']) / recap1['total_minutes'] * 100) if recap1['total_minutes'] > 0 else 0
},
'tracks_change': {
'absolute': recap2['total_tracks'] - recap1['total_tracks'],
'percentage': ((recap2['total_tracks'] - recap1['total_tracks']) / recap1['total_tracks'] * 100) if recap1['total_tracks'] > 0 else 0
},
'personality_change': {
'from': recap1['personality_type'],
'to': recap2['personality_type'],
'changed': recap1['personality_type'] != recap2['personality_type']
}
}
return success_response({
'comparison': comparison,
'recap1': recap1,
'recap2': recap2
})
except Exception as e:
logger.error(f"Error comparing years: {e}")
return error_response("Internal server error", 500)
+426
View File
@@ -0,0 +1,426 @@
"""
Spotify Downloader API endpoints for SwingMusic
Provides REST API for Spotify URL downloading functionality
"""
from flask import Blueprint, request, jsonify
from flask_openapi3 import APIBlueprint, Tag
from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
import asyncio
from swingmusic.services.spotify_downloader import spotify_downloader, DownloadSource
from swingmusic import logger
from swingmusic.utils import create_valid_filename
spotify_bp = APIBlueprint(
'spotify',
__name__,
url_prefix='/api/spotify',
abp_tag=Tag(name='Spotify', description='Spotify downloader operations')
)
class SpotifyURLRequest(BaseModel):
url: str = Field(..., description='Spotify URL (track, album, or playlist)')
quality: Optional[str] = Field('flac', description='Audio quality (flac, mp3_320, mp3_128)')
output_dir: Optional[str] = Field(None, description='Output directory (optional)')
class SpotifyMetadataResponse(BaseModel):
spotify_id: str
title: str
artist: str
album: str
duration_ms: int
image_url: str
release_date: str
track_number: int
total_tracks: int
is_explicit: bool
preview_url: Optional[str]
class DownloadItemResponse(BaseModel):
id: str
spotify_url: str
spotify_id: str
title: str
artist: str
album: str
duration_ms: int
image_url: str
quality: str
source: str
status: str
progress: int
file_path: Optional[str]
error_message: Optional[str]
created_at: float
started_at: Optional[float]
completed_at: Optional[float]
class QueueStatusResponse(BaseModel):
queue_length: int
active_downloads: int
pending_items: int
queue: List[DownloadItemResponse]
active: List[DownloadItemResponse]
history: List[DownloadItemResponse]
class ActionResponse(BaseModel):
success: bool
message: str
item_id: Optional[str] = None
@spotify_bp.post('/metadata', summary='Get Spotify metadata')
async def get_metadata(body: SpotifyURLRequest):
"""
Extract metadata from a Spotify URL without downloading
- **url**: Spotify URL for track, album, or playlist
- **quality**: Preferred audio quality (optional)
Returns metadata for the Spotify content.
"""
try:
metadata = await spotify_downloader.get_metadata(body.url)
if not metadata:
return jsonify({
'error': 'Invalid Spotify URL or failed to fetch metadata',
'success': False
}), 400
return jsonify({
'success': True,
'metadata': {
'spotify_id': metadata.spotify_id,
'title': metadata.title,
'artist': metadata.artist,
'album': metadata.album,
'duration_ms': metadata.duration_ms,
'image_url': metadata.image_url,
'release_date': metadata.release_date,
'track_number': metadata.track_number,
'total_tracks': metadata.total_tracks,
'is_explicit': metadata.is_explicit,
'preview_url': metadata.preview_url
}
})
except Exception as e:
logger.error(f"Error getting Spotify metadata: {e}")
return jsonify({
'error': str(e),
'success': False
}), 500
@spotify_bp.post('/download', summary='Download from Spotify URL')
async def download_from_url(body: SpotifyURLRequest):
"""
Add a Spotify URL to the download queue
- **url**: Spotify URL for track, album, or playlist
- **quality**: Audio quality preference (flac, mp3_320, mp3_128)
- **output_dir**: Custom output directory (optional)
Adds the item to the download queue and returns the download ID.
"""
try:
# Validate quality
valid_qualities = ['flac', 'mp3_320', 'mp3_128']
if body.quality not in valid_qualities:
return jsonify({
'error': f'Invalid quality. Must be one of: {", ".join(valid_qualities)}',
'success': False
}), 400
# Add to download queue
item_id = spotify_downloader.add_download(
spotify_url=body.url,
output_dir=body.output_dir,
quality=body.quality
)
if not item_id:
return jsonify({
'error': 'Failed to add download. Invalid URL or duplicate.',
'success': False
}), 400
return jsonify({
'success': True,
'message': 'Download added to queue',
'item_id': item_id
})
except Exception as e:
logger.error(f"Error adding download: {e}")
return jsonify({
'error': str(e),
'success': False
}), 500
@spotify_bp.get('/queue', summary='Get download queue status')
def get_queue_status():
"""
Get current status of the download queue
Returns information about queued items, active downloads, and history.
"""
try:
status = spotify_downloader.get_queue_status()
return jsonify({
'success': True,
'data': status
})
except Exception as e:
logger.error(f"Error getting queue status: {e}")
return jsonify({
'error': str(e),
'success': False
}), 500
@spotify_bp.post('/cancel/<item_id>', summary='Cancel download')
def cancel_download(item_id: str):
"""
Cancel a pending or active download
- **item_id**: ID of the download item to cancel
Returns success status of the cancellation.
"""
try:
success = spotify_downloader.cancel_download(item_id)
if success:
return jsonify({
'success': True,
'message': 'Download cancelled successfully'
})
else:
return jsonify({
'success': False,
'message': 'Download not found or cannot be cancelled'
}), 404
except Exception as e:
logger.error(f"Error cancelling download: {e}")
return jsonify({
'error': str(e),
'success': False
}), 500
@spotify_bp.post('/retry/<item_id>', summary='Retry failed download')
def retry_download(item_id: str):
"""
Retry a failed download
- **item_id**: ID of the failed download item to retry
Returns success status of the retry operation.
"""
try:
success = spotify_downloader.retry_download(item_id)
if success:
return jsonify({
'success': True,
'message': 'Download added to queue for retry'
})
else:
return jsonify({
'success': False,
'message': 'Download not found or cannot be retried'
}), 404
except Exception as e:
logger.error(f"Error retrying download: {e}")
return jsonify({
'error': str(e),
'success': False
}), 500
@spotify_bp.get('/sources', summary='Get available download sources')
def get_download_sources():
"""
Get list of available download sources and their status
Returns information about available download sources (Tidal, Qobuz, Amazon).
"""
try:
sources = []
for source in DownloadSource:
sources.append({
'name': source.value,
'display_name': source.value.title(),
'enabled': True, # In real implementation, check availability
'priority': list(DownloadSource).index(source)
})
return jsonify({
'success': True,
'sources': sources
})
except Exception as e:
logger.error(f"Error getting download sources: {e}")
return jsonify({
'error': str(e),
'success': False
}), 500
@spotify_bp.get('/qualities', summary='Get available audio qualities')
def get_audio_qualities():
"""
Get list of available audio qualities
Returns supported audio formats and quality options.
"""
try:
qualities = [
{
'id': 'flac',
'name': 'FLAC',
'description': 'Lossless audio quality',
'extension': 'flac',
'bitrate': 'Lossless'
},
{
'id': 'mp3_320',
'name': 'MP3 320kbps',
'description': 'High quality MP3',
'extension': 'mp3',
'bitrate': '320 kbps'
},
{
'id': 'mp3_128',
'name': 'MP3 128kbps',
'description': 'Standard quality MP3',
'extension': 'mp3',
'bitrate': '128 kbps'
}
]
return jsonify({
'success': True,
'qualities': qualities
})
except Exception as e:
logger.error(f"Error getting audio qualities: {e}")
return jsonify({
'error': str(e),
'success': False
}), 500
@spotify_bp.get('/history', summary='Get download history')
def get_download_history():
"""
Get download history
Returns paginated download history.
"""
try:
# Get query parameters
page = int(request.args.get('page', 1))
limit = int(request.args.get('limit', 50))
status_filter = request.args.get('status', None)
# Get history from downloader
status = spotify_downloader.get_queue_status()
history = status.get('history', [])
# Apply status filter
if status_filter:
history = [item for item in history if item.get('status') == status_filter]
# Paginate
total = len(history)
start = (page - 1) * limit
end = start + limit
paginated_history = history[start:end]
return jsonify({
'success': True,
'data': {
'items': paginated_history,
'pagination': {
'page': page,
'limit': limit,
'total': total,
'pages': (total + limit - 1) // limit
}
}
})
except Exception as e:
logger.error(f"Error getting download history: {e}")
return jsonify({
'error': str(e),
'success': False
}), 500
@spotify_bp.delete('/clear-history', summary='Clear download history')
def clear_download_history():
"""
Clear download history
Removes all completed and failed downloads from history.
"""
try:
# Clear history in downloader
spotify_downloader.download_history.clear()
return jsonify({
'success': True,
'message': 'Download history cleared'
})
except Exception as e:
logger.error(f"Error clearing download history: {e}")
return jsonify({
'error': str(e),
'success': False
}), 500
# Error handlers
@spotify_bp.errorhandler(400)
def bad_request(error):
return jsonify({
'error': 'Bad request',
'message': str(error),
'success': False
}), 400
@spotify_bp.errorhandler(404)
def not_found(error):
return jsonify({
'error': 'Not found',
'message': str(error),
'success': False
}), 404
@spotify_bp.errorhandler(500)
def internal_error(error):
return jsonify({
'error': 'Internal server error',
'message': str(error),
'success': False
}), 500
+372
View File
@@ -0,0 +1,372 @@
"""
Spotify Downloader Settings API endpoints
"""
from flask import Blueprint, request, jsonify
from flask_openapi3 import APIBlueprint, Tag
from pydantic import BaseModel, Field
from typing import Optional, Dict, Any
from swingmusic import logger
from swingmusic.config import UserConfig
spotify_settings_bp = APIBlueprint(
'spotify_settings',
__name__,
url_prefix='/api/settings/spotify',
abp_tag=Tag(name='Spotify Settings', description='Spotify downloader settings operations')
)
class SpotifySettingsRequest(BaseModel):
defaultQuality: str = Field('flac', description='Default download quality')
downloadFolder: Optional[str] = Field(None, description='Download folder path')
autoAddToLibrary: bool = Field(True, description='Auto-add downloads to library')
maxConcurrentDownloads: int = Field(3, description='Max concurrent downloads')
sources: Optional[list] = Field(None, description='Download sources configuration')
maxRetryAttempts: int = Field(3, description='Max retry attempts')
cleanupHistoryDays: int = Field(30, description='Auto-cleanup history days')
showExplicitWarning: bool = Field(True, description='Show explicit content warning')
class SpotifySettingsResponse(BaseModel):
success: bool
settings: Optional[Dict[str, Any]] = None
message: Optional[str] = None
# Default settings
DEFAULT_SETTINGS = {
'defaultQuality': 'flac',
'downloadFolder': '',
'autoAddToLibrary': True,
'maxConcurrentDownloads': 3,
'sources': [
{
'name': 'tidal',
'display_name': 'Tidal',
'enabled': True,
'priority': 1,
'config': {
'quality_preference': ['lossless', 'high', 'normal'],
'formats': ['flac', 'mp3']
}
},
{
'name': 'qobuz',
'display_name': 'Qobuz',
'enabled': True,
'priority': 2,
'config': {
'quality_preference': ['lossless', 'high', 'normal'],
'formats': ['flac', 'mp3']
}
},
{
'name': 'amazon',
'display_name': 'Amazon Music',
'enabled': False,
'priority': 3,
'config': {
'quality_preference': ['high', 'normal'],
'formats': ['mp3', 'aac']
}
}
],
'maxRetryAttempts': 3,
'cleanupHistoryDays': 30,
'showExplicitWarning': True
}
def get_spotify_settings():
"""Get Spotify downloader settings from config"""
try:
config = UserConfig()
spotify_settings = config.spotify_downloads if hasattr(config, 'spotify_downloads') else {}
# Merge with defaults
settings = {**DEFAULT_SETTINGS}
settings.update(spotify_settings)
return settings
except Exception as e:
logger.error(f"Error loading Spotify settings: {e}")
return DEFAULT_SETTINGS
def save_spotify_settings(settings_data: dict):
"""Save Spotify downloader settings to config"""
try:
config = UserConfig()
# Update only provided settings
current_settings = get_spotify_settings()
current_settings.update(settings_data)
# Save to config
config.spotify_downloads = current_settings
config.save()
logger.info("Spotify settings saved successfully")
return True
except Exception as e:
logger.error(f"Error saving Spotify settings: {e}")
return False
@spotify_settings_bp.get('/', summary='Get Spotify downloader settings')
def get_settings():
"""
Get current Spotify downloader settings
Returns all Spotify downloader configuration including:
- Default quality settings
- Download folder configuration
- Source priorities and enablement
- Advanced options
"""
try:
settings = get_spotify_settings()
return jsonify({
'success': True,
'settings': settings
})
except Exception as e:
logger.error(f"Error getting Spotify settings: {e}")
return jsonify({
'success': False,
'message': str(e)
}), 500
@spotify_settings_bp.post('/', summary='Update Spotify downloader settings')
def update_settings(body: SpotifySettingsRequest):
"""
Update Spotify downloader settings
- **defaultQuality**: Default download quality (flac, mp3_320, mp3_128)
- **downloadFolder**: Custom download folder path
- **autoAddToLibrary**: Whether to auto-add downloads to library
- **maxConcurrentDownloads**: Maximum concurrent downloads (1-10)
- **sources**: Download sources configuration
- **maxRetryAttempts**: Maximum retry attempts for failed downloads
- **cleanupHistoryDays**: Days to keep download history (0 = disabled)
- **showExplicitWarning**: Show warning for explicit content
Updates the Spotify downloader configuration and saves to user settings.
"""
try:
# Validate inputs
if body.defaultQuality not in ['flac', 'mp3_320', 'mp3_128']:
return jsonify({
'success': False,
'message': 'Invalid quality setting'
}), 400
if not 1 <= body.maxConcurrentDownloads <= 10:
return jsonify({
'success': False,
'message': 'Max concurrent downloads must be between 1 and 10'
}), 400
if not 0 <= body.maxRetryAttempts <= 10:
return jsonify({
'success': False,
'message': 'Max retry attempts must be between 0 and 10'
}), 400
if not 0 <= body.cleanupHistoryDays <= 365:
return jsonify({
'success': False,
'message': 'Cleanup days must be between 0 and 365'
}), 400
# Prepare settings data
settings_data = {
'defaultQuality': body.defaultQuality,
'downloadFolder': body.downloadFolder,
'autoAddToLibrary': body.autoAddToLibrary,
'maxConcurrentDownloads': body.maxConcurrentDownloads,
'sources': body.sources,
'maxRetryAttempts': body.maxRetryAttempts,
'cleanupHistoryDays': body.cleanupHistoryDays,
'showExplicitWarning': body.showExplicitWarning
}
# Remove None values
settings_data = {k: v for k, v in settings_data.items() if v is not None}
# Save settings
if save_spotify_settings(settings_data):
return jsonify({
'success': True,
'message': 'Settings saved successfully'
})
else:
return jsonify({
'success': False,
'message': 'Failed to save settings'
}), 500
except Exception as e:
logger.error(f"Error updating Spotify settings: {e}")
return jsonify({
'success': False,
'message': str(e)
}), 500
@spotify_settings_bp.post('/reset', summary='Reset Spotify settings to defaults')
def reset_settings():
"""
Reset all Spotify downloader settings to default values
Resets all Spotify downloader configuration to factory defaults.
"""
try:
if save_spotify_settings(DEFAULT_SETTINGS):
return jsonify({
'success': True,
'message': 'Settings reset to defaults',
'settings': DEFAULT_SETTINGS
})
else:
return jsonify({
'success': False,
'message': 'Failed to reset settings'
}), 500
except Exception as e:
logger.error(f"Error resetting Spotify settings: {e}")
return jsonify({
'success': False,
'message': str(e)
}), 500
@spotify_settings_bp.delete('/queue', summary='Clear download queue')
def clear_queue():
"""
Clear the entire download queue
Removes all pending and active downloads from the queue.
"""
try:
from swingmusic.services.spotify_downloader import spotify_downloader
# Clear queue
spotify_downloader.download_queue.clear()
return jsonify({
'success': True,
'message': 'Download queue cleared'
})
except Exception as e:
logger.error(f"Error clearing download queue: {e}")
return jsonify({
'success': False,
'message': str(e)
}), 500
@spotify_settings_bp.delete('/history', summary='Clear download history')
def clear_history():
"""
Clear the download history
Removes all completed and failed downloads from history.
"""
try:
from swingmusic.services.spotify_downloader import spotify_downloader
# Clear history
spotify_downloader.download_history.clear()
return jsonify({
'success': True,
'message': 'Download history cleared'
})
except Exception as e:
logger.error(f"Error clearing download history: {e}")
return jsonify({
'success': False,
'message': str(e)
}), 500
@spotify_settings_bp.get('/sources', summary='Get available download sources')
def get_available_sources():
"""
Get list of available download sources
Returns information about supported download sources and their capabilities.
"""
try:
sources = [
{
'name': 'tidal',
'display_name': 'Tidal',
'description': 'High-quality FLAC downloads from Tidal',
'quality_options': ['lossless', 'high', 'normal'],
'formats': ['flac', 'mp3'],
'available': True,
'requires_auth': False,
'max_quality': 'lossless'
},
{
'name': 'qobuz',
'display_name': 'Qobuz',
'description': 'Alternative high-quality source with extensive catalog',
'quality_options': ['lossless', 'high', 'normal'],
'formats': ['flac', 'mp3'],
'available': True,
'requires_auth': True,
'max_quality': 'lossless'
},
{
'name': 'amazon',
'display_name': 'Amazon Music',
'description': 'Fallback source with wide availability',
'quality_options': ['high', 'normal'],
'formats': ['mp3', 'aac'],
'available': False, # Disabled by default
'requires_auth': True,
'max_quality': 'high'
}
]
return jsonify({
'success': True,
'sources': sources
})
except Exception as e:
logger.error(f"Error getting available sources: {e}")
return jsonify({
'success': False,
'message': str(e)
}), 500
# Error handlers
@spotify_settings_bp.errorhandler(400)
def bad_request(error):
return jsonify({
'error': 'Bad request',
'message': str(error),
'success': False
}), 400
@spotify_settings_bp.errorhandler(500)
def internal_error(error):
return jsonify({
'error': 'Internal server error',
'message': str(error),
'success': False
}), 500
+160 -7
View File
@@ -1,5 +1,5 @@
"""
Contains all the track routes.
Contains all the track routes with iOS compatibility enhancements.
"""
import os
@@ -18,6 +18,7 @@ from swingmusic.lib.trackslib import get_silence_paddings
from swingmusic.store.tracks import TrackStore
from swingmusic.utils.files import guess_mime_type
from swingmusic.services.ios_audio_compatibility import ios_audio_manager
bp_tag = Tag(name="File", description="Audio files")
api = APIBlueprint("track", __name__, url_prefix="/file", abp_tags=[bp_tag])
@@ -54,11 +55,12 @@ class SendTrackFileQuery(BaseModel):
@api.get("/<trackhash>/legacy")
def send_track_file_legacy(path: TrackHashSchema, query: SendTrackFileQuery):
"""
Get a playable audio file without Range support
Get a playable audio file without Range support (iOS compatible)
Returns a playable audio file that corresponds to the given filepath. Falls back to track hash if filepath is not found.
Automatically handles iOS compatibility by transcoding to supported formats when needed.
NOTE: Does not support range requests or transcoding.
NOTE: Does not support range requests or transcoding beyond iOS compatibility.
"""
requested_trackhash = path.trackhash.strip()
filepath = query.filepath.strip()
@@ -106,14 +108,165 @@ def send_track_file_legacy(path: TrackHashSchema, query: SendTrackFileQuery):
break
if track is not None:
audio_type = guess_mime_type(track.filepath)
return send_from_directory(
Path(track.filepath).parent,
Path(track.filepath).name,
# Detect iOS capabilities and handle compatibility
user_agent = request.headers.get('User-Agent', '')
ios_capabilities = ios_audio_manager.detect_ios_capabilities(user_agent)
# Create iOS-compatible audio source
audio_source = ios_audio_manager.create_ios_audio_source(
track.filepath,
ios_capabilities,
quality="high"
)
# Use the potentially transcoded file path
final_file_path = audio_source['file_path']
audio_type = audio_source['mime_type']
# Add iOS compatibility headers
response = send_from_directory(
Path(final_file_path).parent,
Path(final_file_path).name,
mimetype=audio_type,
conditional=True,
as_attachment=True,
)
# Add iOS-specific headers
if ios_capabilities.is_ios:
response.headers['Accept-Ranges'] = 'bytes'
response.headers['Cache-Control'] = 'public, max-age=3600'
# Add transcoding info if applicable
if audio_source['needs_transcoding']:
response.headers['X-iOS-Transcoded'] = 'true'
response.headers['X-iOS-Original-Format'] = guess_mime_type(track.filepath)
response.headers['X-iOS-Target-Format'] = audio_source['format']
return response
return msg, 404
@api.get("/<trackhash>/ios")
def send_track_file_ios(path: TrackHashSchema, query: SendTrackFileQuery):
"""
Get a playable audio file optimized for iOS devices
Returns a playable audio file optimized for iOS compatibility with automatic transcoding.
Supports FLAC to ALAC/AAC conversion and proper MIME types for iOS Safari and other browsers.
iOS Features:
- Automatic FLAC to ALAC/AAC transcoding
- Proper MP4 container formatting
- iOS-compatible MIME types
- Optimized bitrate for mobile streaming
- Caching for transcoded files
"""
requested_trackhash = path.trackhash.strip()
filepath = query.filepath.strip()
msg = {"msg": "File Not Found"}
# prevent path traversal
if "/../" in filepath:
return {"msg": "Invalid filepath", "error": "Path traversal detected"}, 400
requested_filepath = Path(filepath).resolve()
# check if filepath is a child of any of the root dirs
for root_dir in UserConfig().rootDirs:
if root_dir == "$home":
root_dir = Path.home()
else:
root_dir = Path(root_dir).resolve()
if root_dir not in requested_filepath.parents:
return {
"msg": "Invalid filepath",
"error": "File not inside root directories",
}, 400
track = None
tracks = TrackStore.get_tracks_by_filepaths([filepath])
if len(tracks) > 0 and os.path.exists(tracks[0].filepath):
for t in tracks:
if os.path.exists(t.filepath) and t.trackhash == requested_trackhash:
track = t
break
else:
group = TrackStore.trackhashmap.get(requested_trackhash)
# When finding by trackhash, sort by bitrate
# and get the first track that exists
if group is not None:
tracks = sorted(group.tracks, key=lambda x: x.bitrate, reverse=True)
for t in tracks:
if os.path.exists(t.filepath):
track = t
break
if track is not None:
# Detect iOS capabilities
user_agent = request.headers.get('User-Agent', '')
ios_capabilities = ios_audio_manager.detect_ios_capabilities(user_agent)
# Determine quality based on query parameter or device capabilities
quality_map = {
'original': 'lossless',
'1411': 'lossless',
'1024': 'lossless',
'512': 'high',
'320': 'high',
'256': 'high',
'128': 'medium',
'96': 'low'
}
quality = quality_map.get(query.quality, 'high')
# Create iOS-optimized audio source
audio_source = ios_audio_manager.create_ios_audio_source(
track.filepath,
ios_capabilities,
quality=quality
)
# Use the potentially transcoded file path
final_file_path = audio_source['file_path']
audio_type = audio_source['mime_type']
# Create response with iOS-specific optimizations
response = send_from_directory(
Path(final_file_path).parent,
Path(final_file_path).name,
mimetype=audio_type,
conditional=True,
as_attachment=False, # Stream inline for iOS
)
# iOS-specific headers for optimal playback
response.headers['Accept-Ranges'] = 'bytes'
response.headers['Cache-Control'] = 'public, max-age=7200' # 2 hours
response.headers['X-Content-Type-Options'] = 'nosniff'
# Add iOS compatibility information
if ios_capabilities.is_ios:
response.headers['X-iOS-Optimized'] = 'true'
response.headers['X-iOS-Device'] = 'iPhone' if 'iPhone' in user_agent else 'iPad' if 'iPad' in user_agent else 'iPod'
# Add transcoding information
if audio_source['needs_transcoding']:
response.headers['X-iOS-Transcoded'] = 'true'
response.headers['X-iOS-Original-Format'] = guess_mime_type(track.filepath)
response.headers['X-iOS-Target-Format'] = audio_source['format']
response.headers['X-iOS-Quality'] = quality
else:
response.headers['X-iOS-Transcoded'] = 'false'
response.headers['X-iOS-Native-Format'] = 'true'
return response
return msg, 404
+439
View File
@@ -0,0 +1,439 @@
"""
Universal Music Downloader API for SwingMusic
Supports multiple music streaming services for universal downloading
"""
from flask import Blueprint, request, jsonify
from typing import Dict, List, Any, Optional
import asyncio
from swingmusic.services.universal_music_downloader import universal_music_downloader, DownloadQuality
from swingmusic.services.universal_url_parser import universal_url_parser, MusicService
from swingmusic import logger
# Create blueprint
universal_downloader_bp = Blueprint('universal_downloader', __name__, url_prefix='/api/universal')
@universal_downloader_bp.route('/download', methods=['POST'])
def add_download():
"""
Add a download from any supported music service URL
Request body:
{
"url": "music service URL",
"quality": "lossless|high|medium|low",
"output_dir": "/path/to/output"
}
"""
try:
data = request.get_json()
if not data or not data.get('url'):
return jsonify({'error': 'URL is required'}), 400
url = data['url'].strip()
quality_str = data.get('quality', 'high')
output_dir = data.get('output_dir')
# Validate quality
try:
quality = DownloadQuality(quality_str)
except ValueError:
return jsonify({'error': f'Invalid quality: {quality_str}'}), 400
# Parse URL
parsed_url = universal_music_downloader.parse_url(url)
if not parsed_url:
return jsonify({'error': 'Unsupported URL format'}), 400
# Add to download queue
item_id = universal_music_downloader.add_download(url, quality, output_dir)
if item_id:
return jsonify({
'success': True,
'item_id': item_id,
'service': parsed_url.service.value,
'item_type': parsed_url.item_type,
'message': f'Added to download queue from {parsed_url.service.value}'
})
else:
return jsonify({'error': 'Failed to add download'}), 500
except Exception as e:
logger.error(f"Error adding download: {e}")
return jsonify({'error': 'Internal server error'}), 500
@universal_downloader_bp.route('/metadata', methods=['POST'])
def get_metadata():
"""
Get metadata for any supported music service URL
Request body:
{
"url": "music service URL"
}
"""
try:
data = request.get_json()
if not data or not data.get('url'):
return jsonify({'error': 'URL is required'}), 400
url = data['url'].strip()
# Parse URL
parsed_url = universal_music_downloader.parse_url(url)
if not parsed_url:
return jsonify({'error': 'Unsupported URL format'}), 400
# Get metadata
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
metadata = loop.run_until_complete(universal_music_downloader.get_metadata(url))
finally:
loop.close()
if metadata:
return jsonify({
'success': True,
'service': metadata.service.value,
'service_id': metadata.service_id,
'item_type': parsed_url.item_type,
'title': metadata.title,
'artist': metadata.artist,
'album': metadata.album,
'duration_ms': metadata.duration_ms,
'image_url': metadata.image_url,
'release_date': metadata.release_date,
'explicit': metadata.explicit,
'preview_url': metadata.preview_url,
'genre': metadata.genre,
'original_url': metadata.original_url,
'download_urls': metadata.download_urls
})
else:
return jsonify({'error': 'Failed to get metadata'}), 404
except Exception as e:
logger.error(f"Error getting metadata: {e}")
return jsonify({'error': 'Internal server error'}), 500
@universal_downloader_bp.route('/queue', methods=['GET'])
def get_queue_status():
"""Get current download queue status"""
try:
status = universal_music_downloader.get_queue_status()
return jsonify(status)
except Exception as e:
logger.error(f"Error getting queue status: {e}")
return jsonify({'error': 'Internal server error'}), 500
@universal_downloader_bp.route('/queue/<item_id>/cancel', methods=['POST'])
def cancel_download(item_id: str):
"""Cancel a download"""
try:
success = universal_music_downloader.cancel_download(item_id)
if success:
return jsonify({'success': True, 'message': 'Download cancelled'})
else:
return jsonify({'error': 'Download not found or cannot be cancelled'}), 404
except Exception as e:
logger.error(f"Error cancelling download: {e}")
return jsonify({'error': 'Internal server error'}), 500
@universal_downloader_bp.route('/queue/<item_id>/retry', methods=['POST'])
def retry_download(item_id: str):
"""Retry a failed download"""
try:
success = universal_music_downloader.retry_download(item_id)
if success:
return jsonify({'success': True, 'message': 'Download retry added to queue'})
else:
return jsonify({'error': 'Download not found or cannot be retried'}), 404
except Exception as e:
logger.error(f"Error retrying download: {e}")
return jsonify({'error': 'Internal server error'}), 500
@universal_downloader_bp.route('/history', methods=['GET'])
def get_download_history():
"""
Get download history
Query parameters:
- limit: number of items (default 100)
- offset: offset for pagination (default 0)
- user_id: user ID for filtering (optional)
"""
try:
limit = min(int(request.args.get('limit', 100)), 500)
offset = int(request.args.get('offset', 0))
user_id = request.args.get('user_id')
if user_id:
user_id = int(user_id)
# Get history from universal downloader
# This would need to be implemented in the service
return jsonify({
'downloads': [],
'total': 0,
'limit': limit,
'offset': offset
})
except ValueError:
return jsonify({'error': 'Invalid parameters'}), 400
except Exception as e:
logger.error(f"Error getting download history: {e}")
return jsonify({'error': 'Internal server error'}), 500
@universal_downloader_bp.route('/services', methods=['GET'])
def get_supported_services():
"""Get list of supported music services"""
try:
services = universal_music_downloader.get_supported_services()
return jsonify({
'services': services,
'total': len(services)
})
except Exception as e:
logger.error(f"Error getting supported services: {e}")
return jsonify({'error': 'Internal server error'}), 500
@universal_downloader_bp.route('/services/<service_name>/enable', methods=['POST'])
def enable_service(service_name: str):
"""Enable a music service"""
try:
from swingmusic.db.spotify import UniversalDownloadSourceTable
# Update service in database
UniversalDownloadSourceTable.update_source(service_name, enabled=True)
return jsonify({
'success': True,
'message': f'{service_name} service enabled'
})
except Exception as e:
logger.error(f"Error enabling service: {e}")
return jsonify({'error': 'Internal server error'}), 500
@universal_downloader_bp.route('/services/<service_name>/disable', methods=['POST'])
def disable_service(service_name: str):
"""Disable a music service"""
try:
from swingmusic.db.spotify import UniversalDownloadSourceTable
# Update service in database
UniversalDownloadSourceTable.update_source(service_name, enabled=False)
return jsonify({
'success': True,
'message': f'{service_name} service disabled'
})
except Exception as e:
logger.error(f"Error disabling service: {e}")
return jsonify({'error': 'Internal server error'}), 500
@universal_downloader_bp.route('/services/<service_name>/config', methods=['GET', 'POST'])
def service_config(service_name: str):
"""Get or update service configuration"""
try:
from swingmusic.db.spotify import UniversalDownloadSourceTable
if request.method == 'GET':
source = UniversalDownloadSourceTable.get_by_service(service_name)
if not source:
return jsonify({'error': 'Service not found'}), 404
return jsonify({
'service': source.service,
'display_name': source.display_name,
'enabled': source.enabled,
'priority': source.priority,
'supported_types': source.supported_types,
'features': source.features,
'config': source.config
})
elif request.method == 'POST':
data = request.get_json()
if not data:
return jsonify({'error': 'No data provided'}), 400
# Update only allowed fields
update_data = {}
allowed_fields = ['enabled', 'priority', 'supported_types', 'features', 'config']
for field in allowed_fields:
if field in data:
update_data[field] = data[field]
if update_data:
UniversalDownloadSourceTable.update_source(service_name, **update_data)
return jsonify({'success': True, 'message': 'Service configuration updated'})
except Exception as e:
logger.error(f"Error handling service config: {e}")
return jsonify({'error': 'Internal server error'}), 500
@universal_downloader_bp.route('/validate-url', methods=['POST'])
def validate_url():
"""
Validate and parse a music service URL
Request body:
{
"url": "music service URL"
}
"""
try:
data = request.get_json()
if not data or not data.get('url'):
return jsonify({'error': 'URL is required'}), 400
url = data['url'].strip()
# Parse URL
parsed_url = universal_music_downloader.parse_url(url)
if parsed_url:
return jsonify({
'valid': True,
'service': parsed_url.service.value,
'item_type': parsed_url.item_type,
'id': parsed_url.id,
'metadata': parsed_url.metadata
})
else:
return jsonify({
'valid': False,
'error': 'Unsupported URL format'
})
except Exception as e:
logger.error(f"Error validating URL: {e}")
return jsonify({'error': 'Internal server error'}), 500
@universal_downloader_bp.route('/statistics', methods=['GET'])
def get_statistics():
"""Get download statistics by service"""
try:
from swingmusic.db.spotify import UniversalDownloadTable
stats = UniversalDownloadTable.get_statistics()
return jsonify({
'statistics': stats,
'generated_at': logger.info(f"Statistics generated")
})
except Exception as e:
logger.error(f"Error getting statistics: {e}")
return jsonify({'error': 'Internal server error'}), 500
@universal_downloader_bp.route('/batch', methods=['POST'])
def batch_download():
"""
Add multiple URLs to download queue
Request body:
{
"urls": ["url1", "url2", "url3"],
"quality": "high",
"output_dir": "/path/to/output"
}
"""
try:
data = request.get_json()
if not data or not data.get('urls'):
return jsonify({'error': 'URLs array is required'}), 400
urls = data['urls']
quality_str = data.get('quality', 'high')
output_dir = data.get('output_dir')
if not isinstance(urls, list):
return jsonify({'error': 'URLs must be an array'}), 400
# Validate quality
try:
quality = DownloadQuality(quality_str)
except ValueError:
return jsonify({'error': f'Invalid quality: {quality_str}'}), 400
# Process each URL
results = []
for url in urls:
url = url.strip()
if not url:
continue
try:
# Parse URL
parsed_url = universal_music_downloader.parse_url(url)
if not parsed_url:
results.append({
'url': url,
'success': False,
'error': 'Unsupported URL format'
})
continue
# Add to download queue
item_id = universal_music_downloader.add_download(url, quality, output_dir)
if item_id:
results.append({
'url': url,
'success': True,
'item_id': item_id,
'service': parsed_url.service.value,
'item_type': parsed_url.item_type
})
else:
results.append({
'url': url,
'success': False,
'error': 'Failed to add to queue'
})
except Exception as e:
logger.error(f"Error processing URL {url}: {e}")
results.append({
'url': url,
'success': False,
'error': 'Processing error'
})
successful = sum(1 for r in results if r['success'])
failed = len(results) - successful
return jsonify({
'total': len(results),
'successful': successful,
'failed': failed,
'results': results
})
except Exception as e:
logger.error(f"Error in batch download: {e}")
return jsonify({'error': 'Internal server error'}), 500
def register_universal_downloader_api(app):
"""Register universal downloader API with Flask app"""
app.register_blueprint(universal_downloader_bp)
logger.info("Universal music downloader API registered")
+601
View File
@@ -0,0 +1,601 @@
"""
Update Tracking API Endpoints
This module provides REST API endpoints for the artist update tracking system,
including following artists, managing preferences, and getting updates.
"""
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any
from flask import Blueprint, request, jsonify
from flask_login import login_required, current_user
from swingmusic.db import db
from swingmusic.services.update_tracker import update_tracker, FollowLevel, ReleaseType
from swingmusic.utils.request import APIError, success_response, error_response
from swingmusic.utils.validators import validate_spotify_id, validate_email
logger = logging.getLogger(__name__)
update_tracking_bp = Blueprint('update_tracking', __name__, url_prefix='/api/updates')
def get_current_user_id() -> int:
"""Get current user ID from Flask-Login"""
return current_user.id if current_user.is_authenticated else None
@update_tracking_bp.route('/follow-artist', methods=['POST'])
@login_required
async def follow_artist():
"""
Follow an artist for update tracking
Request Body:
{
"artist_id": "spotify_artist_id",
"artist_name": "Artist Name",
"follow_level": "followed|favorite|casual",
"auto_download": false,
"preferred_quality": "flac"
}
"""
try:
data = request.get_json()
if not data:
return error_response("Request body is required", 400)
# Validate required fields
artist_id = data.get('artist_id')
artist_name = data.get('artist_name')
if not artist_id or not artist_name:
return error_response("artist_id and artist_name are required", 400)
if not validate_spotify_id(artist_id):
return error_response("Invalid artist ID format", 400)
# Validate follow level
follow_level = data.get('follow_level', 'followed')
if follow_level not in ['casual', 'followed', 'favorite']:
return error_response("Invalid follow level. Must be: casual, followed, or favorite", 400)
# Validate quality preference
preferred_quality = data.get('preferred_quality', 'flac')
if preferred_quality not in ['flac', 'mp3_320', 'mp3_256', 'aac']:
return error_response("Invalid quality preference", 400)
follow_data = {
'user_id': get_current_user_id(),
'artist_id': artist_id,
'artist_name': artist_name,
'follow_level': follow_level,
'auto_download': data.get('auto_download', False),
'preferred_quality': preferred_quality
}
success = await update_tracker.follow_artist(follow_data)
if success:
return success_response({
'message': f'Now following {artist_name}',
'artist_id': artist_id,
'follow_level': follow_level
})
else:
return error_response("Failed to follow artist", 500)
except Exception as e:
logger.error(f"Error following artist: {e}")
return error_response("Internal server error", 500)
@update_tracking_bp.route('/unfollow-artist', methods=['POST'])
@login_required
async def unfollow_artist():
"""
Unfollow an artist
Request Body:
{
"artist_id": "spotify_artist_id"
}
"""
try:
data = request.get_json()
if not data or not data.get('artist_id'):
return error_response("artist_id is required", 400)
artist_id = data['artist_id']
if not validate_spotify_id(artist_id):
return error_response("Invalid artist ID format", 400)
success = await update_tracker.unfollow_artist(get_current_user_id(), artist_id)
if success:
return success_response({
'message': 'Artist unfollowed successfully',
'artist_id': artist_id
})
else:
return error_response("Failed to unfollow artist", 500)
except Exception as e:
logger.error(f"Error unfollowing artist: {e}")
return error_response("Internal server error", 500)
@update_tracking_bp.route('/recent', methods=['GET'])
@login_required
async def get_recent_updates():
"""
Get recent updates for followed artists
Query Parameters:
- limit: Number of updates to return (default: 20, max: 100)
- offset: Offset for pagination (default: 0)
- release_type: Filter by release type (album, single, ep, compilation)
- unread_only: Only return unread updates (true/false)
"""
try:
limit = min(request.args.get('limit', 20, type=int), 100)
offset = request.args.get('offset', 0, type=int)
release_type = request.args.get('release_type')
unread_only = request.args.get('unread_only', 'false').lower() == 'true'
# Validate release type
if release_type and release_type not in ['album', 'single', 'ep', 'compilation']:
return error_response("Invalid release type", 400)
updates = await update_tracker.get_user_updates(
get_current_user_id(),
limit=limit,
offset=offset,
release_type=release_type,
unread_only=unread_only
)
return success_response({
'updates': updates,
'limit': limit,
'offset': offset,
'total': len(updates)
})
except Exception as e:
logger.error(f"Error getting recent updates: {e}")
return error_response("Internal server error", 500)
@update_tracking_bp.route('/settings', methods=['GET'])
@login_required
async def get_settings():
"""
Get user's update tracking settings
"""
try:
settings = await update_tracker.get_user_settings(get_current_user_id())
return success_response(settings)
except Exception as e:
logger.error(f"Error getting settings: {e}")
return error_response("Internal server error", 500)
@update_tracking_bp.route('/settings', methods=['POST'])
@login_required
async def update_settings():
"""
Update user's update tracking settings
Request Body:
{
"enable_artist_monitoring": true,
"check_frequency": "daily",
"auto_download_favorites": false,
"auto_download_followed": false,
"max_auto_downloads_per_week": 5,
"quality_preference": "flac",
"storage_limit_mb": 10240,
"notification_channels": {
"in_app": true,
"push": false,
"email": false,
"discord": false
},
"exclude_explicit": false,
"preferred_release_types": ["album", "ep", "single"]
}
"""
try:
data = request.get_json()
if not data:
return error_response("Request body is required", 400)
# Validate settings
if 'check_frequency' in data and data['check_frequency'] not in ['hourly', 'daily', 'weekly']:
return error_response("Invalid check frequency", 400)
if 'quality_preference' in data and data['quality_preference'] not in ['flac', 'mp3_320', 'mp3_256', 'aac']:
return error_response("Invalid quality preference", 400)
if 'max_auto_downloads_per_week' in data:
max_downloads = data['max_auto_downloads_per_week']
if not isinstance(max_downloads, int) or max_downloads < 0 or max_downloads > 50:
return error_response("Invalid max auto downloads value", 400)
if 'storage_limit_mb' in data:
storage_limit = data['storage_limit_mb']
if not isinstance(storage_limit, int) or storage_limit < 100 or storage_limit > 102400:
return error_response("Invalid storage limit", 400)
success = await update_tracker.update_user_settings(get_current_user_id(), data)
if success:
return success_response({
'message': 'Settings updated successfully',
'settings': data
})
else:
return error_response("Failed to update settings", 500)
except Exception as e:
logger.error(f"Error updating settings: {e}")
return error_response("Internal server error", 500)
@update_tracking_bp.route('/auto-download/<release_id>', methods=['POST'])
@login_required
async def auto_download_release(release_id):
"""
Trigger auto-download for a specific release
Path Parameters:
- release_id: Spotify release ID
"""
try:
if not validate_spotify_id(release_id):
return error_response("Invalid release ID format", 400)
success = await update_tracker.auto_download_release(get_current_user_id(), release_id)
if success:
return success_response({
'message': 'Download queued successfully',
'release_id': release_id
})
else:
return error_response("Failed to queue download", 500)
except Exception as e:
logger.error(f"Error auto-downloading release: {e}")
return error_response("Internal server error", 500)
@update_tracking_bp.route('/stats', methods=['GET'])
@login_required
async def get_update_stats():
"""
Get user's update tracking statistics
"""
try:
stats = await update_tracker.get_user_stats(get_current_user_id())
return success_response(stats)
except Exception as e:
logger.error(f"Error getting stats: {e}")
return error_response("Internal server error", 500)
@update_tracking_bp.route('/followed-artists', methods=['GET'])
@login_required
async def get_followed_artists():
"""
Get list of followed artists
Query Parameters:
- limit: Number of artists to return (default: 50, max: 200)
- offset: Offset for pagination (default: 0)
- follow_level: Filter by follow level (casual, followed, favorite)
"""
try:
limit = min(request.args.get('limit', 50, type=int), 200)
offset = request.args.get('offset', 0, type=int)
follow_level = request.args.get('follow_level')
# Validate follow level
if follow_level and follow_level not in ['casual', 'followed', 'favorite']:
return error_response("Invalid follow level", 400)
artists = await update_tracker.get_followed_artists(
get_current_user_id(),
limit=limit,
offset=offset,
follow_level=follow_level
)
return success_response({
'artists': artists,
'limit': limit,
'offset': offset,
'total': len(artists)
})
except Exception as e:
logger.error(f"Error getting followed artists: {e}")
return error_response("Internal server error", 500)
@update_tracking_bp.route('/artist/<artist_id>/follow-status', methods=['GET'])
@login_required
async def get_artist_follow_status(artist_id):
"""
Get follow status for a specific artist
Path Parameters:
- artist_id: Spotify artist ID
"""
try:
if not validate_spotify_id(artist_id):
return error_response("Invalid artist ID format", 400)
status = await update_tracker.get_artist_follow_status(get_current_user_id(), artist_id)
if status:
return success_response(status)
else:
return success_response({
'is_following': False,
'artist_id': artist_id
})
except Exception as e:
logger.error(f"Error getting artist follow status: {e}")
return error_response("Internal server error", 500)
@update_tracking_bp.route('/artist/<artist_id>', methods=['PUT'])
@login_required
async def update_artist_follow(artist_id):
"""
Update follow settings for an artist
Path Parameters:
- artist_id: Spotify artist ID
Request Body:
{
"follow_level": "followed|favorite|casual",
"auto_download": true,
"preferred_quality": "flac",
"notification_preferences": {
"in_app": true,
"push": false,
"email": false,
"discord": false
}
}
"""
try:
if not validate_spotify_id(artist_id):
return error_response("Invalid artist ID format", 400)
data = request.get_json()
if not data:
return error_response("Request body is required", 400)
# Validate follow level
if 'follow_level' in data and data['follow_level'] not in ['casual', 'followed', 'favorite']:
return error_response("Invalid follow level", 400)
# Validate quality preference
if 'preferred_quality' in data and data['preferred_quality'] not in ['flac', 'mp3_320', 'mp3_256', 'aac']:
return error_response("Invalid quality preference", 400)
success = await update_tracker.update_artist_follow(
get_current_user_id(),
artist_id,
data
)
if success:
return success_response({
'message': 'Artist follow settings updated',
'artist_id': artist_id,
'settings': data
})
else:
return error_response("Failed to update artist follow settings", 500)
except Exception as e:
logger.error(f"Error updating artist follow: {e}")
return error_response("Internal server error", 500)
@update_tracking_bp.route('/release/<release_id>', methods=['GET'])
@login_required
async def get_release_details(release_id):
"""
Get details for a specific release update
Path Parameters:
- release_id: Spotify release ID
"""
try:
if not validate_spotify_id(release_id):
return error_response("Invalid release ID format", 400)
release = await update_tracker.get_release_details(get_current_user_id(), release_id)
if release:
return success_response(release)
else:
return error_response("Release not found", 404)
except Exception as e:
logger.error(f"Error getting release details: {e}")
return error_response("Internal server error", 500)
@update_tracking_bp.route('/release/<release_id>/mark-read', methods=['POST'])
@login_required
async def mark_release_read(release_id):
"""
Mark a release update as read
Path Parameters:
- release_id: Spotify release ID
"""
try:
if not validate_spotify_id(release_id):
return error_response("Invalid release ID format", 400)
success = await update_tracker.mark_release_read(get_current_user_id(), release_id)
if success:
return success_response({
'message': 'Release marked as read',
'release_id': release_id
})
else:
return error_response("Failed to mark release as read", 500)
except Exception as e:
logger.error(f"Error marking release as read: {e}")
return error_response("Internal server error", 500)
@update_tracking_bp.route('/notifications', methods=['GET'])
@login_required
async def get_notifications():
"""
Get user's update notifications
Query Parameters:
- limit: Number of notifications to return (default: 20, max: 100)
- offset: Offset for pagination (default: 0)
- unread_only: Only return unread notifications (true/false)
"""
try:
limit = min(request.args.get('limit', 20, type=int), 100)
offset = request.args.get('offset', 0, type=int)
unread_only = request.args.get('unread_only', 'false').lower() == 'true'
notifications = await update_tracker.get_notifications(
get_current_user_id(),
limit=limit,
offset=offset,
unread_only=unread_only
)
return success_response({
'notifications': notifications,
'limit': limit,
'offset': offset,
'total': len(notifications)
})
except Exception as e:
logger.error(f"Error getting notifications: {e}")
return error_response("Internal server error", 500)
@update_tracking_bp.route('/notifications/mark-all-read', methods=['POST'])
@login_required
async def mark_all_notifications_read():
"""
Mark all notifications as read for the user
"""
try:
success = await update_tracker.mark_all_notifications_read(get_current_user_id())
if success:
return success_response({
'message': 'All notifications marked as read'
})
else:
return error_response("Failed to mark notifications as read", 500)
except Exception as e:
logger.error(f"Error marking all notifications as read: {e}")
return error_response("Internal server error", 500)
@update_tracking_bp.route('/search/artists', methods=['GET'])
@login_required
async def search_artists_to_follow():
"""
Search for artists to follow
Query Parameters:
- q: Search query
- limit: Number of results to return (default: 10, max: 50)
"""
try:
query = request.args.get('q')
if not query:
return error_response("Search query is required", 400)
limit = min(request.args.get('limit', 10, type=int), 50)
artists = await update_tracker.search_artists(query, limit)
return success_response({
'artists': artists,
'query': query,
'limit': limit,
'total': len(artists)
})
except Exception as e:
logger.error(f"Error searching artists: {e}")
return error_response("Internal server error", 500)
@update_tracking_bp.route('/export/followed-artists', methods=['GET'])
@login_required
async def export_followed_artists():
"""
Export followed artists as JSON or CSV
Query Parameters:
- format: Export format (json|csv) - default: json
"""
try:
export_format = request.args.get('format', 'json').lower()
if export_format not in ['json', 'csv']:
return error_response("Invalid export format. Must be json or csv", 400)
data = await update_tracker.export_followed_artists(get_current_user_id(), export_format)
if export_format == 'csv':
from flask import Response
return Response(
data,
mimetype='text/csv',
headers={'Content-Disposition': 'attachment; filename=followed_artists.csv'}
)
else:
return success_response({'followed_artists': data})
except Exception as e:
logger.error(f"Error exporting followed artists: {e}")
return error_response("Internal server error", 500)
# Error handlers
@update_tracking_bp.errorhandler(404)
def not_found(error):
return error_response("Endpoint not found", 404)
@update_tracking_bp.errorhandler(500)
def internal_error(error):
return error_response("Internal server error", 500)
+392
View File
@@ -0,0 +1,392 @@
"""
Contains all the file upload routes for manual music upload functionality.
"""
import os
import shutil
import pathlib
from pathlib import Path
from datetime import datetime
from typing import List, Optional
import tempfile
import mimetypes
from flask import request, jsonify
from flask_openapi3 import Tag
from pydantic import BaseModel, Field
from flask_openapi3 import APIBlueprint
from werkzeug.utils import secure_filename
from swingmusic import settings
from swingmusic.config import UserConfig
from swingmusic.db.libdata import TrackTable
from swingmusic.api.auth import admin_required
from swingmusic.store.tracks import TrackStore
from swingmusic.utils.metadata import extract_metadata
from swingmusic.serializers.track import serialize_track
tag = Tag(name="Upload", description="Manual music file upload functionality")
api = APIBlueprint("upload", __name__, url_prefix="/upload", abp_tags=[tag])
# Allowed audio file extensions
ALLOWED_EXTENSIONS = {
'mp3', 'flac', 'wav', 'aac', 'm4a', 'ogg', 'wma', 'opus',
'aiff', 'au', 'ra', '3gp', 'amr', 'awb', 'dct', 'dvf',
'm4p', 'mmf', 'mpc', 'msv', 'nmf', 'nsf', 'ogg', 'qcp',
'ra', 'rm', 'sln', 'vox', 'wma', 'wv'
}
# Maximum file size (100MB)
MAX_FILE_SIZE = 100 * 1024 * 1024
def is_allowed_file(filename: str) -> bool:
"""Check if file has an allowed audio extension."""
return '.' in filename and \
filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
def is_path_within_root_dirs(filepath: str) -> bool:
"""
Check if a filepath is within one of the configured root directories.
Prevents directory traversal attacks.
"""
config = UserConfig()
resolved_path = Path(filepath).resolve()
for root_dir in config.rootDirs:
if root_dir == "$home":
root_path = Path.home().resolve()
else:
root_path = Path(root_dir).resolve()
# Check if resolved_path is the root or a child of root
if resolved_path == root_path or root_path in resolved_path.parents:
return True
return False
class UploadResponse(BaseModel):
success: bool = Field(description="Whether the upload was successful")
message: str = Field(description="Status message")
track_id: Optional[str] = Field(None, description="ID of the added track")
filename: Optional[str] = Field(None, description="Name of the uploaded file")
class BatchUploadResponse(BaseModel):
success: bool = Field(description="Whether the batch upload was successful")
message: str = Field(description="Status message")
uploaded_files: List[UploadResponse] = Field(description="List of upload results")
failed_files: List[str] = Field(description="List of failed files")
@api.post("/single")
@admin_required()
def upload_single_file():
"""
Upload a single music file
Uploads a single music file to the configured music folder and adds it to the library.
Supports drag-and-drop and file selection.
"""
try:
if 'file' not in request.files:
return jsonify({
"success": False,
"message": "No file provided"
}), 400
file = request.files['file']
if file.filename == '':
return jsonify({
"success": False,
"message": "No file selected"
}), 400
# Check file extension
if not is_allowed_file(file.filename):
return jsonify({
"success": False,
"message": f"File type not allowed. Supported formats: {', '.join(sorted(ALLOWED_EXTENSIONS))}"
}), 400
# Check file size
file.seek(0, os.SEEK_END)
file_size = file.tell()
file.seek(0)
if file_size > MAX_FILE_SIZE:
return jsonify({
"success": False,
"message": f"File too large. Maximum size is {MAX_FILE_SIZE // (1024*1024)}MB"
}), 400
# Get upload directory from settings or use first root directory
config = UserConfig()
upload_dir = None
# Check if there's a specific upload directory configured
if hasattr(config, 'uploadDir') and config.uploadDir:
upload_dir = Path(config.uploadDir)
else:
# Use the first root directory as default
if config.rootDirs:
first_root = config.rootDirs[0]
if first_root == "$home":
upload_dir = Path.home() / "Music"
else:
upload_dir = Path(first_root)
else:
# Fallback to user's Music directory
upload_dir = Path.home() / "Music"
# Ensure upload directory exists
upload_dir.mkdir(parents=True, exist_ok=True)
# Secure the filename and create full path
filename = secure_filename(file.filename)
file_path = upload_dir / filename
# Handle filename conflicts
counter = 1
original_filename = filename
while file_path.exists():
name, ext = os.path.splitext(original_filename)
filename = f"{name}_{counter}{ext}"
file_path = upload_dir / filename
counter += 1
# Save the file
file.save(file_path)
# Extract metadata and add to library
try:
# This would trigger a library rescan for the specific file
# For now, we'll return the file info and let the frontend handle the refresh
track_info = {
"filepath": str(file_path),
"filename": filename,
"size": file_size
}
return jsonify({
"success": True,
"message": f"File '{filename}' uploaded successfully",
"filename": filename,
"filepath": str(file_path),
"track_info": track_info
})
except Exception as e:
# If metadata extraction fails, still return success for the upload
return jsonify({
"success": True,
"message": f"File '{filename}' uploaded successfully (metadata extraction failed)",
"filename": filename,
"filepath": str(file_path),
"warning": f"Metadata extraction failed: {str(e)}"
})
except Exception as e:
return jsonify({
"success": False,
"message": f"Upload failed: {str(e)}"
}), 500
@api.post("/batch")
@admin_required()
def upload_multiple_files():
"""
Upload multiple music files
Uploads multiple music files to the configured music folder and adds them to the library.
Supports drag-and-drop of multiple files.
"""
try:
if 'files' not in request.files:
return jsonify({
"success": False,
"message": "No files provided"
}), 400
files = request.files.getlist('files')
if not files:
return jsonify({
"success": False,
"message": "No files selected"
}), 400
uploaded_files = []
failed_files = []
# Get upload directory (same logic as single upload)
config = UserConfig()
upload_dir = None
if hasattr(config, 'uploadDir') and config.uploadDir:
upload_dir = Path(config.uploadDir)
else:
if config.rootDirs:
first_root = config.rootDirs[0]
if first_root == "$home":
upload_dir = Path.home() / "Music"
else:
upload_dir = Path(first_root)
else:
upload_dir = Path.home() / "Music"
upload_dir.mkdir(parents=True, exist_ok=True)
for file in files:
if file.filename == '':
continue
try:
# Check file extension
if not is_allowed_file(file.filename):
failed_files.append(f"{file.filename} - File type not allowed")
continue
# Check file size
file.seek(0, os.SEEK_END)
file_size = file.tell()
file.seek(0)
if file_size > MAX_FILE_SIZE:
failed_files.append(f"{file.filename} - File too large")
continue
# Secure filename and handle conflicts
filename = secure_filename(file.filename)
file_path = upload_dir / filename
counter = 1
original_filename = filename
while file_path.exists():
name, ext = os.path.splitext(original_filename)
filename = f"{name}_{counter}{ext}"
file_path = upload_dir / filename
counter += 1
# Save the file
file.save(file_path)
uploaded_files.append({
"success": True,
"message": f"File '{filename}' uploaded successfully",
"filename": filename,
"filepath": str(file_path),
"size": file_size
})
except Exception as e:
failed_files.append(f"{file.filename} - {str(e)}")
total_files = len(uploaded_files) + len(failed_files)
success_count = len(uploaded_files)
return jsonify({
"success": len(uploaded_files) > 0,
"message": f"Uploaded {success_count} of {total_files} files",
"uploaded_files": uploaded_files,
"failed_files": failed_files
})
except Exception as e:
return jsonify({
"success": False,
"message": f"Batch upload failed: {str(e)}"
}), 500
@api.get("/config")
def get_upload_config():
"""
Get upload configuration
Returns the current upload configuration including allowed file types,
maximum file size, and upload directory.
"""
config = UserConfig()
# Determine upload directory
upload_dir = None
if hasattr(config, 'uploadDir') and config.uploadDir:
upload_dir = config.uploadDir
elif config.rootDirs:
first_root = config.rootDirs[0]
if first_root == "$home":
upload_dir = str(Path.home() / "Music")
else:
upload_dir = first_root
else:
upload_dir = str(Path.home() / "Music")
return jsonify({
"allowed_extensions": sorted(list(ALLOWED_EXTENSIONS)),
"max_file_size": MAX_FILE_SIZE,
"max_file_size_mb": MAX_FILE_SIZE // (1024 * 1024),
"upload_directory": upload_dir,
"supported_formats": [
{"ext": ext, "description": get_format_description(ext)}
for ext in sorted(ALLOWED_EXTENSIONS)
]
})
def get_format_description(extension: str) -> str:
"""Get a user-friendly description for a file format."""
descriptions = {
'mp3': 'MP3 Audio',
'flac': 'FLAC Lossless Audio',
'wav': 'WAV Audio',
'aac': 'AAC Audio',
'm4a': 'M4A Audio',
'ogg': 'OGG Vorbis Audio',
'wma': 'WMA Audio',
'opus': 'Opus Audio',
'aiff': 'AIFF Audio',
'au': 'AU Audio',
'ra': 'RealAudio',
'3gp': '3GP Audio',
'amr': 'AMR Audio',
'awb': 'AWB Audio',
'dct': 'DCT Audio',
'dvf': 'DVF Audio',
'm4p': 'M4P Audio',
'mmf': 'MMF Audio',
'mpc': 'MPC Audio',
'msv': 'MSV Audio',
'nmf': 'NMF Audio',
'nsf': 'NSF Audio',
'qcp': 'QCP Audio',
'rm': 'RealMedia Audio',
'sln': 'SLN Audio',
'vox': 'VOX Audio',
'wv': 'WavPack Audio'
}
return descriptions.get(extension.lower(), f'{extension.upper()} Audio')
@api.post("/rescan")
@admin_required()
def trigger_library_rescan():
"""
Trigger library rescan
Triggers a library rescan to detect newly uploaded files.
"""
try:
# This would integrate with the existing library scanning system
# For now, return a success response
return jsonify({
"success": True,
"message": "Library rescan triggered successfully"
})
except Exception as e:
return jsonify({
"success": False,
"message": f"Failed to trigger library rescan: {str(e)}"
}), 500