"""
def process_email_queue(self) -> List[Dict]:
"""Process queued emails (simulation)"""
processed = []
for email in self.email_queue:
email['status'] = 'sent'
email['sent_at'] = datetime.now().isoformat()
processed.append(email)
self.sent_emails.append(email)
self.email_queue.clear()
return processed
def get_email_statistics(self, user_email: str) -> Dict:
"""Get email statistics for user"""
user_emails = [e for e in self.sent_emails if e['to'] == user_email]
return {
'total_sent': len(user_emails),
'by_template': {},
'recent_emails': user_emails[-5:],
'engagement_rate': 0.85 # Simulated
}
class WeeklyRecapService:
"""Enhanced weekly recap generation service"""
def __init__(self, analytics_service: AnalyticsService):
self.analytics = analytics_service
self.achievement_system = AchievementSystem()
def generate_weekly_recap(self, user_id: str) -> Dict:
"""Generate comprehensive weekly recap with achievements"""
end_date = datetime.now()
start_date = end_date - timedelta(days=7)
# Get user's emotion data for the week
user_emotions = [
e for e in self.analytics.emotion_data
if e['user_id'] == user_id and
start_date <= datetime.fromisoformat(e['timestamp']) <= end_date
]
if not user_emotions:
return {
"message": "No emotion data available for this week",
"suggestion": "Upload some images to start tracking your emotions!"
}
# Calculate comprehensive statistics
emotion_summary = self.calculate_emotion_summary(user_emotions)
mood_patterns = self.analyze_mood_patterns(user_emotions)
recommendations = self.generate_recommendations(emotion_summary)
achievements = self.achievement_system.check_achievements(user_id, user_emotions)
insights = self.generate_insights(emotion_summary, mood_patterns)
# Calculate improvement metrics
previous_week_emotions = self.get_previous_week_data(user_id, start_date)
improvement_metrics = self.calculate_improvement_metrics(user_emotions, previous_week_emotions)
return {
'week_start': start_date.strftime('%Y-%m-%d'),
'week_end': end_date.strftime('%Y-%m-%d'),
'total_detections': len(user_emotions),
'emotion_summary': emotion_summary,
'mood_patterns': mood_patterns,
'recommendations': recommendations,
'insights': insights,
'achievements': achievements,
'improvement_metrics': improvement_metrics,
'engagement_score': self.calculate_engagement_score(user_emotions),
'next_week_goals': self.suggest_next_week_goals(emotion_summary)
}
def calculate_emotion_summary(self, emotion_data: List[Dict]) -> Dict:
"""Calculate comprehensive emotion summary statistics"""
emotion_totals = {}
confidence_totals = {}
daily_emotions = {}
for record in emotion_data:
# Daily grouping
date = record['timestamp'][:10]
if date not in daily_emotions:
daily_emotions[date] = []
daily_emotions[date].append(record['emotions'])
# Overall totals
for emotion, confidence in record['emotions'].items():
if emotion not in emotion_totals:
emotion_totals[emotion] = []
confidence_totals[emotion] = []
emotion_totals[emotion].append(confidence)
confidence_totals[emotion].append(confidence)
# Calculate averages and statistics
emotion_averages = {}
emotion_stability = {}
for emotion, confidences in emotion_totals.items():
emotion_averages[emotion] = sum(confidences) / len(confidences)
emotion_stability[emotion] = np.std(confidences) if len(confidences) > 1 else 0
# Find dominant emotion and calculate diversity
dominant_emotion = max(emotion_averages, key=emotion_averages.get) if emotion_averages else None
emotion_diversity = len([e for e in emotion_averages.values() if e > 0.1]) # Emotions with >10% presence
return {
'averages': emotion_averages,
'stability': emotion_stability,
'dominant_emotion': dominant_emotion,
'emotion_distribution': emotion_totals,
'daily_breakdown': daily_emotions,
'emotion_diversity': emotion_diversity,
'consistency_score': 1 - (sum(emotion_stability.values()) / len(emotion_stability)) if emotion_stability else 0
}
def analyze_mood_patterns(self, emotion_data: List[Dict]) -> Dict:
"""Analyze detailed mood patterns throughout the week"""
daily_moods = {}
hourly_patterns = {}
mood_transitions = []
for i, record in enumerate(emotion_data):
timestamp = datetime.fromisoformat(record['timestamp'])
date = timestamp.strftime('%Y-%m-%d')
hour = timestamp.hour
# Daily patterns
if date not in daily_moods:
daily_moods[date] = []
daily_moods[date].append(record['emotions'])
# Hourly patterns
if hour not in hourly_patterns:
hourly_patterns[hour] = []
hourly_patterns[hour].append(record['emotions'])
# Mood transitions
if i > 0:
prev_dominant = max(emotion_data[i-1]['emotions'], key=emotion_data[i-1]['emotions'].get)
curr_dominant = max(record['emotions'], key=record['emotions'].get)
if prev_dominant != curr_dominant:
mood_transitions.append({
'from': prev_dominant,
'to': curr_dominant,
'timestamp': record['timestamp']
})
# Analyze patterns
most_stable_day = self.find_most_stable_day(daily_moods)
most_active_hour = self.find_most_active_hour(hourly_patterns)
return {
'daily_patterns': daily_moods,
'hourly_patterns': hourly_patterns,
'mood_transitions': mood_transitions,
'most_stable_day': most_stable_day,
'most_active_hour': most_active_hour,
'pattern_insights': self.generate_pattern_insights(daily_moods, hourly_patterns)
}
def generate_recommendations(self, emotion_summary: Dict) -> List[Dict]:
"""Generate personalized recommendations based on emotions"""
recommendations = []
dominant_emotion = emotion_summary.get('dominant_emotion')
emotion_diversity = emotion_summary.get('emotion_diversity', 0)
consistency_score = emotion_summary.get('consistency_score', 0)
# Emotion-specific recommendations
emotion_recommendations = {
'happy': [
{
'type': 'activity',
'title': 'Share Your Joy',
'description': 'Your happiness is contagious! Consider sharing positive moments with friends and family.',
'action': 'Share a happy moment on social media or call a friend'
},
{
'type': 'music',
'title': 'Upbeat Playlist',
'description': 'Keep the positive energy flowing with upbeat music.',
'action': 'Create an energetic playlist'
}
],
'sad': [
{
'type': 'wellness',
'title': 'Self-Care Time',
'description': 'It\'s important to acknowledge and process sad feelings.',
'action': 'Try meditation, journaling, or gentle exercise'
},
{
'type': 'music',
'title': 'Comforting Music',
'description': 'Soothing music can help process emotions and provide comfort.',
'action': 'Listen to calming or acoustic music'
}
],
'angry': [
{
'type': 'wellness',
'title': 'Channel Your Energy',
'description': 'Anger can be transformed into productive energy.',
'action': 'Try physical exercise or creative expression'
},
{
'type': 'mindfulness',
'title': 'Breathing Exercises',
'description': 'Deep breathing can help regulate intense emotions.',
'action': 'Practice 4-7-8 breathing technique'
}
],
'neutral': [
{
'type': 'exploration',
'title': 'Explore New Experiences',
'description': 'Neutral emotions are a great time to try new things.',
'action': 'Try a new hobby or listen to different music genres'
}
]
}
if dominant_emotion in emotion_recommendations:
recommendations.extend(emotion_recommendations[dominant_emotion])
# Diversity-based recommendations
if emotion_diversity < 3:
recommendations.append({
'type': 'growth',
'title': 'Emotional Range Expansion',
'description': 'Consider activities that might evoke different emotions to increase emotional awareness.',
'action': 'Try new experiences, watch different types of movies, or explore various music genres'
})
# Consistency-based recommendations
if consistency_score < 0.5:
recommendations.append({
'type': 'stability',
'title': 'Emotional Stability',
'description': 'Your emotions show some variability. This is normal, but consider stress management techniques.',
'action': 'Establish daily routines, practice mindfulness, or consider talking to someone'
})
return recommendations
def generate_insights(self, emotion_summary: Dict, mood_patterns: Dict) -> List[str]:
"""Generate personalized insights"""
insights = []
# Emotion diversity insights
diversity = emotion_summary.get('emotion_diversity', 0)
if diversity >= 5:
insights.append("🌈 You experienced a rich range of emotions this week, showing great emotional awareness and depth.")
elif diversity >= 3:
insights.append("😊 You had a good emotional range this week, experiencing several different feelings.")
else:
insights.append("🎯 Your emotions were quite focused this week. Consider exploring activities that might broaden your emotional experiences.")
# Stability insights
consistency = emotion_summary.get('consistency_score', 0)
if consistency > 0.7:
insights.append("⚖️ Your emotions were quite stable this week, showing good emotional regulation.")
elif consistency > 0.4:
insights.append("📊 Your emotional patterns showed moderate stability with some natural variation.")
else:
insights.append("🌊 Your emotions varied quite a bit this week. This can be normal during times of change or stress.")
# Pattern insights
transitions = mood_patterns.get('mood_transitions', [])
if len(transitions) > 5:
insights.append("🔄 You experienced several mood transitions this week, indicating dynamic emotional processing.")
# Time-based insights
most_active_hour = mood_patterns.get('most_active_hour')
if most_active_hour is not None:
if 6 <= most_active_hour <= 12:
insights.append("🌅 You tend to be most emotionally active in the morning hours.")
elif 12 <= most_active_hour <= 18:
insights.append("☀️ Your peak emotional activity occurs during afternoon hours.")
else:
insights.append("🌙 You're most emotionally active during evening hours.")
return insights
def get_previous_week_data(self, user_id: str, current_week_start: datetime) -> List[Dict]:
"""Get previous week's emotion data for comparison"""
prev_week_end = current_week_start - timedelta(days=1)
prev_week_start = prev_week_end - timedelta(days=7)
return [
e for e in self.analytics.emotion_data
if e['user_id'] == user_id and
prev_week_start <= datetime.fromisoformat(e['timestamp']) <= prev_week_end
]
def calculate_improvement_metrics(self, current_week: List[Dict], previous_week: List[Dict]) -> Dict:
"""Calculate improvement metrics compared to previous week"""
if not previous_week:
return {'message': 'No previous week data for comparison'}
# Calculate average happiness and overall positivity
def calculate_positivity(emotion_data):
if not emotion_data:
return 0
positive_emotions = ['happy', 'surprised', 'neutral']
total_positivity = 0
count = 0
for record in emotion_data:
for emotion, confidence in record['emotions'].items():
if emotion in positive_emotions:
total_positivity += confidence
count += 1
return total_positivity / count if count > 0 else 0
current_positivity = calculate_positivity(current_week)
previous_positivity = calculate_positivity(previous_week)
positivity_change = current_positivity - previous_positivity
return {
'positivity_score': current_positivity,
'positivity_change': positivity_change,
'improvement_direction': 'improving' if positivity_change > 0.05 else 'stable' if abs(positivity_change) <= 0.05 else 'declining',
'activity_level': len(current_week) - len(previous_week),
'consistency_improvement': self.calculate_consistency_improvement(current_week, previous_week)
}
def calculate_engagement_score(self, emotion_data: List[Dict]) -> float:
"""Calculate user engagement score based on activity"""
if not emotion_data:
return 0.0
# Factors: frequency, consistency, time spread
frequency_score = min(len(emotion_data) / 10, 1.0) # Max score at 10+ detections
# Time spread (how many different days)
dates = set([record['timestamp'][:10] for record in emotion_data])
spread_score = min(len(dates) / 7, 1.0) # Max score for all 7 days
# Average confidence (quality of detections)
avg_confidence = sum([max(record['emotions'].values()) for record in emotion_data]) / len(emotion_data)
confidence_score = avg_confidence
return (frequency_score + spread_score + confidence_score) / 3
def suggest_next_week_goals(self, emotion_summary: Dict) -> List[Dict]:
"""Suggest goals for the next week"""
goals = []
dominant_emotion = emotion_summary.get('dominant_emotion')
diversity = emotion_summary.get('emotion_diversity', 0)
# Always suggest consistency goal
goals.append({
'type': 'consistency',
'title': 'Daily Check-ins',
'description': 'Upload at least one image per day to track your emotional journey',
'target': 7,
'current': 0
})
# Emotion-specific goals
if dominant_emotion == 'sad':
goals.append({
'type': 'wellness',
'title': 'Self-Care Activities',
'description': 'Engage in 3 self-care activities this week',
'target': 3,
'current': 0
})
elif dominant_emotion == 'happy':
goals.append({
'type': 'sharing',
'title': 'Spread Joy',
'description': 'Share your positive moments with others',
'target': 2,
'current': 0
})
# Diversity goal
if diversity < 4:
goals.append({
'type': 'exploration',
'title': 'Emotional Exploration',
'description': 'Try activities that evoke different emotions',
'target': 5,
'current': diversity
})
return goals
def find_most_stable_day(self, daily_moods: Dict) -> str:
"""Find the day with most stable emotions"""
if not daily_moods:
return None
stability_scores = {}
for date, moods in daily_moods.items():
if len(moods) > 1:
# Calculate variance in dominant emotions
dominant_emotions = [max(mood, key=mood.get) for mood in moods]
stability = len(set(dominant_emotions)) / len(dominant_emotions)
stability_scores[date] = 1 - stability # Higher score = more stable
else:
stability_scores[date] = 1.0 # Single mood = perfectly stable
return max(stability_scores, key=stability_scores.get) if stability_scores else None
def find_most_active_hour(self, hourly_patterns: Dict) -> int:
"""Find the hour with most emotional activity"""
if not hourly_patterns:
return None
activity_scores = {hour: len(moods) for hour, moods in hourly_patterns.items()}
return max(activity_scores, key=activity_scores.get) if activity_scores else None
def generate_pattern_insights(self, daily_patterns: Dict, hourly_patterns: Dict) -> List[str]:
"""Generate insights about patterns"""
insights = []
# Daily pattern insights
if len(daily_patterns) >= 5:
insights.append("You maintained consistent emotional tracking throughout most of the week.")
elif len(daily_patterns) >= 3:
insights.append("You tracked emotions on several days this week.")
# Hourly pattern insights
if hourly_patterns:
peak_hours = sorted(hourly_patterns.keys(), key=lambda h: len(hourly_patterns[h]), reverse=True)[:3]
if peak_hours:
insights.append(f"Your most active emotional periods were around {peak_hours[0]}:00.")
return insights
def calculate_consistency_improvement(self, current_week: List[Dict], previous_week: List[Dict]) -> float:
"""Calculate improvement in emotional consistency"""
def calculate_consistency(emotion_data):
if len(emotion_data) < 2:
return 0
# Calculate variance in dominant emotions
dominant_emotions = []
for record in emotion_data:
dominant = max(record['emotions'], key=record['emotions'].get)
dominant_emotions.append(dominant)
# Consistency = how often the same emotion appears
from collections import Counter
emotion_counts = Counter(dominant_emotions)
max_count = max(emotion_counts.values())
return max_count / len(dominant_emotions)
current_consistency = calculate_consistency(current_week)
previous_consistency = calculate_consistency(previous_week)
return current_consistency - previous_consistency
class AchievementSystem:
"""Achievement and milestone tracking system"""
def __init__(self):
self.achievements = {
'first_emotion': {
'title': 'First Steps',
'description': 'Detected your first emotion',
'icon': '🎯',
'type': 'milestone',
'requirement': 1
},
'emotion_explorer': {
'title': 'Emotion Explorer',
'description': 'Detected 5 different emotions',
'icon': '🌈',
'type': 'diversity',
'requirement': 5
},
'consistent_tracker': {
'title': 'Consistent Tracker',
'description': 'Tracked emotions for 7 consecutive days',
'icon': '📅',
'type': 'consistency',
'requirement': 7
},
'emotion_master': {
'title': 'Emotion Master',
'description': 'Completed 50 emotion detections',
'icon': '🏆',
'type': 'volume',
'requirement': 50
},
'mood_analyst': {
'title': 'Mood Analyst',
'description': 'Maintained emotional awareness for 30 days',
'icon': '📊',
'type': 'longevity',
'requirement': 30
},
'happiness_champion': {
'title': 'Happiness Champion',
'description': 'Detected happiness as dominant emotion 10 times',
'icon': '😊',
'type': 'emotion_specific',
'requirement': 10,
'emotion': 'happy'
}
}
self.user_achievements = {}
def check_achievements(self, user_id: str, emotion_data: List[Dict]) -> List[Dict]:
"""Check and award achievements based on emotion data"""
if user_id not in self.user_achievements:
self.user_achievements[user_id] = []
new_achievements = []
user_earned = [ach['id'] for ach in self.user_achievements[user_id]]
for achievement_id, achievement in self.achievements.items():
if achievement_id in user_earned:
continue
if self.check_achievement_requirement(achievement, emotion_data, user_id):
new_achievement = {
'id': achievement_id,
'title': achievement['title'],
'description': achievement['description'],
'icon': achievement['icon'],
'earned_at': datetime.now().isoformat(),
'type': achievement['type']
}
self.user_achievements[user_id].append(new_achievement)
new_achievements.append(new_achievement)
return new_achievements
def check_achievement_requirement(self, achievement: Dict, emotion_data: List[Dict], user_id: str) -> bool:
"""Check if achievement requirement is met"""
achievement_type = achievement['type']
requirement = achievement['requirement']
if achievement_type == 'milestone':
return len(emotion_data) >= requirement
elif achievement_type == 'diversity':
unique_emotions = set()
for record in emotion_data:
dominant = max(record['emotions'], key=record['emotions'].get)
unique_emotions.add(dominant)
return len(unique_emotions) >= requirement
elif achievement_type == 'consistency':
if len(emotion_data) < requirement:
return False
# Check for consecutive days
dates = [datetime.fromisoformat(record['timestamp']).date() for record in emotion_data]
dates = sorted(set(dates))
consecutive_days = 1
max_consecutive = 1
for i in range(1, len(dates)):
if (dates[i] - dates[i-1]).days == 1:
consecutive_days += 1
max_consecutive = max(max_consecutive, consecutive_days)
else:
consecutive_days = 1
return max_consecutive >= requirement
elif achievement_type == 'volume':
return len(emotion_data) >= requirement
elif achievement_type == 'longevity':
if not emotion_data:
return False
first_date = min([datetime.fromisoformat(record['timestamp']).date() for record in emotion_data])
last_date = max([datetime.fromisoformat(record['timestamp']).date() for record in emotion_data])
return (last_date - first_date).days >= requirement
elif achievement_type == 'emotion_specific':
target_emotion = achievement.get('emotion')
count = 0
for record in emotion_data:
dominant = max(record['emotions'], key=record['emotions'].get)
if dominant == target_emotion:
count += 1
return count >= requirement
return False
def get_user_achievements(self, user_id: str) -> List[Dict]:
"""Get all achievements for a user"""
return self.user_achievements.get(user_id, [])
def get_achievement_progress(self, user_id: str, emotion_data: List[Dict]) -> Dict:
"""Get progress towards unearned achievements"""
if user_id not in self.user_achievements:
self.user_achievements[user_id] = []
user_earned = [ach['id'] for ach in self.user_achievements[user_id]]
progress = {}
for achievement_id, achievement in self.achievements.items():
if achievement_id in user_earned:
continue
current_progress = self.calculate_achievement_progress(achievement, emotion_data)
progress[achievement_id] = {
'title': achievement['title'],
'description': achievement['description'],
'icon': achievement['icon'],
'current': current_progress,
'required': achievement['requirement'],
'percentage': min((current_progress / achievement['requirement']) * 100, 100)
}
return progress
def calculate_achievement_progress(self, achievement: Dict, emotion_data: List[Dict]) -> int:
"""Calculate current progress towards an achievement"""
achievement_type = achievement['type']
if achievement_type == 'milestone' or achievement_type == 'volume':
return len(emotion_data)
elif achievement_type == 'diversity':
unique_emotions = set()
for record in emotion_data:
dominant = max(record['emotions'], key=record['emotions'].get)
unique_emotions.add(dominant)
return len(unique_emotions)
elif achievement_type == 'consistency':
dates = [datetime.fromisoformat(record['timestamp']).date() for record in emotion_data]
dates = sorted(set(dates))
consecutive_days = 1
max_consecutive = 1
for i in range(1, len(dates)):
if (dates[i] - dates[i-1]).days == 1:
consecutive_days += 1
max_consecutive = max(max_consecutive, consecutive_days)
else:
consecutive_days = 1
return max_consecutive
elif achievement_type == 'longevity':
if not emotion_data:
return 0
first_date = min([datetime.fromisoformat(record['timestamp']).date() for record in emotion_data])
last_date = max([datetime.fromisoformat(record['timestamp']).date() for record in emotion_data])
return (last_date - first_date).days
elif achievement_type == 'emotion_specific':
target_emotion = achievement.get('emotion')
count = 0
for record in emotion_data:
dominant = max(record['emotions'], key=record['emotions'].get)
if dominant == target_emotion:
count += 1
return count
return 0
class YouTubeService:
"""YouTube music integration service (uses backend data)"""
def __init__(self):
self.playlists = {}
self.user_preferences = {}
self.listening_history = {}
def create_mood_playlist(self, user_id: str, mood: str, tracks: List[Dict]) -> Dict:
"""Create a mood-based playlist from backend recommendations"""
playlist_id = str(uuid.uuid4())
playlist = {
'id': playlist_id,
'name': f"{mood.title()} Vibes",
'mood': mood,
'tracks': tracks[:20], # Limit to 20 tracks
'created_at': datetime.now().isoformat(),
'user_id': user_id,
'play_count': 0,
'last_played': None
}
if user_id not in self.playlists:
self.playlists[user_id] = []
self.playlists[user_id].append(playlist)
return playlist
def get_user_playlists(self, user_id: str) -> List[Dict]:
"""Get all playlists for a user"""
return self.playlists.get(user_id, [])
def play_track(self, user_id: str, track: Dict) -> Dict:
"""Simulate playing a track and update history"""
if user_id not in self.listening_history:
self.listening_history[user_id] = []
play_record = {
'track': track,
'played_at': datetime.now().isoformat(),
'duration': track.get('duration', 180) # Default 3 minutes
}
self.listening_history[user_id].append(play_record)
return play_record
def get_listening_history(self, user_id: str) -> List[Dict]:
"""Get user's listening history"""
return self.listening_history.get(user_id, [])
class UtilsService:
"""Utility functions for the application"""
@staticmethod
def format_date(date_string: str, format_type: str = 'friendly') -> str:
"""Format date string in various formats"""
try:
dt = datetime.fromisoformat(date_string.replace('Z', '+00:00'))
if format_type == 'friendly':
return dt.strftime('%B %d, %Y at %I:%M %p')
elif format_type == 'short':
return dt.strftime('%m/%d/%Y')
elif format_type == 'time':
return dt.strftime('%I:%M %p')
else:
return dt.strftime('%Y-%m-%d %H:%M:%S')
except:
return date_string
@staticmethod
def calculate_time_ago(date_string: str) -> str:
"""Calculate human-readable time ago"""
try:
dt = datetime.fromisoformat(date_string.replace('Z', '+00:00'))
now = datetime.now()
diff = now - dt
if diff.days > 0:
return f"{diff.days} day{'s' if diff.days != 1 else ''} ago"
elif diff.seconds > 3600:
hours = diff.seconds // 3600
return f"{hours} hour{'s' if hours != 1 else ''} ago"
elif diff.seconds > 60:
minutes = diff.seconds // 60
return f"{minutes} minute{'s' if minutes != 1 else ''} ago"
else:
return "Just now"
except:
return "Unknown"
@staticmethod
def validate_email(email: str) -> bool:
"""Validate email format"""
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return re.match(pattern, email) is not None
@staticmethod
def validate_password(password: str) -> Dict:
"""Validate password strength"""
errors = []
if len(password) < 8:
errors.append("Password must be at least 8 characters long")
if not re.search(r'[A-Z]', password):
errors.append("Password must contain at least one uppercase letter")
if not re.search(r'[a-z]', password):
errors.append("Password must contain at least one lowercase letter")
if not re.search(r'\d', password):
errors.append("Password must contain at least one number")
return {
'valid': len(errors) == 0,
'errors': errors,
'strength': 'strong' if len(errors) == 0 else 'weak'
}
@staticmethod
def sanitize_input(text: str) -> str:
"""Sanitize user input"""
# Remove HTML tags and dangerous characters
text = re.sub(r'<[^>]*>', '', text)
text = re.sub(r'[<>&"\'`]', '', text)
return text.strip()
@staticmethod
def generate_color_from_emotion(emotion: str) -> str:
"""Generate color based on emotion"""
emotion_colors = {
'happy': '#FFD700',
'sad': '#4169E1',
'angry': '#DC143C',
'fear': '#8B008B',
'surprise': '#FF8C00',
'disgust': '#228B22',
'neutral': '#708090',
'contempt': '#800080'
}
return emotion_colors.get(emotion.lower(), '#708090')
@staticmethod
def calculate_emotion_intensity(emotions: Dict) -> str:
"""Calculate overall emotion intensity"""
if not emotions:
return 'low'
max_confidence = max(emotions.values())
if max_confidence > 0.8:
return 'high'
elif max_confidence > 0.5:
return 'medium'
else:
return 'low'
# ============================================================================
# INITIALIZE SERVICES
# ============================================================================
# Initialize all services
analytics_service = AnalyticsService()
api_service = APIService(API_BASE_URL)
email_service = EmailService()
weekly_recap_service = WeeklyRecapService(analytics_service)
youtube_service = YouTubeService()
utils_service = UtilsService()
# Global state management
current_user = None
current_session = None
# ============================================================================
# CORE FUNCTIONS
# ============================================================================
def authenticate_user(email: str, password: str) -> Tuple[bool, str, Dict]:
"""Authenticate user with enhanced error handling"""
global current_user, current_session
try:
# Validate inputs
if not email or not password:
return False, "Email and password are required", {}
if not utils_service.validate_email(email):
return False, "Please enter a valid email address", {}
# Track authentication attempt
analytics_service.track_event('login_attempt', email, {'timestamp': datetime.now().isoformat()})
# Call backend API
result = api_service.authenticate_user(email, password)
if result['success']:
current_user = result['user']
current_session = str(uuid.uuid4())
# Track successful login
analytics_service.track_event('login_success', email, {
'session_id': current_session,
'timestamp': datetime.now().isoformat()
})
# Send welcome back notification
email_service.send_welcome_email(email, current_user.get('name', 'User'))
return True, "Login successful! Welcome back!", current_user
else:
# Track failed login
analytics_service.track_event('login_failure', email, {
'error': result['error'],
'timestamp': datetime.now().isoformat()
})
return False, result.get('error', 'Login failed'), {}
except Exception as e:
analytics_service.track_event('login_error', email, {
'error': str(e),
'timestamp': datetime.now().isoformat()
})
return False, f"Login error: {str(e)}", {}
def register_user(email: str, password: str, name: str, confirm_password: str) -> Tuple[bool, str, Dict]:
"""Register new user with comprehensive validation"""
global current_user, current_session
try:
# Validate inputs
if not all([email, password, name, confirm_password]):
return False, "All fields are required", {}
if not utils_service.validate_email(email):
return False, "Please enter a valid email address", {}
if password != confirm_password:
return False, "Passwords do not match", {}
# Validate password strength
password_validation = utils_service.validate_password(password)
if not password_validation['valid']:
return False, "; ".join(password_validation['errors']), {}
# Sanitize inputs
name = utils_service.sanitize_input(name)
email = email.lower().strip()
# Track registration attempt
analytics_service.track_event('signup_attempt', email, {'timestamp': datetime.now().isoformat()})
# Call backend API
result = api_service.register_user(email, password, name)
if result['success']:
current_user = result['user']
current_session = str(uuid.uuid4())
# Track successful registration
analytics_service.track_event('signup_success', email, {
'session_id': current_session,
'timestamp': datetime.now().isoformat()
})
# Send welcome email
email_service.send_welcome_email(email, name)
return True, f"Welcome to VibeStory, {name}! Your account has been created successfully.", current_user
else:
# Track failed registration
analytics_service.track_event('signup_failure', email, {
'error': result['error'],
'timestamp': datetime.now().isoformat()
})
return False, result.get('error', 'Registration failed'), {}
except Exception as e:
analytics_service.track_event('signup_error', email, {
'error': str(e),
'timestamp': datetime.now().isoformat()
})
return False, f"Registration error: {str(e)}", {}
def detect_emotion_from_image(image) -> Tuple[bool, str, Dict]:
"""Detect emotion from uploaded image with comprehensive processing"""
global current_user
try:
if not image:
return False, "Please upload an image", {}
if not current_user:
return False, "Please log in to analyze emotions", {}
user_id = current_user.get('id', 'anonymous')
# Track emotion detection attempt
analytics_service.track_event('emotion_detection_start', user_id, {
'timestamp': datetime.now().isoformat()
})
# Process image
start_time = time.time()
# Call backend API for emotion detection
result = api_service.detect_emotion(image)
processing_time = time.time() - start_time
if result['success']:
# Extract emotion data
emotions = result.get('emotions', {})
moods = result.get('moods', [])
genres = result.get('genres', [])
tracks = result.get('tracks', [])
source = result.get('source', 'unknown')
# Calculate metadata
image_metadata = {
'size': getattr(image, 'size', (0, 0)),
'format': getattr(image, 'format', 'unknown'),
'mode': getattr(image, 'mode', 'unknown')
}
# Enhanced emotion data with processing info
enhanced_emotions = {
**emotions,
'processing_time': processing_time,
'confidence_score': max(emotions.values()) if emotions else 0,
'detection_source': source
}
# Track emotion detection in analytics
emotion_record = analytics_service.track_emotion(
user_id,
enhanced_emotions,
image_metadata
)
# Create mood-based playlist
if tracks:
dominant_emotion = max(emotions, key=emotions.get) if emotions else 'neutral'
playlist = youtube_service.create_mood_playlist(user_id, dominant_emotion, tracks)
# Track successful detection
analytics_service.track_event('emotion_detection_success', user_id, {
'dominant_emotion': max(emotions, key=emotions.get) if emotions else 'unknown',
'confidence': max(emotions.values()) if emotions else 0,
'processing_time': processing_time,
'timestamp': datetime.now().isoformat()
})
# Check for achievements
user_emotions = [e for e in analytics_service.emotion_data if e['user_id'] == user_id]
new_achievements = weekly_recap_service.achievement_system.check_achievements(user_id, user_emotions)
# Send achievement notifications
for achievement in new_achievements:
email_service.send_achievement_notification(current_user['email'], achievement)
return True, "Emotion detected successfully!", {
'emotions': emotions,
'moods': moods,
'genres': genres,
'tracks': tracks,
'source': source,
'processing_time': processing_time,
'achievements': new_achievements,
'playlist_created': len(tracks) > 0,
'emotion_record_id': emotion_record['id']
}
else:
# Track failed detection
analytics_service.track_event('emotion_detection_failure', user_id, {
'error': result['error'],
'processing_time': processing_time,
'timestamp': datetime.now().isoformat()
})
return False, result.get('error', 'Emotion detection failed'), {}
except Exception as e:
if current_user:
analytics_service.track_event('emotion_detection_error', current_user.get('id', 'anonymous'), {
'error': str(e),
'timestamp': datetime.now().isoformat()
})
return False, f"Analysis error: {str(e)}", {}
def generate_analytics_dashboard(user_id: str = None) -> Dict:
"""Generate comprehensive analytics dashboard"""
try:
if not user_id and current_user:
user_id = current_user.get('id')
if not user_id:
return {'error': 'User not authenticated'}
# Get analytics data
dashboard_data = analytics_service.get_analytics_dashboard(user_id)
# Get achievement progress
user_emotions = [e for e in analytics_service.emotion_data if e['user_id'] == user_id]
achievement_progress = weekly_recap_service.achievement_system.get_achievement_progress(user_id, user_emotions)
earned_achievements = weekly_recap_service.achievement_system.get_user_achievements(user_id)
# Get playlists
user_playlists = youtube_service.get_user_playlists(user_id)
# Get listening history
listening_history = youtube_service.get_listening_history(user_id)
# Enhanced dashboard with all data
enhanced_dashboard = {
**dashboard_data,
'achievement_progress': achievement_progress,
'earned_achievements': earned_achievements,
'playlists': user_playlists,
'listening_history': listening_history[-10:], # Last 10 tracks
'user_stats': {
'total_playlists': len(user_playlists),
'total_listening_time': sum([track.get('duration', 0) for track in listening_history]),
'favorite_genre': 'Pop', # Would be calculated from listening history
'streak_days': dashboard_data.get('session_duration', {}).get('current_session_duration', 0) // 86400
}
}
return enhanced_dashboard
except Exception as e:
return {'error': f'Dashboard generation failed: {str(e)}'}
def generate_weekly_recap_report(user_id: str = None) -> Dict:
"""Generate comprehensive weekly recap"""
try:
if not user_id and current_user:
user_id = current_user.get('id')
if not user_id:
return {'error': 'User not authenticated'}
# Generate recap
recap_data = weekly_recap_service.generate_weekly_recap(user_id)
# Send email recap if user has email
if current_user and 'email' in current_user:
email_service.send_weekly_recap(current_user['email'], recap_data)
return recap_data
except Exception as e:
return {'error': f'Weekly recap generation failed: {str(e)}'}
def create_story_from_emotion(emotion_data: Dict, story_prompt: str = "") -> Dict:
"""Create an emotion-based story"""
try:
if not emotion_data:
return {'error': 'No emotion data provided'}
dominant_emotion = max(emotion_data, key=emotion_data.get)
confidence = emotion_data[dominant_emotion]
# Story templates based on emotions
story_templates = {
'happy': [
"The sun was shining brightly as {character} discovered something wonderful...",
"With a smile that could light up the world, {character} began their adventure...",
"Joy filled the air as {character} realized their dream was coming true..."
],
'sad': [
"The rain matched {character}'s mood as they walked through the empty streets...",
"With tears in their eyes, {character} remembered better times...",
"The melancholy evening reminded {character} of what they had lost..."
],
'angry': [
"Fury burned in {character}'s eyes as they faced the injustice...",
"The storm inside {character} was about to break free...",
"With clenched fists, {character} decided enough was enough..."
],
'surprised': [
"Never in their wildest dreams did {character} expect this...",
"The unexpected turn of events left {character} speechless...",
"As the door opened, {character} gasped in amazement..."
],
'neutral': [
"It was an ordinary day when {character} noticed something different...",
"In the quiet moments, {character} found clarity...",
"Sometimes the most profound changes begin with simple observations..."
]
}
# Select template
templates = story_templates.get(dominant_emotion, story_templates['neutral'])
selected_template = templates[0] # Could be randomized
# Generate story
character_name = story_prompt.strip() if story_prompt else "Alex"
story_beginning = selected_template.format(character=character_name)
story_data = {
'id': str(uuid.uuid4()),
'title': f"A {dominant_emotion.title()} Story",
'beginning': story_beginning,
'emotion_inspiration': dominant_emotion,
'confidence': confidence,
'character': character_name,
'created_at': datetime.now().isoformat(),
'author': current_user.get('name', 'Anonymous') if current_user else 'Anonymous'
}
# Track story creation
if current_user:
analytics_service.track_event('story_created', current_user.get('id'), {
'emotion': dominant_emotion,
'confidence': confidence,
'timestamp': datetime.now().isoformat()
})
return {'success': True, 'story': story_data}
except Exception as e:
return {'error': f'Story creation failed: {str(e)}'}
def get_backend_health() -> Dict:
"""Check backend health status"""
try:
result = api_service.get_health_status()
if result['success']:
health_data = result['data']
return {
'status': 'healthy',
'backend_connected': True,
'details': health_data
}
else:
return {
'status': 'unhealthy',
'backend_connected': False,
'error': result.get('error', 'Unknown error')
}
except Exception as e:
return {
'status': 'error',
'backend_connected': False,
'error': str(e)
}
# ============================================================================
# GRADIO INTERFACE FUNCTIONS
# ============================================================================
def create_login_interface():
"""Create enhanced login interface with WORKING backend integration"""
return f"""
🎭VibeStory
Welcome Back
Sign in to continue your emotional journey
Or continue with
Don't have an account?
"""
def create_signup_interface():
"""Create enhanced signup interface with WORKING backend integration"""
return f"""
🎭VibeStory
Join VibeStory
Create your account to start your emotional journey
Or continue with
Already have an account?
"""
# ============================================================================
# MAIN GRADIO APPLICATION
# ============================================================================
def create_main_app():
"""Create the main Gradio application with all features"""
with gr.Blocks(
title="🎭 VibeStory - AI Emotion Recognition Platform",
theme=gr.themes.Base(
primary_hue="purple",
secondary_hue="pink",
neutral_hue="slate"
),
css=CUSTOM_CSS
) as app:
# Global state
user_state = gr.State(None)
session_state = gr.State(None)
# Rest of your app code...
# Main application header
gr.HTML("""
🎭 VibeStory
AI-Powered Emotion Recognition & Music Recommendation Platform
Comprehensive analysis of your emotional journey this week
""")
with gr.Row():
generate_recap_btn = gr.Button("📊 Generate Weekly Recap", variant="primary")
email_recap_btn = gr.Button("📧 Email Recap", variant="secondary")
weekly_recap_display = gr.HTML(label="Weekly Recap")
# Weekly recap function
def generate_recap():
if not current_user:
return "Please log in to generate weekly recap"
recap_data = generate_weekly_recap_report()
if 'error' in recap_data:
return f"
{recap_data['error']}
"
return create_weekly_recap_html(recap_data)
def email_recap():
if not current_user:
return "Please log in to email recap"
recap_data = generate_weekly_recap_report()
if 'error' not in recap_data:
email_service.send_weekly_recap(current_user['email'], recap_data)
return "Weekly recap sent to your email!"
return "Failed to generate recap for email"
generate_recap_btn.click(generate_recap, outputs=[weekly_recap_display])
email_recap_btn.click(email_recap, outputs=[weekly_recap_display])
# ============================================================================
# STORY CREATION TAB
# ============================================================================
with gr.Tab("📖 Story Creation", id="story_tab"):
gr.HTML("""
Create Emotion-Based Stories
Transform your emotions into creative stories
""")
with gr.Row():
with gr.Column():
story_emotion_input = gr.Dropdown(
choices=['happy', 'sad', 'angry', 'surprised', 'neutral', 'fear'],
label="Select Emotion for Story",
value="happy"
)
story_character_input = gr.Textbox(
label="Character Name",
placeholder="Enter character name (optional)",
value="Alex"
)
create_story_btn = gr.Button("✨ Create Story", variant="primary")
with gr.Column():
story_output = gr.HTML(label="Generated Story")
# Story creation function
def create_story(emotion, character_name):
if not current_user:
return "Please log in to create stories"
# Create mock emotion data
emotion_data = {emotion: 0.8}
result = create_story_from_emotion(emotion_data, character_name)
if result.get('success'):
story = result['story']
return create_story_display_html(story)
else:
return f"
{result.get('error', 'Story creation failed')}
"
create_story_btn.click(
create_story,
inputs=[story_emotion_input, story_character_input],
outputs=[story_output]
)
# ============================================================================
# MUSIC PLAYER TAB
# ============================================================================
with gr.Tab("🎵 Music Player", id="music_tab"):
gr.HTML("""
Your Personalized Music Player
Listen to emotion-based music recommendations
""")
with gr.Row():
refresh_playlists_btn = gr.Button("🔄 Refresh Playlists", variant="secondary")
with gr.Row():
user_playlists = gr.HTML(label="Your Playlists")
listening_history = gr.HTML(label="Listening History")
# Music player functions
def load_music_player():
if not current_user:
return "Please log in to access music player", ""
user_id = current_user.get('id')
playlists = youtube_service.get_user_playlists(user_id)
history = youtube_service.get_listening_history(user_id)
playlists_html = create_playlists_html(playlists)
history_html = create_listening_history_html(history)
return playlists_html, history_html
refresh_playlists_btn.click(
load_music_player,
outputs=[user_playlists, listening_history]
)
# ============================================================================
# PROFILE TAB
# ============================================================================
with gr.Tab("👤 Profile", id="profile_tab") as profile_tab:
gr.HTML("""
Your Profile & Settings
Manage your account and preferences
""")
with gr.Row():
with gr.Column():
profile_info = gr.HTML(label="Profile Information")
linked_accounts = gr.HTML(label="Linked Accounts")
with gr.Column():
user_stats = gr.HTML(label="User Statistics")
logout_btn = gr.Button("🚪 Logout", variant="stop")
# Profile functions
def load_profile():
if not current_user:
return "Please log in to view profile", "", ""
profile_html = create_profile_info_html(current_user)
accounts_html = create_linked_accounts_html(current_user)
stats_html = create_user_stats_html(current_user)
return profile_html, accounts_html, stats_html
def handle_logout():
global current_user, current_session
current_user = None
current_session = None
return None, "Logged out successfully", gr.update(selected="login_tab")
# Load profile on tab selection
profile_tab.select(
load_profile,
outputs=[profile_info, linked_accounts, user_stats]
)
logout_btn.click(
handle_logout,
outputs=[user_state, profile_info, main_tabs]
)
# ============================================================================
# BACKEND STATUS TAB
# ============================================================================
with gr.Tab("🔧 Backend Status", id="status_tab"):
gr.HTML("""
Backend Health & Status
Monitor system health and performance
""")
with gr.Row():
check_health_btn = gr.Button("🏥 Check Backend Health", variant="primary")
test_connection_btn = gr.Button("🔌 Test Connection", variant="secondary")
backend_status_display = gr.HTML(label="Backend Status")
# Backend status functions
def check_backend_health():
health_data = get_backend_health()
return create_health_status_html(health_data)
def test_backend_connection():
try:
result = api_service.get_health_status()
if result['success']:
return """
""")
return app
# ============================================================================
# HTML CREATION HELPER FUNCTIONS
# ============================================================================
def create_emotion_results_html(emotions: Dict, moods: List[str], source: str) -> str:
"""Create HTML for emotion analysis results"""
if not emotions:
return "
No emotions detected
"
dominant_emotion = max(emotions, key=emotions.get)
confidence = emotions[dominant_emotion]
emotion_color = utils_service.generate_color_from_emotion(dominant_emotion)
emotions_html = ""
for emotion, conf in sorted(emotions.items(), key=lambda x: x[1], reverse=True):
color = utils_service.generate_color_from_emotion(emotion)
emotions_html += f"""
{emotion.title()}
{conf:.1%}
"""
moods_html = ""
for mood in moods[:5]: # Show top 5 moods
moods_html += f"""
{mood}
"""
return f"""
😊
Dominant Emotion: {dominant_emotion.title()}
Confidence: {confidence:.1%} | Source: {source}
All Detected Emotions:
{emotions_html}
Associated Moods:
{moods_html}
"""
def create_music_recommendations_html(tracks: List[Dict]) -> str:
"""Create HTML for music recommendations"""
if not tracks:
return "
No music recommendations available
"
tracks_html = ""
for i, track in enumerate(tracks[:10]): # Show top 10 tracks
tracks_html += f"""
{track.get('title', 'Unknown Title')}
by {track.get('artist', 'Unknown Artist')}
{track.get('source', 'recommendation')}
"""
return f"""
🎵 Music Recommendations
Based on your detected emotions, here are some songs you might enjoy:
{tracks_html}
💡 Tip: Click "Play" to add to your listening history, or "Open" to listen on YouTube Music
"""
def create_achievements_html(achievements: List[Dict]) -> str:
"""Create HTML for achievement notifications"""
if not achievements:
return ""
achievements_html = ""
for achievement in achievements:
achievements_html += f"""