app/core/state.py

"""
Estado global de la aplicación: configuración, retriever,
modelo de embeddings y metadata.  Todos los módulos que
necesiten acceder o mutar el estado deben importar este módulo
como `import app.core.state as state` y acceder a sus atributos.
"""

from threading import Lock
from typing import Any, Dict, List

import faiss
import numpy as np

from utils import base_utils as bu
from utils import retrieval_utils as ru
from utils.retrievers import get_retriever


# ---------------------------------------------------------------------------
# Funciones de inicialización (no dependen de estado global)
# ---------------------------------------------------------------------------

def load_embedding_model(config: dict):
    model_name = config["embeddings"]["model_name"]
    return ru.load_model(model_name)


def _needs_e5_prefix() -> bool:
    """Los modelos intfloat/e5-* requieren los prefijos `query:` / `passage:`."""
    name = str(CONFIG.get("embeddings", {}).get("model_name", "")).lower()
    return "e5" in name


def embed_query(model, text: str, normalize: bool = True) -> np.ndarray:
    if _needs_e5_prefix() and not text.lstrip().lower().startswith("query:"):
        text = f"query: {text}"
    vec = model.encode(text)
    vec = np.asarray(vec, dtype="float32")[None, :]
    if normalize:
        faiss.normalize_L2(vec)
    return vec


# ---------------------------------------------------------------------------
# Estado global (inicializado al importar el módulo, igual que antes)
# ---------------------------------------------------------------------------

CONFIG: Dict[str, Any] = bu.load_config("configs/config.json")
RETRIEVER = get_retriever(CONFIG)
EMBED_MODEL = load_embedding_model(CONFIG)
PROMPTS: Dict[str, Any] = bu.load_prompts()

META_PROMPTS: Dict[str, Any] = (
    PROMPTS.get("meta", {}) if isinstance(PROMPTS, dict) else {}
)
TRIGGERS: Dict[str, Any] = (
    META_PROMPTS.get("triggers", {}) if isinstance(META_PROMPTS, dict) else {}
)

NO_INFO_PREFIX: str = META_PROMPTS.get(
    "no_information_prefix",
    "Não há informações suficientes na base.",
)
ABOUT_BOT_TEXT: str = META_PROMPTS.get(
    "about_bot",
    (
        "Sou o Chatbot NORM, um assistente especializado em química e materiais "
        "radioativos de ocorrência natural (NORM). Respondo com base nos documentos "
        "técnicos indexados na base de conhecimento deste sistema."
    ),
)
GREETING_TEXT: str = META_PROMPTS.get(
    "greeting",
    (
        "Olá! Sou o Chatbot NORM, um assistente treinado para responder perguntas "
        "sobre química e NORM usando os documentos técnicos indexados. Você pode, "
        "por exemplo, pedir um resumo de um documento específico ou perguntar sobre "
        "um isótopo radioativo."
    ),
)
HELP_SCOPE_TEXT: str = META_PROMPTS.get(
    "help_scope",
    (
        "Posso ajudar respondendo perguntas sobre química, radioatividade natural "
        "(NORM), isótopos específicos e documentos técnicos indexados neste sistema. "
        "Faça perguntas objetivas e, se possível, mencione o tema ou documento de "
        "interesse para eu recuperar os trechos mais relevantes."
    ),
)

METADATA: List[Dict[str, Any]] = getattr(RETRIEVER, "metadata", [])

# Lista de documentos mostrados na última listagem (persistida por sessão do servidor)
LAST_LISTED_DOCS: List[Dict[str, Any]] = []

# Lock para evitar condições de corrida ao reindexar documentos
INDEX_UPDATE_LOCK = Lock()