Create performance_system.py

performance_system.py

@@ -0,0 +1,238 @@
+# 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()