# performance_system.py from datetime import datetime, timedelta import pandas as pd import numpy as np from typing import Dict, List, Optional, Tuple import json import logging # Configuração de logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class PerformanceConstants: """Constantes para análise de desempenho""" MINIMUM_STUDY_HOURS = 4.0 IDEAL_CONSISTENCY = 0.7 LOW_PERFORMANCE_THRESHOLD = 0.3 MEDIUM_PERFORMANCE_THRESHOLD = 0.6 MIN_DAYS_FOR_TREND = 7 MAX_DAYS_ANALYSIS = 30 class PerformanceAnalyzer: # [O código existente permanece o mesmo] def get_performance_metrics(self, user_id: str, days: int = 30) -> Dict: """Obtém métricas detalhadas de desempenho""" try: cursor = self.conn.cursor() end_date = datetime.now().date() start_date = end_date - timedelta(days=days) cursor.execute(''' SELECT date, topic, horas_estudadas, performance_score FROM study_progress WHERE user_id = ? AND date BETWEEN ? AND ? ORDER BY date ''', (user_id, start_date, end_date)) data = cursor.fetchall() metrics = { "daily_metrics": {}, "topic_metrics": {}, "overall_metrics": { "total_hours": 0, "avg_performance": 0, "study_days": 0 } } for date, topic, hours, score in data: # Métricas diárias if date not in metrics["daily_metrics"]: metrics["daily_metrics"][date] = { "hours": 0, "topics": set() } metrics["daily_metrics"][date]["hours"] += hours metrics["daily_metrics"][date]["topics"].add(topic) # Métricas por tópico if topic not in metrics["topic_metrics"]: metrics["topic_metrics"][topic] = { "total_hours": 0, "scores": [], "last_study": None } metrics["topic_metrics"][topic]["total_hours"] += hours metrics["topic_metrics"][topic]["scores"].append(score) metrics["topic_metrics"][topic]["last_study"] = date # Métricas gerais metrics["overall_metrics"]["total_hours"] += hours # Calcular médias e estatísticas if data: all_scores = [score for _, _, _, score in data] metrics["overall_metrics"]["avg_performance"] = np.mean(all_scores) metrics["overall_metrics"]["study_days"] = len(metrics["daily_metrics"]) return metrics except Exception as e: logger.error(f"Erro ao obter métricas de desempenho: {e}") return None class StudyMaterialGenerator: # [O código existente permanece o mesmo] def generate_daily_plan(self, user_id: str, available_hours: float, performance_data: Dict) -> Dict[str, any]: """Gera plano de estudos diário personalizado""" try: weak_areas = sorted( performance_data["topic_metrics"].items(), key=lambda x: np.mean(x[1]["scores"]) if x[1]["scores"] else 0 ) plan = { "distribuicao_horas": {}, "prioridades": [], "recursos_sugeridos": [] } # Distribuir horas disponíveis remaining_hours = available_hours for area, metrics in weak_areas: if remaining_hours <= 0: break # Áreas com desempenho mais baixo recebem mais tempo weight = 1 - (np.mean(metrics["scores"]) if metrics["scores"] else 0) hours_allocated = min(remaining_hours, available_hours * weight) plan["distribuicao_horas"][area] = round(hours_allocated, 1) remaining_hours -= hours_allocated # Adicionar recursos recomendados plan["recursos_sugeridos"].extend( self.get_recommended_resources(area, metrics) ) return plan except Exception as e: logger.error(f"Erro ao gerar plano diário: {e}") return None def get_recommended_resources(self, area: str, metrics: Dict) -> List[str]: """Retorna recursos recomendados baseados no desempenho""" resources = [] avg_score = np.mean(metrics["scores"]) if metrics["scores"] else 0 if avg_score < 0.3: resources.extend([ "📚 Material básico teórico", "📝 Resumos esquematizados", "🎥 Vídeo-aulas introdutórias" ]) elif avg_score < 0.6: resources.extend([ "📋 Questões comentadas", "🏥 Casos clínicos simples", "📊 Mapas mentais avançados" ]) else: resources.extend([ "🎯 Questões complexas", "🏥 Casos clínicos avançados", "📑 Artigos científicos" ]) return resources class ProgressTracker: # [O código existente permanece o mesmo] def calculate_study_streak(self, user_id: str) -> Dict[str, any]: """Calcula sequência atual de estudos""" try: cursor = self.conn.cursor() cursor.execute(''' SELECT DISTINCT date FROM study_progress WHERE user_id = ? ORDER BY date DESC ''', (user_id,)) dates = [row[0] for row in cursor.fetchall()] if not dates: return { "current_streak": 0, "longest_streak": 0, "last_study_date": None } current_streak = 1 longest_streak = 1 current_date = datetime.strptime(dates[0], '%Y-%m-%d').date() for i in range(1, len(dates)): date = datetime.strptime(dates[i], '%Y-%m-%d').date() if (current_date - date).days == 1: current_streak += 1 longest_streak = max(longest_streak, current_streak) else: break current_date = date return { "current_streak": current_streak, "longest_streak": longest_streak, "last_study_date": dates[0] } except Exception as e: logger.error(f"Erro ao calcular sequência de estudos: {e}") return None def initialize_performance_system(db_connection) -> Tuple[PerformanceAnalyzer, StudyMaterialGenerator, ProgressTracker]: """Inicializa o sistema de performance completo""" try: analyzer = PerformanceAnalyzer(db_connection) material_gen = StudyMaterialGenerator(db_connection) tracker = ProgressTracker(db_connection) return analyzer, material_gen, tracker except Exception as e: logger.error(f"Erro ao inicializar sistema de performance: {e}") return None, None, None if __name__ == "__main__": # Código para testes import sqlite3 try: conn = sqlite3.connect('revalida.db') analyzer, material_gen, tracker = initialize_performance_system(conn) # Teste básico test_user = "test_user_1" metrics = analyzer.get_performance_metrics(test_user) if metrics: print("Sistema funcionando corretamente") print(f"Métricas obtidas: {json.dumps(metrics, indent=2)}") except Exception as e: print(f"Erro nos testes: {e}") finally: conn.close()