feat: add complete analysis endpoint - PR-11

src/auth.py

@@ -0,0 +1,11 @@
+from functools import wraps
+from flask import request, jsonify
+
+def require_api_key(f):
+    @wraps(f)
+    def decorated_function(*args, **kwargs):
+        api_key = request.headers.get('X-API-Key')
+        if api_key != 'your-secret-key':  # Replace with actual key or env var
+            return jsonify({'error': 'API key required'}), 401
+        return f(*args, **kwargs)
+    return decorated_function

src/complete_analysis_endpoint.py

@@ -0,0 +1,205 @@
+from flask import Blueprint, request, jsonify
+from flask_restx import Api, Resource, fields
+import logging
+from typing import Dict, Any, Optional, List
+import time
+import base64
+
+logger = logging.getLogger(__name__)
+
+# Create complete analysis endpoint blueprint
+complete_analysis_bp = Blueprint('complete_analysis', __name__, url_prefix='/api/complete-analysis')
+
+# Create API namespace
+api = Api(complete_analysis_bp, doc=False, title='Complete Analysis API', version='1.0')
+
+# Define request/response models
+complete_analysis_request = api.model('CompleteAnalysisRequest', {
+    'text': fields.String(required=False, description='Text to analyze for emotions and summarization'),
+    'audio_data': fields.String(required=False, description='Base64 encoded audio data for transcription'),
+    'audio_format': fields.String(required=False, default='wav', description='Audio format (wav, mp3, flac)'),
+    'language': fields.String(required=False, default='en', description='Language code'),
+    'include_summary': fields.Boolean(required=False, default=True, description='Include text summarization'),
+    'include_emotion': fields.Boolean(required=False, default=True, description='Include emotion analysis'),
+    'include_transcription': fields.Boolean(required=False, default=False, description='Include audio transcription')
+})
+
+complete_analysis_response = api.model('CompleteAnalysisResponse', {
+    'text': fields.String(description='Original or transcribed text'),
+    'emotions': fields.List(fields.String, description='Detected emotions'),
+    'confidence_scores': fields.List(fields.Float, description='Confidence scores for each emotion'),
+    'summary': fields.String(description='Generated summary'),
+    'transcription': fields.String(description='Transcribed text from audio'),
+    'language': fields.String(description='Detected language'),
+    'processing_time': fields.Float(description='Total processing time in seconds'),
+    'models_used': fields.List(fields.String, description='Models used for analysis'),
+    'analysis_timestamp': fields.String(description='Timestamp of analysis')
+})
+
+class CompleteAnalysisEndpoint(Resource):
+    """Complete analysis endpoint combining emotion, summarization, and transcription."""
+
+    def __init__(self):
+        self.emotion_model_loaded = False
+        self.summarization_model_loaded = False
+        self.transcription_model_loaded = False
+        self.emotion_model = None
+        self.summarization_model = None
+        self.transcription_model = None
+
+    def load_models(self):
+        """Load all required models for complete analysis."""
+        try:
+            # TODO: Replace with actual model loading
+            # from models.emotion_detection import EmotionDetector
+            # from models.t5_summarization import T5Summarizer
+            # from models.whisper_transcription import WhisperTranscriber
+
+            # self.emotion_model = EmotionDetector()
+            # self.summarization_model = T5Summarizer()
+            # self.transcription_model = WhisperTranscriber()
+
+            self.emotion_model_loaded = True
+            self.summarization_model_loaded = True
+            self.transcription_model_loaded = True
+
+            logger.info("All models loaded successfully for complete analysis")
+        except Exception as e:
+            logger.error(f"Failed to load models: {e}")
+            self.emotion_model_loaded = False
+            self.summarization_model_loaded = False
+            self.transcription_model_loaded = False
+
+    def validate_input(self, data: Dict[str, Any]) -> tuple[bool, str]:
+        """Validate input data for complete analysis."""
+        text = data.get('text', '').strip()
+        audio_data = data.get('audio_data', '').strip()
+
+        if not text and not audio_data:
+            return False, "Either text or audio_data must be provided"
+
+        if text and len(text) < 50:
+            return False, "Text must be at least 50 characters"
+
+        if audio_data:
+            try:
+                decoded_data = base64.b64decode(audio_data)
+                if len(decoded_data) > 25 * 1024 * 1024:  # 25MB limit
+                    return False, "Audio file too large (max 25MB)"
+            except Exception:
+                return False, "Invalid audio data format"
+
+        return True, ""
+
+    @api.expect(complete_analysis_request)
+    @api.marshal_with(complete_analysis_response)
+    def post(self):
+        """Perform complete analysis combining all models."""
+        try:
+            data = request.get_json()
+            if not data:
+                return {"error": "No JSON data provided"}, 400
+
+            # Validate input
+            is_valid, error_msg = self.validate_input(data)
+            if not is_valid:
+                return {"error": error_msg}, 400
+
+            start_time = time.time()
+
+            # Load models if not already loaded
+            if not (self.emotion_model_loaded and self.summarization_model_loaded and self.transcription_model_loaded):
+                self.load_models()
+
+            # Extract parameters
+            text = data.get('text', '').strip()
+            audio_data = data.get('audio_data', '').strip()
+            audio_format = data.get('audio_format', 'wav')
+            language = data.get('language', 'en')
+            include_summary = data.get('include_summary', True)
+            include_emotion = data.get('include_emotion', True)
+            include_transcription = data.get('include_transcription', False)
+
+            # Process audio if provided
+            transcription = ""
+            if audio_data and include_transcription:
+                if self.transcription_model_loaded and self.transcription_model:
+                    # TODO: Replace with actual transcription
+                    # transcription = self.transcription_model.transcribe(audio_data, language)
+                    transcription = f"[MOCK] Transcribed audio in {language}: This is a sample transcription."
+                else:
+                    transcription = f"[MOCK] Transcribed audio in {language}: This is a sample transcription."
+
+            # Use transcribed text if no text provided
+            if not text and transcription:
+                text = transcription
+
+            # Perform emotion analysis
+            emotions = []
+            confidence_scores = []
+            if text and include_emotion:
+                if self.emotion_model_loaded and self.emotion_model:
+                    # TODO: Replace with actual emotion analysis
+                    # result = self.emotion_model.analyze(text)
+                    # emotions = result['emotions']
+                    # confidence_scores = result['confidence_scores']
+                    emotions = ["joy", "sadness", "anger"]
+                    confidence_scores = [0.8, 0.6, 0.3]
+                else:
+                    emotions = ["joy", "sadness", "anger"]
+                    confidence_scores = [0.8, 0.6, 0.3]
+
+            # Perform summarization
+            summary = ""
+            if text and include_summary:
+                if self.summarization_model_loaded and self.summarization_model:
+                    # TODO: Replace with actual summarization
+                    # summary = self.summarization_model.summarize(text)
+                    summary = f"[MOCK] Summary of {len(text)} characters: {text[:50]}..."
+                else:
+                    summary = f"[MOCK] Summary of {len(text)} characters: {text[:50]}..."
+
+            processing_time = time.time() - start_time
+
+            # Determine models used
+            models_used = []
+            if include_emotion and self.emotion_model_loaded:
+                models_used.append("emotion-detection")
+            if include_summary and self.summarization_model_loaded:
+                models_used.append("t5-summarization")
+            if include_transcription and self.transcription_model_loaded:
+                models_used.append("whisper-transcription")
+
+            return {
+                "text": text,
+                "emotions": emotions,
+                "confidence_scores": confidence_scores,
+                "summary": summary,
+                "transcription": transcription,
+                "language": language,
+                "processing_time": processing_time,
+                "models_used": models_used,
+                "analysis_timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
+            }
+
+        except Exception as e:
+            logger.error(f"Complete analysis failed: {e}")
+            return {"error": "Complete analysis failed"}, 500
+
+# Register the endpoint
+api.add_resource(CompleteAnalysisEndpoint, '/')
+
+# Health check for complete analysis endpoint
+@complete_analysis_bp.route('/health', methods=['GET'])
+def health_check():
+    """Health check for complete analysis endpoint."""
+    endpoint = CompleteAnalysisEndpoint()
+    return jsonify({
+        "status": "healthy",
+        "endpoint": "complete_analysis",
+        "models_loaded": {
+            "emotion": endpoint.emotion_model_loaded,
+            "summarization": endpoint.summarization_model_loaded,
+            "transcription": endpoint.transcription_model_loaded
+        }
+    })

