dashboard.py

from flask import Flask, render_template, jsonify
from flask_socketio import SocketIO, emit
import json
import threading
import time
from datetime import datetime
from collections import defaultdict, deque
import logging
import os
import random

app = Flask(__name__)
app.config['SECRET_KEY'] = 'sentiment-dashboard-secret'
socketio = SocketIO(app, cors_allowed_origins="*")

# In-memory storage for dashboard data
sentiment_counts = {'positive': 0, 'negative': 0, 'neutral': 0}
recent_tweets = deque(maxlen=50)  # Keep last 50 tweets
hourly_sentiment = defaultdict(lambda: {'positive': 0, 'negative': 0, 'neutral': 0})

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Check environment for mock mode
USE_MOCK = os.environ.get("USE_MOCK", "true").lower() == "true"

def kafka_consumer_thread():
    """Background thread to consume processed tweets from Kafka or generate mock data"""
    if USE_MOCK:
        logger.info("Running in MOCK mode - generating demo data")
        mock_tweet_generator()
    else:
        logger.info("Running in KAFKA mode - connecting to real Kafka")
        real_kafka_consumer()

def mock_tweet_generator():
    """Generate mock tweets for demo purposes"""
    sentiments = ["positive", "neutral", "negative"]
    
    # Sample mock tweets for demo
    sample_tweets = [
        "I absolutely love this new Python framework! Amazing! 🐍✨",
        "Just finished my first machine learning project! So excited! 🚀",
        "Beautiful sunny day! Perfect for coding ☕️💻",
        "Finally understood how Kafka works! Awesome technology 🎉",
        "Ugh, spent 3 hours debugging this error. So frustrated 😤",
        "This API documentation is terrible. Nothing works 😡", 
        "Why is deployment always so painful? 💔",
        "Working on a new feature. Should be ready next week.",
        "Attending a tech conference tomorrow. Looking forward to it.",
        "Updated the dependencies. Everything seems fine.",
        "Django vs Flask debate continues. Both are good.",
        "Love how clean Python code can be. Beautiful language!",
        "FastAPI is becoming my go-to for REST APIs. So fast!",
        "NumPy arrays are much faster than regular lists.",
        "Jupyter notebooks are perfect for data exploration.",
    ]
    
    tweet_count = 0
    
    while True:
        try:
            # Generate a mock tweet
            sentiment = random.choice(sentiments)
            tweet_text = random.choice(sample_tweets)
            
            tweet_data = {
                'text': tweet_text,
                'sentiment': sentiment,
                'timestamp': datetime.now().strftime('%H:%M:%S'),
                'author_id': f'user_{random.randint(1000, 9999)}'
            }
            
            # Update sentiment counts
            sentiment_counts[sentiment] += 1
            
            # Add to recent tweets
            recent_tweets.append(tweet_data)
            
            # Update hourly data
            hour = datetime.now().strftime('%H:00')
            hourly_sentiment[hour][sentiment] += 1
            
            # Emit real-time update to connected clients
            socketio.emit('sentiment_update', {
                'sentiment_counts': dict(sentiment_counts),
                'recent_tweets': list(recent_tweets),
                'hourly_data': dict(hourly_sentiment)
            })
            
            tweet_count += 1
            logger.info(f"Generated mock tweet #{tweet_count} with sentiment: {sentiment}")
            
            # Random delay between tweets (1-3 seconds for demo)
            time.sleep(random.uniform(1, 3))
            
        except Exception as e:
            logger.error(f"Error in mock tweet generator: {e}")
            time.sleep(5)

