Upload 11 files

oauth/__init__.py

@@ -0,0 +1 @@
+# Arquivo vazio para marcar o diretório como um pacote Python 

oauth/cadastro.py

@@ -0,0 +1,250 @@
+import sys
+from pathlib import Path
+
+# Adicionar o diretório raiz ao PYTHONPATH
+root_dir = Path(__file__).parent.parent
+sys.path.append(str(root_dir))
+
+import streamlit as st
+import pandas as pd
+from oauth.user_manager import UserManager
+from oauth.supabase_conex import SupabaseConnection
+from datetime import datetime
+from sollai.assistentes import CriaAssistente, AssistenteConfig
+
+def create_admin_user(email: str, password: str, nome: str, is_admin: bool = False):
+    """Cria um novo usuário diretamente"""
+    try:
+        # Usar admin client para ter permissões totais
+        admin_client = SupabaseConnection().get_admin_client()
+        
+        # 1. Criar usuário na auth
+        auth_response = admin_client.auth.admin.create_user({
+            "email": email,
+            "password": password,
+            "email_confirm": True,
+            "user_metadata": {
+                "nome": nome,
+                "is_admin": is_admin
+            }
+        })
+        
+        # Debug detalhado
+        print("\n=== Debug Auth Response ===")
+        print(f"Tipo: {type(auth_response)}")
+        print(f"Dir: {dir(auth_response)}")
+        print(f"Raw: {auth_response}")
+        
+        if not auth_response:
+            raise Exception("Falha na criação da autenticação")
+        
+        # Tentar diferentes formas de acessar o ID
+        user_id = None
+        if hasattr(auth_response, 'user'):
+            user_id = auth_response.user.id
+        elif hasattr(auth_response, 'id'):
+            user_id = auth_response.id
+        else:
+            # Último recurso: converter para dict
+            auth_dict = auth_response.model_dump() if hasattr(auth_response, 'model_dump') else vars(auth_response)
+            print(f"Auth Dict: {auth_dict}")
+            user_id = auth_dict.get('id') or auth_dict.get('user', {}).get('id')
+        
+        if not user_id:
+            raise Exception("Não foi possível obter o ID do usuário")
+        
+        # 2. Inserir dados na tabela public.users
+        user_data = {
+            "id": user_id,
+            "email": email,
+            "nome": nome,
+            "is_admin": is_admin,
+            "created_at": datetime.utcnow().isoformat(),
+            "updated_at": datetime.utcnow().isoformat(),
+            "is_active": True
+        }
+        
+        response = admin_client.from_('users')\
+            .insert(user_data)\
+            .execute()
+        
+        return response.data[0]
+        
+    except Exception as e:
+        print(f"\n=== Erro Detalhado ===")
+        print(f"Tipo do erro: {type(e)}")
+        print(f"Mensagem: {str(e)}")
+        import traceback
+        print(f"Traceback:\n{traceback.format_exc()}")
+        raise Exception(f"Erro ao criar usuário: {str(e)}")
+
+def delete_user(user_id: str):
+    """Deleta um usuário do sistema"""
+    try:
+        admin_client = SupabaseConnection().get_admin_client()
+        
+        # 1. Deletar da tabela users primeiro
+        response = admin_client.from_('users')\
+            .delete()\
+            .eq('id', user_id)\
+            .execute()
+            
+        # 2. Deletar da autenticação
+        auth_response = admin_client.auth.admin.delete_user(user_id)
+        
+        return True
+        
+    except Exception as e:
+        print(f"\n=== Erro ao deletar usuário ===")
+        print(f"Tipo do erro: {type(e)}")
+        print(f"Mensagem: {str(e)}")
+        import traceback
+        print(f"Traceback:\n{traceback.format_exc()}")
+        raise Exception(f"Erro ao deletar usuário: {str(e)}")
+
+def create_user(email: str, password: str, nome: str):
+    """Cria um novo usuário e seu assistente"""
+    try:
+        # 1. Criar usuário no Supabase
+        user = create_admin_user(
+            email=email,
+            password=password,
+            nome=nome,
+            is_admin=False
+        )
+        
+        # 2. Criar assistente no Pinecone
+        safe_name = "".join(c for c in nome if c.isalnum() or c.isspace()).strip()
+        safe_name = safe_name.lower().replace(" ", "-")
+        assistant_name = f"assistant-{safe_name}-{user['id'][:8]}"
+        
+        config = AssistenteConfig(
+            nome=assistant_name,
+            instrucoes="Você é uma assistente amigável que responde perguntas do usuário. Responda sempre em português do Brasil."
+        )
+        
+        try:
+            criador = CriaAssistente()
+            assistente = criador.criar(config)
+            
+            # 3. Atualizar usuário com referências do assistente
+            admin_client = SupabaseConnection().get_admin_client()
+            response = admin_client.from_('users')\
+                .update({
+                    "assistant_name": assistant_name  # Salvando apenas o nome
+                })\
+                .eq('id', user['id'])\
+                .execute()
+            
+            st.success(f"✅ Usuário {nome} criado com sucesso!")
+            st.success(f"✅ Assistente '{assistant_name}' criado com sucesso!")
+            
+            return response.data[0]
+            
+        except Exception as e:
+            st.error(f"❌ Erro ao criar assistente: {str(e)}")
+            delete_user(user['id'])
+            raise
+            
+    except Exception as e:
+        st.error(f"❌ Erro ao criar usuário: {str(e)}")
+        raise
+
+def main():
+    st.set_page_config(page_title="Cadastro de Usuários", page_icon="👤")
+    
+    st.title("👤 Cadastro de Usuários")
+    st.markdown("### Área Administrativa")
+    
+    # Formulário de cadastro
+    with st.form("admin_cadastro_form"):
+        nome = st.text_input("Nome completo", help="Digite o nome completo do usuário")
+        email = st.text_input("Email", help="Digite o email do usuário")
+        password = st.text_input("Senha", type="password", help="Mínimo 6 caracteres")
+        confirm_password = st.text_input("Confirmar senha", type="password")
+        is_admin = st.checkbox("👑 Usuário Administrador")
+        
+        if st.form_submit_button("Criar Usuário"):
+            try:
+                # Validações
+                if not all([nome, email, password, confirm_password]):
+                    st.error("❌ Todos os campos são obrigatórios!")
+                    st.stop()
+                
+                if password != confirm_password:
+                    st.error("❌ As senhas não coincidem!")
+                    st.stop()
+                
+                if len(password) < 6:
+                    st.error("❌ A senha deve ter no mínimo 6 caracteres!")
+                    st.stop()
+                
+                # Criar usuário
+                user = create_user(
+                    email=email,
+                    password=password,
+                    nome=nome
+                )
+                
+                st.json(user)  # Mostra os dados do usuário criado
+                
+            except Exception as e:
+                st.error(f"❌ Erro ao criar usuário: {str(e)}")
+    
+    # Lista de usuários existentes
+    st.markdown("---")
+    st.subheader("📋 Usuários Cadastrados")
+    
+    try:
+        admin_client = SupabaseConnection().get_admin_client()
+        response = admin_client.from_('users').select('*').execute()
+        
+        if response.data:
+            # Criar DataFrame com os dados
+            df = pd.DataFrame(response.data)
+            
+            # Mostrar tabela
+            st.dataframe(
+                df,
+                column_config={
+                    "nome": "Nome",
+                    "email": "Email",
+                    "is_admin": "Admin",
+                    "is_active": "Ativo",
+                    "created_at": "Criado em"
+                },
+                hide_index=True
+            )
+            
+            # Seleção de usuário para deletar
+            usuarios = [(f"{u['nome']} ({u['email']})", u['id']) for u in response.data]
+            selected_user = st.selectbox(
+                "Selecione um usuário para deletar:",
+                options=usuarios,
+                format_func=lambda x: x[0]
+            )
+            
+            # Botão de deletar com confirmação
+            if st.button("🗑️ Deletar Usuário"):
+                if st.session_state.get('is_admin'):  # Verifica se é admin
+                    user_id = selected_user[1]
+                    
+                    # Confirmação adicional
+                    if st.warning(f"⚠️ Tem certeza que deseja deletar este usuário? Esta ação não pode ser desfeita."):
+                        try:
+                            delete_user(user_id)
+                            st.success("✅ Usuário deletado com sucesso!")
+                            st.rerun()  # Atualiza a página
+                        except Exception as e:
+                            st.error(f"❌ Erro ao deletar usuário: {str(e)}")
+                else:
+                    st.error("❌ Apenas administradores podem deletar usuários!")
+                
+        else:
+            st.info("Nenhum usuário cadastrado.")
+            
+    except Exception as e:
+        st.error(f"❌ Erro ao listar usuários: {str(e)}")
+
+if __name__ == "__main__":
+    main()

