from datetime import datetime, timedelta from sqlalchemy.orm import Session from sqlalchemy import func, and_ from typing import Optional, Dict import random from app.models.message import MotivationalMessage, UserMessageLog from app.models.habit import Habit, HabitCompletion from app.models.task import Task from app.models.health import HealthMetric from app.models.streak import Streak class MessageService: """Serviço para gerar mensagens contextuais inteligentes""" def __init__(self, db: Session): self.db = db def get_message_of_the_day(self, user_id: str) -> Dict: """Retorna a mensagem mais relevante para o usuário""" # 1. Verificar streak (prioridade alta) streak_message = self._check_streak_achievements(user_id) if streak_message: self._log_message(user_id, streak_message) return streak_message # 2. Verificar inatividade (recuperação) inactive_message = self._check_inactivity(user_id) if inactive_message: self._log_message(user_id, inactive_message) return inactive_message # 3. Verificar lembretes importantes reminder_message = self._check_reminders(user_id) if reminder_message: self._log_message(user_id, reminder_message) return reminder_message # 4. Verificar progresso e milestones progress_message = self._check_progress(user_id) if progress_message: self._log_message(user_id, progress_message) return progress_message # 5. Mensagem baseada em hora do dia time_message = self._get_time_based_message() if time_message: self._log_message(user_id, time_message) return time_message # 6. Mensagem motivacional padrão default_message = self._get_daily_motivation() self._log_message(user_id, default_message) return default_message def _check_streak_achievements(self, user_id: str) -> Optional[Dict]: """Verifica conquistas de streak""" # Buscar maior streak do usuário max_streak = self.db.query(func.max(Streak.current_streak)).filter( Streak.user_id == user_id ).scalar() or 0 # Definir condições de streak conditions = [ (100, 'streak_100'), (60, 'streak_60'), (30, 'streak_30'), (14, 'streak_14'), (7, 'streak_7') ] for days, condition in conditions: if max_streak >= days: # Verificar se já mostrou essa mensagem hoje already_shown = self.db.query(UserMessageLog).filter( and_( UserMessageLog.user_id == user_id, UserMessageLog.message_type == 'streak_achievement', func.date(UserMessageLog.shown_at) == datetime.now().date() ) ).first() if not already_shown: message = self.db.query(MotivationalMessage).filter( and_( MotivationalMessage.message_type == 'streak_achievement', MotivationalMessage.trigger_condition == condition, MotivationalMessage.is_active == True ) ).first() if message: return self._message_to_dict(message) return None def _check_inactivity(self, user_id: str) -> Optional[Dict]: """Verifica se o usuário está inativo""" last_log = self.db.query(UserMessageLog).filter( UserMessageLog.user_id == user_id ).order_by(UserMessageLog.shown_at.desc()).first() if not last_log: return None days_inactive = (datetime.now() - last_log.shown_at).days if days_inactive >= 14: condition = 'inactive_14days' elif days_inactive >= 7: condition = 'inactive_7days' else: return None message = self.db.query(MotivationalMessage).filter( and_( MotivationalMessage.message_type == 'recovery', MotivationalMessage.trigger_condition == condition, MotivationalMessage.is_active == True ) ).first() return self._message_to_dict(message) if message else None def _check_reminders(self, user_id: str) -> Optional[Dict]: """Verifica lembretes importantes""" # Verificar última medição de saúde last_metric = self.db.query(HealthMetric).filter( HealthMetric.user_id == user_id ).order_by(HealthMetric.measurement_date.desc()).first() if last_metric: days_since_last = (datetime.now().date() - last_metric.measurement_date).days if days_since_last >= 7: condition = 'no_weight_7days' elif days_since_last >= 3: condition = 'no_weight_3days' else: condition = None if condition: message = self.db.query(MotivationalMessage).filter( and_( MotivationalMessage.message_type == 'reminder', MotivationalMessage.trigger_condition == condition, MotivationalMessage.is_active == True ) ).first() if message: return self._message_to_dict(message) # Verificar hábitos pendentes pending_habits = self.db.query(Habit).filter( and_( Habit.user_id == user_id, Habit.is_active == True, ~Habit.id.in_( self.db.query(HabitCompletion.habit_id).filter( and_( HabitCompletion.user_id == user_id, HabitCompletion.completion_date == datetime.now().date() ) ) ) ) ).count() if pending_habits > 0: message = self.db.query(MotivationalMessage).filter( and_( MotivationalMessage.message_type == 'reminder', MotivationalMessage.trigger_condition == 'habits_pending', MotivationalMessage.is_active == True ) ).first() if message: msg_dict = self._message_to_dict(message) msg_dict['message_text'] = f"🎯 Você tem {pending_habits} hábitos esperando por você hoje!" return msg_dict return None def _check_progress(self, user_id: str) -> Optional[Dict]: """Verifica progresso e milestones""" # Calcular consistência do mês first_day_month = datetime.now().replace(day=1).date() total_habits_month = self.db.query(func.count(Habit.id)).filter( and_( Habit.user_id == user_id, Habit.start_date <= datetime.now().date() ) ).scalar() completed_habits_month = self.db.query(func.count(HabitCompletion.id)).filter( and_( HabitCompletion.user_id == user_id, HabitCompletion.completion_date >= first_day_month ) ).scalar() if total_habits_month > 0: consistency = (completed_habits_month / total_habits_month) * 100 if consistency >= 80: message = self.db.query(MotivationalMessage).filter( and_( MotivationalMessage.message_type == 'progress_milestone', MotivationalMessage.trigger_condition == 'consistency_80', MotivationalMessage.is_active == True ) ).first() if message: msg_dict = self._message_to_dict(message) msg_dict['message_text'] = f"💎 {int(consistency)}% de consistência esse mês! Top 10% dos usuários!" return msg_dict return None def _get_time_based_message(self) -> Optional[Dict]: """Retorna mensagem baseada na hora do dia""" hour = datetime.now().hour if 5 <= hour < 12: condition = 'morning' elif 12 <= hour < 18: condition = 'afternoon' elif 18 <= hour < 22: condition = 'evening' else: return None message = self.db.query(MotivationalMessage).filter( and_( MotivationalMessage.message_type == 'time_based', MotivationalMessage.trigger_condition == condition, MotivationalMessage.is_active == True ) ).first() return self._message_to_dict(message) if message else None def _get_daily_motivation(self) -> Dict: """Retorna uma mensagem motivacional aleatória""" messages = self.db.query(MotivationalMessage).filter( and_( MotivationalMessage.message_type == 'daily_motivation', MotivationalMessage.is_active == True ) ).all() if messages: message = random.choice(messages) return self._message_to_dict(message) # Fallback return { 'id': None, 'message_text': '💪 A transformação acontece um dia de cada vez. Esse dia é hoje!', 'icon': '💪', 'message_type': 'daily_motivation', 'priority': 1 } def _message_to_dict(self, message: MotivationalMessage) -> Dict: """Converte mensagem para dicionário""" return { 'id': str(message.id) if message.id else None, 'message_text': message.message_text, 'icon': message.icon, 'message_type': message.message_type, 'priority': message.priority } def _log_message(self, user_id: str, message: Dict): """Registra mensagem exibida no histórico""" log = UserMessageLog( user_id=user_id, message_id=message.get('id'), message_text=message['message_text'], message_type=message['message_type'] ) self.db.add(log) self.db.commit()