Spaces:
Sleeping
Sleeping
| """ | |
| 🤖 N8n Assistant - Versão Corrigida (SEM ERRO DE IMAGEM) | |
| Chatbot inteligente para dúvidas sobre n8n - Compatível com Hugging Face Spaces | |
| CORREÇÃO APLICADA: | |
| - Removido componente Image problemático que causava erro 404 | |
| - Mantida toda funcionalidade do sistema | |
| """ | |
| import os | |
| import yaml | |
| import json | |
| import logging | |
| import time | |
| from typing import Optional, Tuple | |
| import gradio as gr | |
| # Configurar logging | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| # Importações com tratamento de erro | |
| try: | |
| from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings | |
| from llama_index.llms.openai import OpenAI | |
| from llama_index.embeddings.openai import OpenAIEmbedding | |
| from huggingface_hub import snapshot_download | |
| logger.info("✅ Bibliotecas importadas com sucesso") | |
| except ImportError as e: | |
| logger.error(f"❌ Erro ao importar bibliotecas: {e}") | |
| raise | |
| class N8nAssistant: | |
| """Assistente N8n simplificado e funcional""" | |
| def __init__(self): | |
| self.index = None | |
| self.query_engine = None | |
| self.docs_dir = None | |
| self.inicializado = False | |
| def setup_openai(self) -> bool: | |
| """Configurar OpenAI""" | |
| try: | |
| api_key = os.getenv("OPENAI_API_KEY") | |
| if not api_key: | |
| logger.error("❌ OPENAI_API_KEY não encontrada") | |
| return False | |
| os.environ["OPENAI_API_KEY"] = api_key | |
| logger.info("✅ OpenAI configurada") | |
| return True | |
| except Exception as e: | |
| logger.error(f"❌ Erro OpenAI: {e}") | |
| return False | |
| def extrair_conteudo_arquivos(self, pasta: str) -> str: | |
| """Extrair conteúdo dos arquivos""" | |
| texto_final = "" | |
| if not os.path.exists(pasta): | |
| logger.error(f"❌ Pasta não encontrada: {pasta}") | |
| return "" | |
| for root, dirs, files in os.walk(pasta): | |
| for file in files: | |
| caminho_arquivo = os.path.join(root, file) | |
| try: | |
| if file.endswith(('.yml', '.yaml')): | |
| with open(caminho_arquivo, 'r', encoding='utf-8') as f: | |
| data = yaml.safe_load(f) | |
| texto = yaml.dump(data, allow_unicode=True) | |
| texto_final += f"\n\n### Arquivo: {file}\n{texto}" | |
| elif file.endswith('.json'): | |
| with open(caminho_arquivo, 'r', encoding='utf-8') as f: | |
| data = json.load(f) | |
| texto = json.dumps(data, indent=2, ensure_ascii=False) | |
| texto_final += f"\n\n### Arquivo: {file}\n{texto}" | |
| elif file.endswith(('.md', '.txt')): | |
| with open(caminho_arquivo, 'r', encoding='utf-8') as f: | |
| texto = f.read() | |
| texto_final += f"\n\n### Arquivo: {file}\n{texto}" | |
| except Exception as e: | |
| logger.warning(f"⚠️ Erro ao ler {file}: {e}") | |
| continue | |
| return texto_final | |
| def gerar_documentacao(self, pasta_origem: str) -> bool: | |
| """Gerar arquivo de documentação""" | |
| try: | |
| texto = self.extrair_conteudo_arquivos(pasta_origem) | |
| if not texto.strip(): | |
| logger.warning("⚠️ Nenhum conteúdo encontrado") | |
| return False | |
| with open("documentacao.txt", 'w', encoding='utf-8') as f: | |
| f.write(texto) | |
| logger.info("✅ Documentação gerada") | |
| return True | |
| except Exception as e: | |
| logger.error(f"❌ Erro ao gerar documentação: {e}") | |
| return False | |
| def baixar_docs(self) -> bool: | |
| """Baixar documentação do HF""" | |
| try: | |
| logger.info("📥 Baixando documentação...") | |
| self.docs_dir = snapshot_download( | |
| repo_id="Jeice/n8n-docs-v2", | |
| repo_type="dataset" | |
| ) | |
| logger.info("✅ Download concluído") | |
| return True | |
| except Exception as e: | |
| logger.error(f"❌ Erro no download: {e}") | |
| return False | |
| def criar_index(self) -> bool: | |
| """Criar índice vetorial""" | |
| try: | |
| # Carregar documentos | |
| if not os.path.exists("documentacao.txt"): | |
| logger.error("❌ documentacao.txt não encontrado") | |
| return False | |
| documents = SimpleDirectoryReader(input_files=["documentacao.txt"]).load_data() | |
| if not documents: | |
| logger.error("❌ Nenhum documento carregado") | |
| return False | |
| # Configurar LLM | |
| Settings.llm = OpenAI( | |
| model="gpt-3.5-turbo", | |
| temperature=0.1, | |
| system_prompt=( | |
| "Você é um assistente especialista em n8n. " | |
| "Responda sempre em português do Brasil, de forma clara e objetiva, " | |
| "baseado exclusivamente na documentação fornecida. " | |
| "Se não souber, diga que não há informações suficientes." | |
| ) | |
| ) | |
| Settings.embed_model = OpenAIEmbedding() | |
| # Criar índice | |
| logger.info("🧠 Criando índice...") | |
| self.index = VectorStoreIndex.from_documents(documents) | |
| self.query_engine = self.index.as_query_engine() | |
| logger.info("✅ Índice criado") | |
| return True | |
| except Exception as e: | |
| logger.error(f"❌ Erro ao criar índice: {e}") | |
| return False | |
| def inicializar(self) -> Tuple[bool, str]: | |
| """Inicializar sistema completo""" | |
| try: | |
| # 1. Configurar OpenAI | |
| if not self.setup_openai(): | |
| return False, "Erro na configuração OpenAI" | |
| # 2. Baixar docs | |
| if not self.baixar_docs(): | |
| return False, "Erro no download da documentação" | |
| # 3. Gerar documentação | |
| if not self.gerar_documentacao(self.docs_dir): | |
| return False, "Erro ao processar documentação" | |
| # 4. Criar índice | |
| if not self.criar_index(): | |
| return False, "Erro ao criar índice" | |
| self.inicializado = True | |
| return True, "Sistema inicializado com sucesso" | |
| except Exception as e: | |
| logger.error(f"❌ Erro na inicialização: {e}") | |
| return False, f"Erro: {str(e)}" | |
| def responder(self, pergunta: str) -> str: | |
| """Responder pergunta""" | |
| if not pergunta or not pergunta.strip(): | |
| return "⚠️ Por favor, digite uma pergunta." | |
| if not self.inicializado or not self.query_engine: | |
| return "❌ Sistema não inicializado. Recarregue a página." | |
| try: | |
| logger.info(f"🤔 Pergunta: {pergunta[:50]}...") | |
| response = self.query_engine.query(pergunta) | |
| return str(response) | |
| except Exception as e: | |
| logger.error(f"❌ Erro ao responder: {e}") | |
| return f"❌ Erro ao processar pergunta: {str(e)}" | |
| # Inicializar sistema | |
| logger.info("🚀 Inicializando N8n Assistant...") | |
| assistant = N8nAssistant() | |
| sucesso, mensagem = assistant.inicializar() | |
| if sucesso: | |
| logger.info(f"✅ {mensagem}") | |
| else: | |
| logger.error(f"❌ {mensagem}") | |
| def processar_pergunta(pergunta: str) -> str: | |
| """Wrapper para Gradio""" | |
| if not sucesso: | |
| return f"❌ Sistema não inicializado: {mensagem}" | |
| return assistant.responder(pergunta) | |
| # Interface Gradio - SEM COMPONENTE IMAGE PROBLEMÁTICO | |
| with gr.Blocks(theme=gr.themes.Soft(), title="N8n Assistant") as demo: | |
| # Cabeçalho | |
| gr.Markdown( | |
| f""" | |
| # 🤖 N8n Assistant | |
| Assistente inteligente para dúvidas sobre **n8n** baseado na documentação oficial. | |
| """ | |
| #**Status:** {'✅ Sistema Pronto' if sucesso else '❌ ' + mensagem} | |
| ) | |
| # Layout principal - SEM IMAGEM | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| # REMOVIDO: componente gr.Image que causava erro 404 | |
| gr.Markdown("### 🤖 N8n Bot") | |
| with gr.Column(scale=4): | |
| gr.Markdown("## Como posso ajudar você com o n8n?") | |
| with gr.Row(): | |
| with gr.Column(scale=3): | |
| input_box = gr.Textbox( | |
| label="Sua pergunta", | |
| placeholder="Ex: Como criar um workflow no n8n?", | |
| lines=3 | |
| ) | |
| with gr.Row(): | |
| enviar_btn = gr.Button("🚀 Perguntar", variant="primary") | |
| limpar_btn = gr.Button("🧹 Limpar") | |
| with gr.Column(scale=4): | |
| output_box = gr.Textbox( | |
| label="Resposta", | |
| placeholder="Sua resposta aparecerá aqui...", | |
| lines=12 | |
| ) | |
| # Exemplos | |
| with gr.Accordion("💡 Exemplos de Perguntas", open=False): | |
| gr.Markdown( | |
| """ | |
| - Como criar um workflow no n8n? | |
| - Para que serve o node HTTP Request? | |
| - Como integrar n8n com Google Sheets? | |
| - Como configurar webhooks no n8n? | |
| - Quais são as melhores práticas para workflows? | |
| - Como debugar erros nos nodes? | |
| - Como usar condições nos workflows? | |
| - Quais nodes usar para automação de email? | |
| """ | |
| ) | |
| # Eventos | |
| enviar_btn.click( | |
| fn=processar_pergunta, | |
| inputs=input_box, | |
| outputs=output_box | |
| ) | |
| limpar_btn.click( | |
| lambda: ("", ""), | |
| None, | |
| [input_box, output_box] | |
| ) | |
| input_box.submit( | |
| fn=processar_pergunta, | |
| inputs=input_box, | |
| outputs=output_box | |
| ) | |
| # Lançar aplicação | |
| if __name__ == "__main__": | |
| demo.launch( | |
| server_name="0.0.0.0", | |
| server_port=7860, | |
| show_error=True | |
| ) | |