| from flask import Flask, request, jsonify |
| from flask_cors import CORS |
| from deepface import DeepFace |
| import base64 |
| import io |
| from PIL import Image |
| import numpy as np |
| import logging |
| import traceback |
| import os |
| import spotipy |
| from spotipy.oauth2 import SpotifyClientCredentials |
|
|
| |
| os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2' |
| os.environ['TF_ENABLE_ONEDNN_OPTS'] = '0' |
|
|
| |
| logging.basicConfig( |
| level=logging.INFO, |
| format='%(asctime)s - %(levelname)s - %(message)s' |
| ) |
| logger = logging.getLogger(__name__) |
|
|
| app = Flask(__name__) |
| CORS(app) |
|
|
| |
| try: |
| |
| SPOTIFY_CLIENT_ID = os.getenv('SPOTIFY_CLIENT_ID', 'your_client_id_here') |
| SPOTIFY_CLIENT_SECRET = os.getenv('SPOTIFY_CLIENT_SECRET', 'your_client_secret_here') |
| |
| client_credentials_manager = SpotifyClientCredentials( |
| client_id=SPOTIFY_CLIENT_ID, |
| client_secret=SPOTIFY_CLIENT_SECRET |
| ) |
| spotify = spotipy.Spotify(client_credentials_manager=client_credentials_manager) |
| logger.info("Spotify API initialized successfully") |
| except Exception as e: |
| logger.error(f"Failed to initialize Spotify API: {str(e)}") |
| spotify = None |
|
|
| |
| EMOTION_TO_MOOD = { |
| 'happy': ['happy hits', 'feel good', 'party music', 'upbeat pop'], |
| 'sad': ['sad songs', 'emotional ballads', 'melancholic', 'rainy day'], |
| 'angry': ['workout motivation', 'rock anthems', 'intense metal', 'aggressive'], |
| 'neutral': ['chill vibes', 'focus music', 'ambient', 'lo-fi beats'], |
| 'surprise': ['party hits', 'dance pop', 'exciting', 'upbeat'], |
| 'fear': ['calming music', 'meditation', 'peaceful', 'relaxation'], |
| 'disgust': ['energizing', 'workout', 'rock music', 'alternative'] |
| } |
|
|
| @app.route('/', methods=['GET']) |
| def home(): |
| return jsonify({ |
| "message": "Emotion Detection + Music Recommendation API", |
| "endpoints": { |
| "/emotion": "POST - Detect emotion from image", |
| "/health": "GET - Health check" |
| }, |
| "music_provider": "Spotify" |
| }) |
|
|
| @app.route('/health', methods=['GET']) |
| def health(): |
| return jsonify({ |
| "status": "healthy", |
| "spotify_available": spotify is not None |
| }), 200 |
|
|
| @app.route('/emotion', methods=['POST', 'OPTIONS']) |
| def detect_emotion(): |
| if request.method == 'OPTIONS': |
| return jsonify({}), 200 |
| |
| try: |
| logger.info("Received emotion detection request") |
| |
| data = request.get_json() |
| |
| if not data or 'image' not in data: |
| logger.error("No image data in request") |
| return jsonify({ |
| "success": False, |
| "error": "No image data provided" |
| }), 400 |
| |
| image_data = data['image'] |
| logger.info(f"Received image data, length: {len(image_data)}") |
| |
| if ',' in image_data: |
| image_data = image_data.split(',')[1] |
| |
| try: |
| image_bytes = base64.b64decode(image_data) |
| logger.info(f"Decoded image bytes: {len(image_bytes)} bytes") |
| except Exception as e: |
| logger.error(f"Base64 decode error: {str(e)}") |
| return jsonify({ |
| "success": False, |
| "error": "Invalid base64 image data" |
| }), 400 |
| |
| try: |
| image = Image.open(io.BytesIO(image_bytes)) |
| logger.info(f"Image opened - Format: {image.format}, Size: {image.size}, Mode: {image.mode}") |
| except Exception as e: |
| logger.error(f"PIL image open error: {str(e)}") |
| return jsonify({ |
| "success": False, |
| "error": "Invalid image format" |
| }), 400 |
| |
| if image.mode != 'RGB': |
| logger.info(f"Converting image from {image.mode} to RGB") |
| image = image.convert('RGB') |
| |
| img_array = np.array(image) |
| logger.info(f"Numpy array shape: {img_array.shape}, dtype: {img_array.dtype}") |
| |
| try: |
| logger.info("Starting DeepFace analysis...") |
| result = DeepFace.analyze( |
| img_array, |
| actions=['emotion'], |
| enforce_detection=False, |
| silent=True, |
| detector_backend='opencv' |
| ) |
| logger.info("DeepFace analysis completed successfully") |
| except Exception as e: |
| logger.error(f"DeepFace analysis error: {str(e)}") |
| logger.error(traceback.format_exc()) |
| return jsonify({ |
| "success": False, |
| "error": f"Emotion detection failed: {str(e)}" |
| }), 500 |
| |
| if isinstance(result, list): |
| result = result[0] |
| |
| dominant_emotion = result['dominant_emotion'] |
| emotion_scores = result['emotion'] |
| |
| emotion_scores_serializable = { |
| emotion: float(score) for emotion, score in emotion_scores.items() |
| } |
| |
| logger.info(f"Dominant emotion: {dominant_emotion}") |
| |
| music_recommendations = [] |
| if spotify: |
| try: |
| music_recommendations = get_music_from_spotify(dominant_emotion) |
| logger.info(f"Found {len(music_recommendations)} music recommendations") |
| except Exception as e: |
| logger.warning(f"Failed to get music recommendations: {str(e)}") |
| |
| return jsonify({ |
| "success": True, |
| "dominant_emotion": dominant_emotion, |
| "all_emotions": emotion_scores_serializable, |
| "confidence": float(emotion_scores[dominant_emotion]), |
| "music_recommendations": music_recommendations, |
| "suggested_moods": EMOTION_TO_MOOD.get(dominant_emotion, []) |
| }) |
| |
| except Exception as e: |
| logger.error(f"Unexpected error: {str(e)}") |
| logger.error(traceback.format_exc()) |
| return jsonify({ |
| "success": False, |
| "error": f"Internal server error: {str(e)}" |
| }), 500 |
|
|
| def get_music_from_spotify(emotion): |
| """Get music playlists from Spotify based on emotion""" |
| try: |
| mood_keywords = EMOTION_TO_MOOD.get(emotion, ['chill']) |
| recommendations = [] |
| |
| for keyword in mood_keywords[:3]: |
| try: |
| results = spotify.search(q=keyword, type='playlist', limit=3) |
| |
| for playlist in results['playlists']['items']: |
| if playlist: |
| |
| thumbnail = '' |
| if playlist.get('images') and len(playlist['images']) > 0: |
| thumbnail = playlist['images'][0]['url'] |
| |
| recommendations.append({ |
| 'title': playlist.get('name', ''), |
| 'playlist_id': playlist.get('id', ''), |
| 'playlist_url': playlist.get('external_urls', {}).get('spotify', ''), |
| 'thumbnail': thumbnail, |
| 'author': playlist.get('owner', {}).get('display_name', 'Spotify'), |
| 'item_count': playlist.get('tracks', {}).get('total', 0), |
| 'mood': keyword, |
| 'provider': 'spotify' |
| }) |
| |
| if len(recommendations) >= 6: |
| break |
| |
| except Exception as e: |
| logger.warning(f"Search failed for keyword '{keyword}': {str(e)}") |
| continue |
| |
| return recommendations[:6] |
| |
| except Exception as e: |
| logger.error(f"Error getting music recommendations: {str(e)}") |
| return [] |
|
|
| def start(): |
| logger.info("Starting Emotion Detection + Music Recommendation API on port 7860...") |
| app.run(host='0.0.0.0', port=7860) |
|
|
| if __name__ == '__main__': |
| start() |
|
|