Add Flask app and sentiment analysis functionality with tests

requirements.txt

@@ -14,6 +14,11 @@ psycopg2-binary==2.9.9
 APScheduler==3.10.4
 google-cloud-bigquery==3.25.0
 Flask==3.0.0
+
+textblob==0.17.1
+nltk==3.8.1
+
 reportlab==4.0.7
 qrcode==7.4.2
-Pillow==10.1.0
+Pillow==10.1.0
+

src/app.py

@@ -0,0 +1,92 @@
+from flask import Flask, request, jsonify
+from textblob import TextBlob
+
+app = Flask(__name__)
+
+def analyze_sentiment(text):
+    """
+    Analyze sentiment of text using TextBlob.
+
+    Args:
+        text (str): The review text to analyze
+
+    Returns:
+        dict: Contains sentiment label and polarity score
+    """
+    if not text or not text.strip():
+        return {
+            'sentiment': 'neutral',
+            'polarity': 0.0,
+            'message': 'Empty text provided'
+        }
+
+    # Create TextBlob object and get polarity
+    blob = TextBlob(text)
+    polarity = blob.sentiment.polarity
+
+    # Determine sentiment based on polarity score
+    if polarity > 0.1:
+        sentiment = 'positive'
+    elif polarity < -0.1:
+        sentiment = 'negative'
+    else:
+        sentiment = 'neutral'
+
+    return {
+        'sentiment': sentiment,
+        'polarity': round(polarity, 3),
+        'subjectivity': round(blob.sentiment.subjectivity, 3)
+    }
+
+@app.route('/analyze-review', methods=['POST'])
+def analyze_review():
+    """
+    Endpoint to analyze sentiment of event reviews.
+
+    Expected JSON payload:
+        {
+            "text": "The event was amazing and well organized!"
+        }
+
+    Returns:
+        JSON response with sentiment analysis results
+    """
+    try:
+        # Get JSON data from request
+        data = request.get_json()
+
+        if not data:
+            return jsonify({
+                'error': 'No JSON data provided'
+            }), 400
+
+        # Extract text from request
+        text = data.get('text', '')
+
+        if not text:
+            return jsonify({
+                'error': 'No text field provided in request'
+            }), 400
+
+        # Analyze sentiment
+        result = analyze_sentiment(text)
+
+        return jsonify({
+            'success': True,
+            'text': text,
+            'analysis': result
+        }), 200
+
+    except Exception as e:
+        return jsonify({
+            'success': False,
+            'error': str(e)
+        }), 500
+
+@app.route('/health', methods=['GET'])
+def health_check():
+    """Health check endpoint"""
+    return jsonify({'status': 'healthy'}), 200
+
+if __name__ == '__main__':
+    app.run(debug=True, port=5000)

tests/test_app.py

@@ -0,0 +1,63 @@
+import pytest
+import json
+from app import app, analyze_sentiment
+
+@pytest.fixture
+def client():
+    """Create a test client for the Flask app."""
+    with app.test_client() as client:
+        yield client
+
+
+# ---------- Unit tests for analyze_sentiment ----------
+def test_analyze_sentiment_positive():
+    result = analyze_sentiment("This event was fantastic and enjoyable!")
+    assert result['sentiment'] == 'positive'
+    assert result['polarity'] > 0
+
+def test_analyze_sentiment_negative():
+    result = analyze_sentiment("This was the worst event ever, terrible experience.")
+    assert result['sentiment'] == 'negative'
+    assert result['polarity'] < 0
+
+def test_analyze_sentiment_neutral():
+    result = analyze_sentiment("The event happened.")
+    assert result['sentiment'] == 'neutral'
+    assert result['polarity'] == 0.0 or abs(result['polarity']) <= 0.1
+
+def test_analyze_sentiment_empty_text():
+    result = analyze_sentiment("   ")
+    assert result['sentiment'] == 'neutral'
+    assert result['polarity'] == 0.0
+    assert result['message'] == 'Empty text provided'
+
+
+# ---------- Integration tests for Flask endpoints ----------
+def test_health_check(client):
+    response = client.get('/health')
+    data = response.get_json()
+    assert response.status_code == 200
+    assert data['status'] == 'healthy'
+
+
+def test_analyze_review_valid(client):
+    payload = {"text": "I really loved this event, it was well organized!"}
+    response = client.post('/analyze-review', json=payload)
+    data = response.get_json()
+    assert response.status_code == 200
+    assert data['success'] is True
+    assert data['analysis']['sentiment'] == 'positive'
+
+
+def test_analyze_review_no_json(client):
+    response = client.post('/analyze-review', data="not json")
+    data = response.get_json()
+    assert response.status_code == 400
+    assert 'error' in data
+
+
+def test_analyze_review_missing_text_field(client):
+    response = client.post('/analyze-review', json={})
+    data = response.get_json()
+    assert response.status_code == 400
+    assert 'error' in data