def real_kafka_consumer():
    """Real Kafka consumer for production use"""
    try:
        from kafka import KafkaConsumer
        from kafka.errors import NoBrokersAvailable
        
        def create_kafka_consumer(max_retries=10, retry_delay=5):
            """Create Kafka consumer with retry logic"""
            for attempt in range(max_retries):
                try:
                    consumer = KafkaConsumer(
                        'sentiment-results',
                        bootstrap_servers=['kafka:9092'],
                        value_deserializer=lambda m: json.loads(m.decode('utf-8')),
                        consumer_timeout_ms=1000,
                        auto_offset_reset='earliest',
                        enable_auto_commit=True,
                        group_id='dashboard-group'
                    )
                    logger.info("Successfully connected to Kafka consumer!")
                    return consumer
                except NoBrokersAvailable as e:
                    logger.warning(f"Kafka not ready, attempt {attempt + 1}/{max_retries}. Retrying in {retry_delay}s...")
                    time.sleep(retry_delay)
                except Exception as e:
                    logger.error(f"Unexpected error connecting to Kafka: {e}")
                    time.sleep(retry_delay)
            
            raise Exception(f"Could not connect to Kafka consumer after {max_retries} attempts")

        # Wait for Kafka and Spark to be ready
        logger.info("Waiting for Kafka and Spark services to be ready...")
        time.sleep(10)
        
        consumer = create_kafka_consumer()
        logger.info("Connected to Kafka consumer for dashboard - waiting for processed tweets...")
        
        message_count = 0
        
        while True:
            try:
                # Poll for messages with timeout
                message_batch = consumer.poll(timeout_ms=1000)
                
                if message_batch:
                    logger.info(f"Received batch with {len(message_batch)} topic partitions")
                    
                    for topic_partition, messages in message_batch.items():
                        logger.info(f"Processing {len(messages)} messages from {topic_partition}")
                        
                        for message in messages:
                            try:
                                tweet_data = message.value
                                message_count += 1
                                logger.info(f"Message {message_count}: Received tweet data: {tweet_data}")
                                
                                # Update sentiment counts
                                sentiment = tweet_data.get('sentiment', 'neutral')
                                sentiment_counts[sentiment] += 1
                                
                                # Add to recent tweets
                                recent_tweets.append({
                                    'text': tweet_data.get('tweet_text', '')[:100] + '...' if len(tweet_data.get('tweet_text', '')) > 100 else tweet_data.get('tweet_text', ''),
                                    'sentiment': sentiment,
                                    'timestamp': datetime.now().strftime('%H:%M:%S'),
                                    'author_id': tweet_data.get('author_id', 'Unknown')
                                })
                                
                                # Update hourly data
                                hour = datetime.now().strftime('%H:00')
                                hourly_sentiment[hour][sentiment] += 1
                                
                                # Emit real-time update to connected clients
                                socketio.emit('sentiment_update', {
                                    'sentiment_counts': dict(sentiment_counts),
                                    'recent_tweets': list(recent_tweets),
                                    'hourly_data': dict(hourly_sentiment)
                                })
                                
                                logger.info(f"Processed tweet with sentiment: {sentiment} - Total counts: {dict(sentiment_counts)}")
                                
                            except Exception as e:
                                logger.error(f"Error processing individual tweet data: {e}")
                else:
                    if message_count == 0:
                        logger.info("No messages received yet, continuing to poll...")
                    time.sleep(1)
                        
            except Exception as e:
                logger.error(f"Error in polling loop: {e}")
                time.sleep(5)
                
    except ImportError:
        logger.warning("kafka-python not available, falling back to mock mode")
        mock_tweet_generator()
    except Exception as e:
        logger.error(f"Error in real Kafka consumer: {e}")
        logger.info("Falling back to mock mode")
        mock_tweet_generator()

@app.route('/')
def dashboard():
    """Main dashboard page"""
    return render_template('dashboard.html')

@app.route('/api/data')
def get_data():
    """API endpoint to get current dashboard data"""
    data = {
        'sentiment_counts': dict(sentiment_counts),
        'recent_tweets': list(recent_tweets),
        'hourly_data': dict(hourly_sentiment),
        'total_tweets': sum(sentiment_counts.values())
    }
    logger.info(f"API request - returning data: {data}")
    return jsonify(data)

@socketio.on('connect')
def handle_connect():
    """Handle client connection"""
    logger.info("Client connected to dashboard")
    emit('sentiment_update', {
        'sentiment_counts': dict(sentiment_counts),
        'recent_tweets': list(recent_tweets),
        'hourly_data': dict(hourly_sentiment)
    })

if __name__ == '__main__':
    # Start consumer thread (either mock or real Kafka)
    consumer_thread = threading.Thread(target=kafka_consumer_thread, daemon=True)
    consumer_thread.start()
    
    mode = "MOCK" if USE_MOCK else "KAFKA"
    logger.info(f"Starting sentiment dashboard in {mode} mode on port 5000")
    
    if USE_MOCK:
        logger.info("Dashboard will display mock demo data")
    else:
        logger.info("Dashboard will display data once Spark processes tweets from Kafka")
    
    # Get port from environment (Hugging Face Spaces uses port 7860)
    port = int(os.environ.get('PORT', 5000))
    
    # Fix for Werkzeug warning - use allow_unsafe_werkzeug for development
    socketio.run(app, host='0.0.0.0', port=port, debug=False, allow_unsafe_werkzeug=True)