app.py

"""
🤖 N8n Assistant - Open Source (GRÁTIS, CPU-friendly)
- Sem OpenAI
- LLM: google/flan-t5-base (fallback flan-t5-small)
- Embeddings: all-MiniLM-L6-v2 (fallback paraphrase-MiniLM-L3-v2)
- Baixa dataset Jeice/n8n-docs-v2 e gera documentacao.txt
- Logs detalhados p/ depuração
"""

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

import gradio as gr
from huggingface_hub import snapshot_download

# LlamaIndex
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings, ServiceContext
from llama_index.core.settings import Settings as LISettings
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.llms.huggingface import HuggingFaceLLM

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

# -------------------------
# Modelos (CPU-friendly)
# -------------------------
PRIMARY_LLM = "google/flan-t5-base"
FALLBACK_LLM = "google/flan-t5-small"

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

# -------------------------
# App
# -------------------------
class N8nAssistant:
    def __init__(self):
        self.docs_dir = None
        self.index = None
        self.query_engine = None
        self.inicializado = False
        self.llm_model_used = None
        self.emb_model_used = None

    # ---------- Dataset ----------
    def baixar_docs(self) -> bool:
        """Baixa o dataset com a documentação."""
        try:
            logger.info("📥 Baixando dataset Jeice/n8n-docs-v2 ...")
            self.docs_dir = snapshot_download(
                repo_id="Jeice/n8n-docs-v2",
                repo_type="dataset"
            )
            logger.info(f"✅ Dataset baixado em: {self.docs_dir}")
            try:
                logger.info(f"📂 Itens no diretório raiz do dataset: {os.listdir(self.docs_dir)}")
                data_path = os.path.join(self.docs_dir, "data")
                if os.path.isdir(data_path):
                    logger.info(f"📂 Pasta /data encontrada. Itens: {os.listdir(data_path)}")
            except Exception as e:
                logger.warning(f"⚠️ Não consegui listar arquivos do dataset: {e}")
            return True
        except Exception as e:
            logger.error(f"❌ Erro ao baixar dataset: {e}")
            return False

    # ---------- Consolidação ----------
    def extrair_conteudo_arquivos(self, pasta: str) -> str:
        """Varre todas as subpastas e agrega .yml/.yaml/.json/.md/.txt em um único texto."""
        extensoes = ('.yml', '.yaml', '.json', '.md', '.txt')
        texto_final = []
        if not os.path.exists(pasta):
            logger.error(f"❌ Pasta não existe: {pasta}")
            return ""

        total_arquivos = 0
        for root, _, files in os.walk(pasta):
            logger.info(f"🔎 Explorando: {root} | {len(files)} arquivos")
            for file in files:
                caminho = os.path.join(root, file)
                if not file.lower().endswith(extensoes):
                    continue
                total_arquivos += 1
                try:
                    if file.lower().endswith(('.yml', '.yaml')):
                        with open(caminho, 'r', encoding='utf-8') as f:
                            data = yaml.safe_load(f)
                        texto = yaml.dump(data, allow_unicode=True, sort_keys=False)
                    elif file.lower().endswith('.json'):
                        with open(caminho, 'r', encoding='utf-8') as f:
                            data = json.load(f)
                        texto = json.dumps(data, ensure_ascii=False, indent=2)
                    else:  # .md / .txt
                        with open(caminho, 'r', encoding='utf-8', errors='ignore') as f:
                            texto = f.read()

                    texto_final.append(f"\n\n### Arquivo: {os.path.relpath(caminho, pasta)}\n{texto}")
                except Exception as e:
                    logger.warning(f"⚠️ Erro lendo {caminho}: {e}")

        logger.info(f"🧾 Total de arquivos agregados: {total_arquivos}")
        return "".join(texto_final)

    def gerar_documentacao(self) -> bool:
        """Gera documentacao.txt a partir do dataset (raiz + /data se existir)."""
        try:
            if not self.docs_dir:
                logger.error("❌ docs_dir não definido")
                return False

            partes = []
            # raiz do dataset
            partes.append(self.extrair_conteudo_arquivos(self.docs_dir))
            # subpasta /data (comum em datasets do HF)
            data_path = os.path.join(self.docs_dir, "data")
            if os.path.isdir(data_path):
                partes.append(self.extrair_conteudo_arquivos(data_path))

            texto = "\n".join([p for p in partes if p and p.strip()])
            if not texto.strip():
                logger.error("❌ Nenhum conteúdo válido encontrado no dataset")
                return False

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

            # Loga um preview
            preview = texto[:1500]
            logger.info(f"📝 documentacao.txt gerado (preview 1500 chars):\n{preview}")
            return True
        except Exception as e:
            logger.error(f"❌ Erro ao gerar documentacao.txt: {e}")
            return False

    # ---------- Modelos ----------
    def configurar_embeddings(self) -> bool:
        for emb in (PRIMARY_EMB, FALLBACK_EMB):
            try:
                LISettings.embed_model = HuggingFaceEmbedding(model_name=emb)
                self.emb_model_used = emb
                logger.info(f"✅ Embeddings carregados: {emb}")
                return True
            except Exception as e:
                logger.warning(f"⚠️ Falhou carregar embeddings {emb}: {e}")
        return False

    def configurar_llm(self) -> bool:
        gen_kwargs = {
            "temperature": 0.2,
            "do_sample": True,
            "top_p": 0.9
        }
        for name in (PRIMARY_LLM, FALLBACK_LLM):
            try:
                llm = HuggingFaceLLM(
                    model_name=name,
                    tokenizer_name=name,
                    context_window=2048,
                    max_new_tokens=384,   # menor = mais leve em CPU
                    generate_kwargs=gen_kwargs,
                    device_map="auto",
                    model_kwargs={"torch_dtype": "auto"},
                    system_prompt=(
                        "Você é um assistente especialista em n8n. "
                        "Responda 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."
                    ),
                )
                LISettings.llm = llm
                self.llm_model_used = name
                logger.info(f"✅ LLM carregado: {name}")
                return True
            except Exception as e:
                logger.warning(f"⚠️ Falhou carregar LLM {name}: {e}")
        return False

    # ---------- Index ----------
    def criar_index(self) -> bool:
        try:
            if not os.path.exists("documentacao.txt"):
                logger.error("❌ documentacao.txt não existe")
                return False

            # Carrega o único arquivo consolidado
            docs = SimpleDirectoryReader(input_files=["documentacao.txt"]).load_data()
            if not docs:
                logger.error("❌ Nenhum documento carregado de documentacao.txt")
                with open("documentacao.txt", "r", encoding="utf-8") as f:
                    logger.error("📄 documentacao.txt (trecho): " + f.read()[:1200])
                return False

            logger.info(f"📚 {len(docs)} documento(s) prontos para indexação")
            self.index = VectorStoreIndex.from_documents(docs)
            self.query_engine = self.index.as_query_engine()
            logger.info("✅ Índice e QueryEngine criados")
            return True
        except Exception as e:
            logger.error(f"❌ Erro ao criar índice: {e}")
            return False

    # ---------- Orquestração ----------
    def inicializar(self) -> Tuple[bool, str]:
        try:
            if not self.baixar_docs():
                return False, "Erro ao baixar dataset"
            if not self.gerar_documentacao():
                return False, "Erro ao gerar documentacao.txt"
            if not self.configurar_embeddings():
                return False, "Erro ao configurar embeddings"
            if not self.configurar_llm():
                return False, "Erro ao configurar LLM"
            if not self.criar_index():
                return False, "Erro ao criar índice"
            self.inicializado = True
            return True, f"Pronto | LLM: {self.llm_model_used} | Emb: {self.emb_model_used}"
        except Exception as e:
            logger.error(f"❌ Erro na inicialização: {e}")
            return False, f"Erro na inicialização: {e}"

    def responder(self, pergunta: str) -> str:
        if 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]}")
            resp = self.query_engine.query(pergunta)
            return str(resp)
        except Exception as e:
            logger.error(f"❌ Erro na resposta: {e}")
            return f"❌ Erro ao processar a pergunta: {e}"