oauth/login.py

@@ -0,0 +1,38 @@
+import streamlit as st
+from oauth.user_manager import UserManager
+from sollai.assistentes import Assistente
+
+def main():
+    st.title("🔐 Login")
+    
+    with st.form("login_form"):
+        email = st.text_input("Email")
+        password = st.text_input("Senha", type="password")
+        
+        if st.form_submit_button("Entrar"):
+            try:
+                user_manager = UserManager()
+                user = user_manager.authenticate_user(email, password)
+                
+                if user:
+                    # Guardar dados do usuário
+                    st.session_state['user'] = user
+                    st.session_state['is_authenticated'] = True
+                    
+                    # Carregar assistente usando o nome
+                    assistant_name = user.get('assistant_name')
+                    if assistant_name:
+                        try:
+                            assistente = Assistente(assistant_name)
+                            st.session_state['assistant'] = assistente
+                            st.success("✅ Login realizado com sucesso!")
+                            st.rerun()
+                        except Exception as e:
+                            st.error(f"❌ Erro ao carregar assistente: {str(e)}")
+                    else:
+                        st.error("❌ Você não possui um assistente configurado.")
+                else:
+                    st.error("❌ Email ou senha inválidos!")
+                    
+            except Exception as e:
+                st.error(f"❌ Erro ao fazer login: {str(e)}")

