# app.py import gradio as gr from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline import json import os import sqlite3 from datetime import datetime, timedelta import pandas as pd from typing import Dict, List, Optional class RevalidaDatabase: def __init__(self): self.conn = sqlite3.connect('revalida.db') self.create_tables() self.load_previous_questions() def create_tables(self): """Cria tabelas necessárias""" self.conn.execute(''' CREATE TABLE IF NOT EXISTS users ( user_id TEXT PRIMARY KEY, name TEXT, study_hours INTEGER, start_date DATE, target_date DATE, weak_areas TEXT )''') self.conn.execute(''' CREATE TABLE IF NOT EXISTS study_progress ( user_id TEXT, date DATE, topic TEXT, hours_studied FLOAT, questions_answered INTEGER, performance_score FLOAT, FOREIGN KEY (user_id) REFERENCES users(user_id) )''') self.conn.execute(''' CREATE TABLE IF NOT EXISTS previous_questions ( id INTEGER PRIMARY KEY, year INTEGER, area TEXT, question_text TEXT, options TEXT, correct_answer TEXT, explanation TEXT )''') self.conn.commit() def load_previous_questions(self): """Carrega questões de provas anteriores do arquivo JSON""" try: with open('previous_questions.json', 'r', encoding='utf-8') as f: questions = json.load(f) cursor = self.conn.cursor() for q in questions: cursor.execute(''' INSERT OR IGNORE INTO previous_questions (year, area, question_text, options, correct_answer, explanation) VALUES (?, ?, ?, ?, ?, ?) ''', (q['year'], q['area'], q['question_text'], json.dumps(q['options']), q['correct_answer'], q['explanation'])) self.conn.commit() except Exception as e: print(f"Erro ao carregar questões: {e}") class StudyPlanGenerator: def __init__(self, db: RevalidaDatabase): self.db = db def create_study_plan(self, user_id: str, available_hours: int, target_date: str, weak_areas: List[str]) -> Dict: """Gera plano de estudos personalizado""" # Calcula período total de estudo start_date = datetime.now() target_date = datetime.strptime(target_date, '%Y-%m-%d') total_days = (target_date - start_date).days # Define distribuição de tópicos topics = { "Clínica Médica": 0.25, "Cirurgia": 0.20, "Pediatria": 0.15, "Ginecologia e Obstetrícia": 0.15, "Medicina de Família": 0.15, "Saúde Mental": 0.10 } # Ajusta pesos para áreas fracas for area in weak_areas: if area in topics: topics[area] += 0.05 # Normaliza pesos total_weight = sum(topics.values()) topics = {k: v/total_weight for k, v in topics.items()} # Gera cronograma schedule = {} for week in range(total_days // 7 + 1): schedule[f"Semana {week+1}"] = { topic: round(hours * available_hours * topics[topic], 1) for topic in topics } return schedule class RevalidaBot: def __init__(self): self.db = RevalidaDatabase() self.planner = StudyPlanGenerator(self.db) # Modelo de linguagem self.model_id = "meta-llama/Llama-2-7b-chat-hf" self.tokenizer = AutoTokenizer.from_pretrained(self.model_id) self.pipeline = pipeline( "text-generation", model=self.model_id, tokenizer=self.tokenizer, max_length=500 ) # Comandos disponíveis self.commands = { "/start": self.start_command, "/plano": self.create_plan, "/progresso": self.check_progress, "/questoes": self.previous_questions, "/estudo": self.study_mode, "/ajuda": self.help_command } def process_message(self, message: str, user_id: str, history: List) -> str: """Processa mensagens recebidas""" try: if message.startswith("/"): command = message.split()[0] if command in self.commands: return self.commands[command](message, user_id) return "Comando não reconhecido. Use /ajuda para ver os comandos disponíveis." return self.generate_response(message, history) except Exception as e: return f"Desculpe, ocorreu um erro: {str(e)}" def create_plan(self, message: str, user_id: str) -> str: """Cria plano de estudos personalizado""" try: # Extrai parâmetros da mensagem # Formato: /plano horas_diarias YYYY-MM-DD área1,área2 parts = message.split() if len(parts) < 3: return "Formato: /plano horas_diarias YYYY-MM-DD área1,área2" hours = int(parts[1]) target_date = parts[2] weak_areas = parts[3].split(',') if len(parts) > 3 else [] plan = self.planner.create_study_plan( user_id, hours, target_date, weak_areas ) # Formata resposta response = "📚 Seu Plano de Estudos:\n\n" for week, topics in plan.items(): response += f"{week}:\n" for topic, hours in topics.items(): response += f"- {topic}: {hours}h\n" response += "\n" return response except Exception as e: return f"Erro ao criar plano: {str(e)}" def check_progress(self, message: str, user_id: str) -> str: """Verifica progresso do usuário""" try: cursor = self.db.conn.cursor() 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,)) progress = cursor.fetchall() response = "📊 Seu Progresso:\n\n" for topic, hours, score in progress: response += f"{topic}:\n" response += f"- Horas estudadas: {hours:.1f}h\n" response += f"- Desempenho médio: {score:.1f}%\n\n" return response except Exception as e: return f"Erro ao verificar progresso: {str(e)}" def previous_questions(self, message: str, user_id: str) -> str: """Busca questões de provas anteriores""" try: # Formato: /questoes área [ano] parts = message.split() if len(parts) < 2: return "Formato: /questoes área [ano]" area = parts[1] year = int(parts[2]) if len(parts) > 2 else None cursor = self.db.conn.cursor() if year: cursor.execute(''' SELECT question_text, options, correct_answer, explanation FROM previous_questions WHERE area = ? AND year = ? ORDER BY RANDOM() LIMIT 1 ''', (area, year)) else: cursor.execute(''' SELECT question_text, options, correct_answer, explanation FROM previous_questions WHERE area = ? ORDER BY RANDOM() LIMIT 1 ''', (area,)) question = cursor.fetchone() if not question: return "Nenhuma questão encontrada com esses critérios." response = f"📝 Questão:\n\n{question[0]}\n\n" options = json.loads(question[1]) for letter, text in options.items(): response += f"{letter}) {text}\n" response += f"\nResposta: {question[2]}\n" response += f"Explicação: {question[3]}" return response except Exception as e: return f"Erro ao buscar questão: {str(e)}" def study_mode(self, message: str, user_id: str) -> str: """Modo estudo com acompanhamento""" try: # Formato: /estudo área horas parts = message.split() if len(parts) < 3: return "Formato: /estudo área horas" area = parts[1] hours = float(parts[2]) # Registra estudo cursor = self.db.conn.cursor() cursor.execute(''' INSERT INTO study_progress (user_id, date, topic, hours_studied, questions_answered, performance_score) VALUES (?, ?, ?, ?, ?, ?) ''', (user_id, datetime.now().date(), area, hours, 0, 0.0)) self.db.conn.commit() return f"✅ Registrado: {hours}h de estudo em {area}\n\nContinue assim! Use /progresso para ver seu desenvolvimento." except Exception as e: return f"Erro ao registrar estudo: {str(e)}" # Outros métodos necessários... def create_interface(): """Cria interface Gradio""" bot = RevalidaBot() with gr.Blocks(title="Assistente Revalida") as interface: gr.Markdown("# 🏥 Assistente Revalida Pro") gr.Markdown("### Seu mentor personalizado para o Revalida") with gr.Row(): with gr.Column(): user_id = gr.Textbox( label="Seu ID (email ou telefone)", placeholder="Digite seu ID único..." ) chatbot = gr.Chatbot() msg = gr.Textbox( placeholder="Digite sua mensagem ou comando aqui...", show_label=False ) clear = gr.ClearButton([msg, chatbot]) def process_message_with_user(message, history): user = user_id.value if user_id.value else "default_user" return bot.process_message(message, user, history) msg.submit( process_message_with_user, [msg, chatbot], [chatbot] ) return interface # Inicia a interface if __name__ == "__main__": interface = create_interface() interface.launch()