src/emotion_endpoint.py

@@ -0,0 +1,128 @@
+from flask import Blueprint, request, jsonify
+from flask_restx import Api, Resource, fields
+import logging
+from typing import Dict, Any, Optional
+import time
+
+logger = logging.getLogger(__name__)
+
+# Create emotion endpoint blueprint
+emotion_bp = Blueprint('emotion', __name__, url_prefix='/api/analyze')
+
+# Create API namespace
+api = Api(emotion_bp, doc=False, title='Emotion Analysis API', version='1.0')
+
+# Define request/response models
+emotion_request = api.model('EmotionRequest', {
+    'text': fields.String(required=True, description='Text to analyze for emotions'),
+    'generate_summary': fields.Boolean(required=False, default=False, description='Generate text summary')
+})
+
+emotion_response = api.model('EmotionResponse', {
+    'emotions': fields.List(fields.String, description='Detected emotions'),
+    'confidence_scores': fields.List(fields.Float, description='Confidence scores for each emotion'),
+    'summary': fields.String(description='Text summary (if requested)'),
+    'processing_time': fields.Float(description='Processing time in seconds'),
+    'text_length': fields.Integer(description='Length of input text'),
+    'timestamp': fields.String(description='Analysis timestamp')
+})
+
+@api.route('/journal')
+class EmotionAnalysis(Resource):
+    """Emotion analysis endpoint for journal entries."""
+
+    @api.expect(emotion_request)
+    @api.marshal_with(emotion_response)
+    def post(self):
+        """Analyze emotions in journal text."""
+        try:
+            start_time = time.time()
+
+            # Get request data
+            data = request.get_json()
+            if not data or 'text' not in data:
+                return {'error': 'Text is required'}, 400
+
+            text = data['text']
+            generate_summary = data.get('generate_summary', False)
+
+            # Validate input
+            if not isinstance(text, str) or len(text.strip()) == 0:
+                return {'error': 'Text must be a non-empty string'}, 400
+
+            if len(text) > 10000:  # 10k character limit
+                return {'error': 'Text too long (max 10,000 characters)'}, 400
+
+            # Mock emotion analysis (replace with actual model integration)
+            emotions, confidence_scores = self._analyze_emotions(text)
+
+            # Generate summary if requested
+            summary = None
+            if generate_summary:
+                summary = self._generate_summary(text)
+
+            processing_time = time.time() - start_time
+
+            # Prepare response
+            response = {
+                'emotions': emotions,
+                'confidence_scores': confidence_scores,
+                'summary': summary,
+                'processing_time': round(processing_time, 3),
+                'text_length': len(text),
+                'timestamp': time.strftime('%Y-%m-%d %H:%M:%S UTC', time.gmtime())
+            }
+
+            logger.info(f"Emotion analysis completed: {len(emotions)} emotions detected in {processing_time:.3f}s")
+            return response, 200
+
+        except Exception as e:
+            logger.error(f"Emotion analysis failed: {e}")
+            return {'error': 'Emotion analysis failed'}, 500
+
+    def _analyze_emotions(self, text: str) -> tuple[list[str], list[float]]:
+        """Analyze emotions in text (mock implementation)."""
+        # Mock emotion detection - replace with actual SAMO BERT model
+        emotions = []
+        confidence_scores = []
+
+        # Simple keyword-based emotion detection for demo
+        text_lower = text.lower()
+
+        emotion_keywords = {
+            'joy': ['happy', 'excited', 'joyful', 'cheerful', 'delighted'],
+            'sadness': ['sad', 'depressed', 'melancholy', 'gloomy', 'sorrowful'],
+            'anger': ['angry', 'mad', 'furious', 'irritated', 'annoyed'],
+            'fear': ['afraid', 'scared', 'terrified', 'anxious', 'worried'],
+            'surprise': ['surprised', 'shocked', 'amazed', 'astonished'],
+            'disgust': ['disgusted', 'revolted', 'repulsed', 'sickened']
+        }
+
+        for emotion, keywords in emotion_keywords.items():
+            confidence = sum(1 for keyword in keywords if keyword in text_lower) / len(keywords)
+            if confidence > 0.1:  # Threshold for detection
+                emotions.append(emotion)
+                confidence_scores.append(min(confidence * 2, 1.0))  # Scale to 0-1
+
+        # If no emotions detected, add neutral
+        if not emotions:
+            emotions = ['neutral']
+            confidence_scores = [0.5]
+
+        return emotions, confidence_scores
+
+    def _generate_summary(self, text: str) -> str:
+        """Generate text summary (mock implementation)."""
+        # Mock summarization - replace with actual T5 model
+        words = text.split()
+        if len(words) <= 20:
+            return text
+
+        # Simple extractive summary (first 20 words)
+        summary_words = words[:20]
+        return ' '.join(summary_words) + '...'
+
+def register_emotion_endpoints(app):
+    """Register emotion endpoints with the Flask app."""
+    app.register_blueprint(emotion_bp)
+    logger.info("Emotion endpoints registered: /api/analyze/journal")