Create database/db_manager.py

database/db_manager.py

@@ -0,0 +1,348 @@
+# database/db_manager.py
+
+import sqlite3
+import json
+import logging
+from datetime import datetime
+from typing import Dict, List, Optional, Any
+import os
+
+# Configuração de logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+class DatabaseManager:
+    """Gerenciador central do banco de dados"""
+    
+    def __init__(self, db_path: str = 'revalida.db'):
+        """Inicializa o gerenciador de banco de dados"""
+        self.db_path = db_path
+        self.conn = None
+        self.initialize_database()
+
+    def get_connection(self) -> sqlite3.Connection:
+        """Obtém conexão com o banco de dados"""
+        if not self.conn:
+            self.conn = sqlite3.connect(self.db_path)
+            self.conn.row_factory = sqlite3.Row
+        return self.conn
+
+    def initialize_database(self) -> None:
+        """Inicializa todas as tabelas necessárias"""
+        try:
+            conn = self.get_connection()
+            cursor = conn.cursor()
+
+            # Tabela de Usuários
+            cursor.execute('''
+            CREATE TABLE IF NOT EXISTS users (
+                user_id TEXT PRIMARY KEY,
+                name TEXT NOT NULL,
+                email TEXT UNIQUE,
+                registration_date DATE,
+                target_date DATE,
+                study_hours INTEGER,
+                weak_areas TEXT,
+                preferences TEXT
+            )''')
+
+            # Tabela de Progresso de Estudo
+            cursor.execute('''
+            CREATE TABLE IF NOT EXISTS study_progress (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                user_id TEXT,
+                date DATE,
+                topic TEXT,
+                hours_studied FLOAT,
+                questions_answered INTEGER,
+                performance_score FLOAT,
+                notes TEXT,
+                FOREIGN KEY (user_id) REFERENCES users(user_id)
+            )''')
+
+            # Tabela de Questões Anteriores
+            cursor.execute('''
+            CREATE TABLE IF NOT EXISTS previous_questions (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                year INTEGER,
+                area TEXT,
+                question_text TEXT,
+                options TEXT,
+                correct_answer TEXT,
+                explanation TEXT,
+                difficulty TEXT,
+                references TEXT,
+                tags TEXT,
+                times_used INTEGER DEFAULT 0,
+                success_rate FLOAT DEFAULT 0.0
+            )''')
+
+            # Tabela de Casos Clínicos
+            cursor.execute('''
+            CREATE TABLE IF NOT EXISTS clinical_cases (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                title TEXT,
+                area TEXT,
+                difficulty TEXT,
+                description TEXT,
+                steps TEXT,
+                expected_answers TEXT,
+                hints TEXT,
+                references TEXT,
+                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+            )''')
+
+            # Tabela de Simulados
+            cursor.execute('''
+            CREATE TABLE IF NOT EXISTS simulados (
+                id TEXT PRIMARY KEY,
+                user_id TEXT,
+                created_at TIMESTAMP,
+                completed_at TIMESTAMP,
+                difficulty TEXT,
+                questions TEXT,
+                answers TEXT,
+                score FLOAT,
+                time_taken INTEGER,
+                analysis TEXT,
+                FOREIGN KEY (user_id) REFERENCES users(user_id)
+            )''')
+
+            # Tabela de Sessões de Estudo
+            cursor.execute('''
+            CREATE TABLE IF NOT EXISTS study_sessions (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                user_id TEXT,
+                start_time TIMESTAMP,
+                end_time TIMESTAMP,
+                topic TEXT,
+                activity_type TEXT,
+                productivity_score FLOAT,
+                notes TEXT,
+                FOREIGN KEY (user_id) REFERENCES users(user_id)
+            )''')
+
+            # Tabela de Metas e Objetivos
+            cursor.execute('''
+            CREATE TABLE IF NOT EXISTS goals (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                user_id TEXT,
+                goal_type TEXT,
+                description TEXT,
+                target_date DATE,
+                progress FLOAT,
+                status TEXT,
+                priority INTEGER,
+                FOREIGN KEY (user_id) REFERENCES users(user_id)
+            )''')
+
+            conn.commit()
+            logger.info("Banco de dados inicializado com sucesso")
+
+        except Exception as e:
+            logger.error(f"Erro ao inicializar banco de dados: {e}")
+            raise
+
+    def load_questions_from_json(self, file_path: str) -> bool:
+        """Carrega questões de um arquivo JSON"""
+        try:
+            if not os.path.exists(file_path):
+                logger.error(f"Arquivo não encontrado: {file_path}")
+                return False
+
+            with open(file_path, 'r', encoding='utf-8') as f:
+                questions = json.load(f)
+
+            conn = self.get_connection()
+            cursor = conn.cursor()
+
+            for q in questions:
+                cursor.execute('''
+                INSERT OR IGNORE INTO previous_questions
+                (year, area, question_text, options, correct_answer, 
+                 explanation, difficulty, references, tags)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+                ''', (
+                    q.get('year'),
+                    q.get('area'),
+                    q.get('question_text'),
+                    json.dumps(q.get('options', {})),
+                    q.get('correct_answer'),
+                    q.get('explanation'),
+                    q.get('difficulty', 'medium'),
+                    json.dumps(q.get('references', [])),
+                    json.dumps(q.get('tags', []))
+                ))
+
+            conn.commit()
+            logger.info(f"Questões carregadas com sucesso de {file_path}")
+            return True
+
+        except Exception as e:
+            logger.error(f"Erro ao carregar questões: {e}")
+            return False
+
+    def add_user(self, user_data: Dict[str, Any]) -> bool:
+        """Adiciona ou atualiza usuário"""
+        try:
+            conn = self.get_connection()
+            cursor = conn.cursor()
+
+            cursor.execute('''
+            INSERT OR REPLACE INTO users
+            (user_id, name, email, registration_date, target_date, 
+             study_hours, weak_areas, preferences)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+            ''', (
+                user_data['user_id'],
+                user_data['name'],
+                user_data.get('email'),
+                user_data.get('registration_date', datetime.now().date()),
+                user_data.get('target_date'),
+                user_data.get('study_hours'),
+                json.dumps(user_data.get('weak_areas', [])),
+                json.dumps(user_data.get('preferences', {}))
+            ))
+
+            conn.commit()
+            logger.info(f"Usuário {user_data['user_id']} adicionado/atualizado com sucesso")
+            return True
+
+        except Exception as e:
+            logger.error(f"Erro ao adicionar usuário: {e}")
+            return False
+
+    def get_user_profile(self, user_id: str) -> Optional[Dict]:
+        """Obtém perfil completo do usuário"""
+        try:
+            conn = self.get_connection()
+            cursor = conn.cursor()
+
+            # Dados básicos do usuário
+            cursor.execute('SELECT * FROM users WHERE user_id = ?', (user_id,))
+            user = cursor.fetchone()
+
+            if not user:
+                return None
+
+            # Convertendo Row para dict
+            user_dict = dict(user)
+            
+            # Decodificando JSON
+            user_dict['weak_areas'] = json.loads(user_dict['weak_areas'])
+            user_dict['preferences'] = json.loads(user_dict['preferences'])
+
+            # Progresso recente
+            cursor.execute('''
+            SELECT topic, SUM(hours_studied) as total_hours,
+                   AVG(performance_score) as avg_score
+            FROM study_progress
+            WHERE user_id = ?
+            GROUP BY topic
+            ''', (user_id,))
+            
+            user_dict['progress'] = {
+                row['topic']: {
+                    'hours': row['total_hours'],
+                    'score': row['avg_score']
+                }
+                for row in cursor.fetchall()
+            }
+
+            # Metas ativas
+            cursor.execute('''
+            SELECT * FROM goals
+            WHERE user_id = ? AND status != 'completed'
+            ORDER BY priority DESC
+            ''', (user_id,))
+            
+            user_dict['active_goals'] = [dict(row) for row in cursor.fetchall()]
+
+            return user_dict
+
+        except Exception as e:
+            logger.error(f"Erro ao obter perfil do usuário: {e}")
+            return None
+
+    def update_study_progress(self, progress_data: Dict[str, Any]) -> bool:
+        """Atualiza progresso de estudo"""
+        try:
+            conn = self.get_connection()
+            cursor = conn.cursor()
+
+            cursor.execute('''
+            INSERT INTO study_progress
+            (user_id, date, topic, hours_studied, 
+             questions_answered, performance_score, notes)
+            VALUES (?, ?, ?, ?, ?, ?, ?)
+            ''', (
+                progress_data['user_id'],
+                progress_data.get('date', datetime.now().date()),
+                progress_data['topic'],
+                progress_data['hours_studied'],
+                progress_data.get('questions_answered', 0),
+                progress_data.get('performance_score', 0.0),
+                progress_data.get('notes')
+            ))
+
+            conn.commit()
+            logger.info(f"Progresso atualizado para usuário {progress_data['user_id']}")
+            return True
+
+        except Exception as e:
+            logger.error(f"Erro ao atualizar progresso: {e}")
+            return False
+
+    def close(self):
+        """Fecha conexão com o banco de dados"""
+        if self.conn:
+            self.conn.close()
+            self.conn = None
+
+    def __enter__(self):
+        """Suporte para context manager"""
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        """Fecha conexão ao sair do contexto"""
+        self.close()
+
+# Função de inicialização
+def initialize_database(db_path: str = 'revalida.db') -> DatabaseManager:
+    """Inicializa e retorna instância do gerenciador de banco de dados"""
+    try:
+        db_manager = DatabaseManager(db_path)
+        return db_manager
+    except Exception as e:
+        logger.error(f"Erro ao inicializar gerenciador de banco de dados: {e}")
+        raise
+
+if __name__ == "__main__":
+    # Código para testes
+    try:
+        db_manager = initialize_database('test_revalida.db')
+        
+        # Teste de inserção de usuário
+        test_user = {
+            'user_id': 'test123',
+            'name': 'Teste User',
+            'email': '[email protected]',
+            'study_hours': 4,
+            'weak_areas': ['ClínicaMédica', 'Cirurgia']
+        }
+        
+        success = db_manager.add_user(test_user)
+        if success:
+            print("Sistema de banco de dados funcionando corretamente")
+            
+            # Recuperar perfil para verificar
+            profile = db_manager.get_user_profile('test123')
+            print(f"Perfil recuperado: {json.dumps(profile, indent=2)}")
+            
+    except Exception as e:
+        print(f"Erro nos testes: {e}")
+    finally:
+        db_manager.close()