oauth/supabase_conex.py

@@ -0,0 +1,74 @@
+import os
+from dotenv import load_dotenv
+from supabase import create_client, Client
+from typing import Optional
+
+load_dotenv()
+
+class SupabaseConnection:
+    """Gerenciador de conexão com Supabase"""
+    
+    _instance: Optional['SupabaseConnection'] = None
+    
+    def __new__(cls):
+        """Implementa Singleton para evitar múltiplas conexões"""
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+        return cls._instance
+    
+    def __init__(self):
+        """Inicializa a conexão com Supabase"""
+        if not hasattr(self, 'client'):
+            self.supabase_url = os.getenv("SUPABASE_URL")
+            self.supabase_key = os.getenv("SUPABASE_KEY")
+            self.service_role_key = os.getenv("SUPABASE_SERVICE_ROLE_KEY")
+            
+            if not all([self.supabase_url, self.supabase_key, self.service_role_key]):
+                raise ValueError(
+                    "SUPABASE_URL, SUPABASE_KEY e SUPABASE_SERVICE_ROLE_KEY devem estar definidos no arquivo .env"
+                )
+            
+            # Cliente normal para operações regulares
+            self.client = create_client(
+                supabase_url=self.supabase_url,
+                supabase_key=self.supabase_key
+            )
+            
+            # Cliente admin para operações administrativas
+            self.admin_client = create_client(
+                supabase_url=self.supabase_url,
+                supabase_key=self.service_role_key
+            )
+    
+    def get_client(self) -> Client:
+        """Retorna o cliente Supabase regular"""
+        return self.client
+    
+    def get_admin_client(self) -> Client:
+        """Retorna o cliente Supabase com privilégios administrativos"""
+        return self.admin_client
+    
+    def test_connection(self) -> bool:
+        """Testa a conexão com Supabase"""
+        try:
+            # Tenta fazer uma query simples
+            print("🔍 Testando conexão com Supabase...")
+            response = self.client.table('users').select("*").limit(1).execute()
+            print("✅ Conexão Supabase bem sucedida!")
+            return True
+            
+        except Exception as e:
+            print(f"❌ Erro na conexão Supabase: {str(e)}")
+            return False
+    
+    def get_table(self, table_name: str):
+        """
+        Retorna uma referência para uma tabela específica
+        
+        Args:
+            table_name: Nome da tabela
+            
+        Returns:
+            PostgrestFilterBuilder: Referência para a tabela
+        """
+        return self.client.table(table_name)

