app.py

"""
🤖 N8n Assistant - Versão Open Source (GRÁTIS)
- Sem OpenAI
- LLM: microsoft/Phi-3.5-mini-instruct (fallback flan-t5-base)
- Embeddings: all-MiniLM-L6-v2 (fallback L3-v2)
- Compatível com Hugging Face Spaces (CPU)
"""

import os
import yaml
import json
import logging
from typing import Optional, Tuple

import gradio as gr

# LlamaIndex (open source stacks)
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.llms.huggingface import HuggingFaceLLM

from huggingface_hub import snapshot_download

# ------------------------------------------------------------
# Logging
# ------------------------------------------------------------
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("n8n-assistant")

# ------------------------------------------------------------
# Configs de modelos (primários + fallbacks)
# ------------------------------------------------------------
PRIMARY_LLM = "microsoft/Phi-3.5-mini-instruct"
FALLBACK_LLM = "google/flan-t5-base"  # muito leve

PRIMARY_EMB = "sentence-transformers/all-MiniLM-L6-v2"
FALLBACK_EMB = "sentence-transformers/paraphrase-MiniLM-L3-v2"

# ------------------------------------------------------------
# Classe principal
# ------------------------------------------------------------
class N8nAssistant:
    """Assistente N8n open-source e funcional"""

    def __init__(self):
        self.index = None
        self.query_engine = None
        self.docs_dir = None
        self.inicializado = False
        self.llm_model_used = None
        self.emb_model_used = None

    # --------- Utilitários de dados ----------
    def extrair_conteudo_arquivos(self, pasta: str) -> str:
        """Extrai conteúdo textual dos arquivos .yml/.yaml/.json/.md/.txt"""
        texto_final = ""

        if not os.path.exists(pasta):
            logger.error(f"❌ Pasta não encontrada: {pasta}")
            return ""

        for root, _, 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:
        """Gera um único arquivo 'documentacao.txt' com todo o conteúdo unificado"""
        try:
            texto = self.extrair_conteudo_arquivos(pasta_origem)
            if not texto.strip():
                logger.warning("⚠️ Nenhum conteúdo encontrado para documentação")
                return False

            with open("documentacao.txt", 'w', encoding='utf-8') as f:
                f.write(texto)

            logger.info("✅ Documentação consolidada em documentacao.txt")
            return True

        except Exception as e:
            logger.error(f"❌ Erro ao gerar documentação: {e}")
            return False

    def baixar_docs(self) -> bool:
        """Baixa a documentação do HF dataset"""
        try:
            logger.info("📥 Baixando documentação do dataset Jeice/n8n-docs-v2 ...")
            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 do dataset: {e}")
            return False

    # --------- Configuração de modelos ----------
    def configurar_embeddings(self) -> bool:
        """Configura embeddings HuggingFace com fallback"""
        for emb in (PRIMARY_EMB, FALLBACK_EMB):
            try:
                Settings.embed_model = HuggingFaceEmbedding(model_name=emb)
                self.emb_model_used = emb
                logger.info(f"✅ Embeddings configurados: {emb}")
                return True
            except Exception as e:
                logger.warning(f"⚠️ Falha ao carregar embeddings {emb}: {e}")
        logger.error("❌ Não foi possível configurar embeddings")
        return False

    def configurar_llm(self) -> bool:
        """Configura LLM HuggingFace com fallback, otimizado para CPU"""
        # parâmetros neutros/seguros para CPU
        gen_kwargs = {
            "temperature": 0.2,
            "do_sample": True,
            "top_p": 0.9
        }
        # tentar primário depois fallback
        for model_name in (PRIMARY_LLM, FALLBACK_LLM):
            try:
                llm = HuggingFaceLLM(
                    model_name=model_name,
                    tokenizer_name=model_name,
                    context_window=4096,
                    max_new_tokens=512,
                    generate_kwargs=gen_kwargs,
                    # device_map="auto" funciona em CPU/GPU no Space
                    device_map="auto",
                    model_kwargs={
                        # dtype padrão (evitar float16 em CPU)
                        "torch_dtype": "auto"
                    },
                    # system_prompt para orientar o estilo de resposta
                    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.llm = llm
                self.llm_model_used = model_name
                logger.info(f"✅ LLM configurado: {model_name}")
                return True
            except Exception as e:
                logger.warning(f"⚠️ Falha ao carregar LLM {model_name}: {e}")

        logger.error("❌ Não foi possível configurar o LLM")
        return False

    # --------- Indexação ----------
    def criar_index(self) -> bool:
        """Cria o índice vetorial a partir de documentacao.txt"""
        try:
            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

            # Criar índice + query engine
            logger.info("🧠 Criando índice (VectorStoreIndex) ...")
            self.index = VectorStoreIndex.from_documents(documents)
            self.query_engine = self.index.as_query_engine()
            logger.info("✅ Índice criado e query_engine pronto")
            return True

        except Exception as e:
            logger.error(f"❌ Erro ao criar índice: {e}")
            return False

    # --------- Orquestração ----------
    def inicializar(self) -> Tuple[bool, str]:
        """Pipeline completo de inicialização (open-source)"""
        try:
            # 1) Baixar docs
            if not self.baixar_docs():
                return False, "Erro ao baixar a documentação (dataset)"

            # 2) Consolidar documentação
            if not self.gerar_documentacao(self.docs_dir):
                return False, "Erro ao processar/consolidar a documentação"

            # 3) Configurar embeddings e LLM (open source)
            if not self.configurar_embeddings():
                return False, "Erro ao configurar embeddings"
            if not self.configurar_llm():
                return False, "Erro ao configurar LLM"

            # 4) Criar índice
            if not self.criar_index():
                return False, "Erro ao criar o índice"

            self.inicializado = True
            return True, (
                f"Sistema inicializado com sucesso | "
                f"LLM: {self.llm_model_used} | Embeddings: {self.emb_model_used}"
            )

        except Exception as e:
            logger.error(f"❌ Erro na inicialização: {e}")
            return False, f"Erro: {str(e)}"

    def responder(self, pergunta: str) -> str:
        """Executa a consulta no query_engine"""
        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[:120]}...")
            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)}"


# ------------------------------------------------------------
# Bootstrap
# ------------------------------------------------------------
logger.info("🚀 Inicializando N8n Assistant (Open Source)...")
assistant = N8nAssistant()
sucesso, mensagem = assistant.inicializar()

if sucesso:
    logger.info(f"✅ {mensagem}")
else:
    logger.error(f"❌ {mensagem}")

# ------------------------------------------------------------
# Gradio UI
# ------------------------------------------------------------
def processar_pergunta(pergunta: str) -> str:
    if not sucesso:
        return f"❌ Sistema não inicializado: {mensagem}"
    return assistant.responder(pergunta)

with gr.Blocks(theme=gr.themes.Soft(), title="N8n Assistant (Open Source)") as demo:
    gr.Markdown(
        f"""
        # 🤖 N8n Assistant (Open Source)
        Assistente para dúvidas sobre **n8n** baseado na documentação oficial e em modelos **open-source**.
        **Status:** {'✅ Sistema Pronto' if sucesso else '❌ ' + mensagem}
        """
    )

    with gr.Row():
        with gr.Column(scale=1):
            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
                    )

    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?
            """
        )

    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)

if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860, show_error=True)