# -------------------------
# Bootstrap
# -------------------------
logger.info("🚀 Subindo N8n Assistant (Open Source, CPU)...")
assistant = N8nAssistant()
ok, status_msg = assistant.inicializar()
if ok:
    logger.info(f"✅ {status_msg}")
else:
    logger.error(f"❌ {status_msg}")

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

with gr.Blocks(theme=gr.themes.Soft(), title="N8n Assistant") as demo:
    gr.Markdown(
        f"""
        # 🤖 N8n Assistant (Open Source)
        Assistente baseado na documentação oficial do **n8n** (dataset do HF).
        **Status:** {'✅ ' + status_msg if ok else '❌ ' + status_msg}
        """
    )
    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):
                    pergunta = gr.Textbox(
                        label="Sua pergunta",
                        placeholder="Ex: Como configurar um Webhook Trigger no n8n?",
                        lines=3
                    )
                    enviar = gr.Button("🚀 Perguntar", variant="primary")
                    limpar = gr.Button("🧹 Limpar")
                with gr.Column(scale=4):
                    resposta = gr.Textbox(
                        label="Resposta",
                        placeholder="A resposta aparecerá aqui...",
                        lines=14
                    )
    with gr.Accordion("💡 Exemplos", open=False):
        gr.Markdown(
            """
            - Como configurar webhooks no n8n?
            - Para que serve o node HTTP Request?
            - Como integrar com Google Sheets?
            - Como debugar erros nos nodes?
            - Quais são boas práticas de workflows?
            """
        )

    enviar.click(fn=processar_pergunta, inputs=pergunta, outputs=resposta)
    limpar.click(lambda: ("", ""), None, [pergunta, resposta])
    pergunta.submit(fn=processar_pergunta, inputs=pergunta, outputs=resposta)

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