oauth/user_manager.py

@@ -0,0 +1,89 @@
+from datetime import datetime
+from typing import Dict, Optional, List
+from oauth.supabase_conex import SupabaseConnection
+
+class UserManager:
+    """Gerenciador de usuários usando Supabase"""
+    
+    def __init__(self):
+        self.supabase = SupabaseConnection().get_client()
+        self.users_table = 'users'
+    
+    def authenticate_user(self, email: str, password: str) -> Optional[Dict]:
+        """Autentica um usuário"""
+        try:
+            # Autenticar usuário
+            auth_response = self.supabase.auth.sign_in_with_password({
+                "email": email,
+                "password": password
+            })
+            
+            if auth_response:
+                # Usar admin_client para evitar recursão nas políticas
+                admin_client = SupabaseConnection().get_admin_client()
+                
+                # Buscar dados completos do usuário
+                user_response = admin_client.from_('users')\
+                    .select('*')\
+                    .eq('email', email)\
+                    .single()\
+                    .execute()
+                    
+                if user_response.data:
+                    print(f"=== Login realizado com sucesso ===")
+                    print(f"Usuário: {user_response.data['nome']} ({user_response.data['email']})")
+                    
+                    # Verificar se tem assistente
+                    if not user_response.data.get('assistant_name'):
+                        print("✗ Usuário não possui assistente configurado")
+                    
+                    return user_response.data
+                    
+            return None
+            
+        except Exception as e:
+            print(f"Erro na autenticação: {str(e)}")
+            raise
+    
+    def get_user(self, user_id: str) -> Optional[Dict]:
+        """Busca um usuário pelo ID"""
+        try:
+            # Usar admin_client para bypass nas políticas
+            admin_client = SupabaseConnection().get_admin_client()
+            
+            response = admin_client.from_(self.users_table)\
+                .select('*')\
+                .eq('id', user_id)\
+                .execute()
+            
+            return response.data[0] if response.data else None
+            
+        except Exception as e:
+            raise Exception(f"Erro ao buscar usuário: {str(e)}")
+    
+    def get_current_user(self) -> Optional[Dict]:
+        """Retorna o usuário atual"""
+        try:
+            session = self.supabase.auth.get_session()
+            if not session:
+                return None
+            
+            return self.get_user(session.user.id)
+        except Exception as e:
+            print(f"Erro ao buscar usuário atual: {str(e)}")
+            return None
+    
+    def is_admin(self, user_id: str) -> bool:
+        """Verifica se um usuário é admin"""
+        try:
+            user = self.get_user(user_id)
+            return user.get('is_admin', False) if user else False
+        except Exception as e:
+            print(f"Erro ao verificar admin: {str(e)}")
+            return False
+
+# Exemplo de uso
+if __name__ == "__main__":
+    user_manager = UserManager()
+    
+