core/cache.py

import asyncio
import json
import re
import uuid
from collections import defaultdict
from pathlib import Path

import numpy as np
# Redis removido para deploy no HF Spaces
# import redis.asyncio as redis
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import FAISS

from config import MANIFEST_FILE, logger, settings
from core.utils import limpar_nome_uc, resolver_pasta_faiss_uc


faiss_cache: dict[str, FAISS] = {}
bm25_cache: dict[str, BM25Retriever] = {}
docs_cache: dict[str, list] = {}

_cache_locks: dict[str, asyncio.Lock] = defaultdict(asyncio.Lock)
_ingestao_locks: dict[str, asyncio.Lock] = defaultdict(asyncio.Lock)


# ── REDIS (DESATIVADO) ───────────────────────────────────────────────────────

redis_client = None

# paraphrase-multilingual-MiniLM-L12-v2 usa 384 dimensões
EMBEDDING_DIM = int(getattr(settings, "embedding_dim", 384))

# Nova versão para evitar conflitos com índices antigos/corrompidos
INDEX_NAME = "idx:semantic_cache_v3"
CACHE_PREFIX = "cache_v3:"

redis_cache_disponivel = False


async def fechar_conexoes() -> None:
    """Fecha ligações Redis no shutdown da app."""
    pass


def _normalizar_uc_cache(uc: str) -> str:
    return limpar_nome_uc(uc or "")


def _normalizar_versao_cache(versao: str) -> str:
    bruto = (versao or "sem_versao").strip().lower()
    bruto = re.sub(r"[^a-z0-9_]", "_", bruto)
    bruto = re.sub(r"_+", "_", bruto).strip("_")

    return bruto or "sem_versao"


def _escape_tag_value(valor: str) -> str:
    """
    Escapa caracteres especiais para queries TAG do RediSearch.
    Mesmo com UCs normalizadas, isto evita crashes em valores inesperados.
    """

    valor = valor or ""

    return re.sub(
        r'([\\{}\[\]\(\)\-:"\'@~&|!<>])',
        r"\\\1",
        valor,
    )


def _redis_erro_indica_indice_inexistente(exc: Exception) -> bool:
    msg = str(exc).lower()

    return (
        "unknown index" in msg
        or "unknown index name" in msg
        or "no such index" in msg
        or "index does not exist" in msg
    )


def _redis_erro_indica_modulo_ausente(exc: Exception) -> bool:
    msg = str(exc).lower()

    return (
        "unknown command" in msg
        or ("ft.search" in msg and "unknown" in msg)
        or ("module" in msg and "search" in msg)
    )


def obter_versao_uc(uc: str) -> str:
    uc_norm = _normalizar_uc_cache(uc)
    manifest_path = resolver_pasta_faiss_uc(uc_norm) / MANIFEST_FILE

    if not manifest_path.exists():
        return "sem_versao"

    try:
        with manifest_path.open("r", encoding="utf-8") as f:
            manifest = json.load(f)

        return _normalizar_versao_cache(manifest.get("version", "sem_versao"))

    except Exception as exc:
        logger.warning(
            "Falha a ler manifest da UC %s: %s",
            uc_norm,
            type(exc).__name__,
        )

        return "sem_versao"


async def init_redis_index() -> None:
    """
    Inicializa o índice vetorial do Redis.

    Se Redis/RediSearch não estiver disponível, a cache semântica é desativada
    sem partir o RAG.
    """

    global redis_cache_disponivel

    if not getattr(settings, "semantic_cache_enabled", True):
        redis_cache_disponivel = False
        logger.info("Cache semântica Redis desativada por configuração.")
        return

    try:
        await redis_client.ping()

    except Exception as exc:
        redis_cache_disponivel = False

        logger.warning(
            "Redis indisponível. Cache semântica desativada. erro=%s | detalhe=%s",
            type(exc).__name__,
            str(exc)[:200],
        )

        return

    try:
        await redis_client.ft(INDEX_NAME).info()

        redis_cache_disponivel = True
        logger.info("Índice Redis %s já existe — a saltar criação.", INDEX_NAME)

        return

    except Exception as exc:
        if _redis_erro_indica_modulo_ausente(exc):
            redis_cache_disponivel = False

            logger.warning(
                "Redis está ativo, mas RediSearch/Redis Stack não parece disponível. "
                "Cache semântica desativada. detalhe=%s",
                str(exc)[:200],
            )

            return

        if not _redis_erro_indica_indice_inexistente(exc):
            redis_cache_disponivel = False

            logger.warning(
                "Erro inesperado ao verificar índice Redis. "
                "Cache semântica desativada. erro=%s | detalhe=%s",
                type(exc).__name__,
                str(exc)[:300],
            )

            return

    try:
        logger.info("A criar índice vetorial Redis %s...", INDEX_NAME)

        schema = (
            TagField("uc"),
            TagField("versao_uc"),
            TextField("resposta_json"),
            VectorField(
                "embedding",
                "FLAT",
                {
                    "TYPE": "FLOAT32",
                    "DIM": EMBEDDING_DIM,
                    "DISTANCE_METRIC": "COSINE",
                },
            ),
        )

        definition = IndexDefinition(
            prefix=[CACHE_PREFIX],
            index_type=IndexType.HASH,
        )

        await redis_client.ft(INDEX_NAME).create_index(
            fields=schema,
            definition=definition,
        )

        redis_cache_disponivel = True
        logger.info("Índice vetorial Redis criado com sucesso: %s", INDEX_NAME)

    except Exception as exc:
        redis_cache_disponivel = False

        logger.warning(
            "Falha ao criar índice Redis. Cache semântica desativada. "
            "erro=%s | detalhe=%s",
            type(exc).__name__,
            str(exc)[:300],
        )


async def procurar_cache_redis(
    uc: str,
    versao_uc: str,
    query_emb: list[float],
    threshold: float,
) -> dict | None:
    """
    Procura resposta semanticamente semelhante no Redis.

    Se o Redis falhar, não interrompe o RAG: apenas ignora a cache.
    """

    if not getattr(settings, "semantic_cache_enabled", True):
        return None

    if not redis_cache_disponivel:
        return None

    try:
        uc_norm = _normalizar_uc_cache(uc)
        versao_norm = _normalizar_versao_cache(versao_uc)

        uc_tag = _escape_tag_value(uc_norm)
        versao_tag = _escape_tag_value(versao_norm)

        emb_bytes = np.array(query_emb, dtype=np.float32).tobytes()
        max_dist = 1.0 - float(threshold)

        query_text = (
            f"(@uc:{{{uc_tag}}} @versao_uc:{{{versao_tag}}})"
            f"=>[KNN 1 @embedding $vec AS distance]"
        )

        query = (
            Query(query_text)
            .sort_by("distance")
            .return_fields("resposta_json", "distance")
            .dialect(2)
        )

        result = await redis_client.ft(INDEX_NAME).search(
            query,
            query_params={"vec": emb_bytes},
        )

        if not result.docs:
            return None

        distance = float(result.docs[0].distance)

        if distance <= max_dist:
            logger.info(
                "Cache hit no Redis. UC=%s versão=%s dist=%.3f",
                uc_norm,
                versao_norm,
                distance,
            )

            resposta_raw = result.docs[0].resposta_json

            if isinstance(resposta_raw, bytes):
                resposta_raw = resposta_raw.decode("utf-8")

            return json.loads(resposta_raw)

    except Exception as exc:
        logger.warning(
            "Erro ao ler cache do Redis. A ignorar cache nesta pergunta. "
            "uc=%s | erro=%s | detalhe=%s",
            uc,
            type(exc).__name__,
            str(exc)[:300],
        )

    return None


async def guardar_cache_redis(
    uc: str,
    versao_uc: str,
    emb: list[float],
    resposta: dict,
    ttl_dias: int = 7,
) -> None:
    """
    Guarda resposta segura no Redis.

    Não guarda pergunta original nem query expandida para evitar fuga de dados
    entre utilizadores.
    """

    if not getattr(settings, "semantic_cache_enabled", True):
        return

    if not redis_cache_disponivel:
        return

    try:
        resposta_segura = {
            "resposta_stu": resposta.get("resposta_stu"),
            "sem_contexto": resposta.get("sem_contexto"),
            "fontes_consultadas": resposta.get("fontes_consultadas", []),
        }

        if "calendario" in resposta:
            resposta_segura["calendario"] = resposta["calendario"]

        uc_norm = _normalizar_uc_cache(uc)
        versao_norm = _normalizar_versao_cache(versao_uc)

        key = f"{CACHE_PREFIX}{uc_norm}:{uuid.uuid4().hex}"
        emb_bytes = np.array(emb, dtype=np.float32).tobytes()

        await redis_client.hset(
            key,
            mapping={
                "uc": uc_norm,
                "versao_uc": versao_norm,
                "embedding": emb_bytes,
                "resposta_json": json.dumps(resposta_segura, ensure_ascii=False),
            },
        )

        await redis_client.expire(key, int(ttl_dias) * 24 * 60 * 60)

    except Exception as exc:
        logger.warning(
            "Erro ao guardar cache no Redis. A resposta ao aluno não é afetada. "
            "erro=%s | detalhe=%s",
            type(exc).__name__,
            str(exc)[:300],
        )


async def invalidar_cache_redis_uc(uc: str) -> None:
    if not redis_cache_disponivel:
        return

    try:
        uc_norm = _normalizar_uc_cache(uc)
        pattern = f"{CACHE_PREFIX}{uc_norm}:*"

        cursor = 0
        total = 0

        while True:
            cursor, keys = await redis_client.scan(
                cursor=cursor,
                match=pattern,
                count=200,
            )

            if keys:
                await redis_client.delete(*keys)
                total += len(keys)

            if cursor == 0:
                break

        logger.info(
            "Cache Redis invalidada para UC=%s (%d chaves)",
            uc_norm,
            total,
        )

    except Exception as exc:
        logger.warning(
            "Erro ao invalidar cache Redis da UC %s: %s | detalhe=%s",
            uc,
            type(exc).__name__,
            str(exc)[:300],
        )


def invalidar_caches_memoria_uc(
    uc: str,
    context_id: int | str | None = None,
    subject_id: int | str | None = None,
) -> None:
    """
    Remove todas as variantes de chaves da UC da cache local (memória).
    """
    uc_limpa = limpar_nome_uc(uc)
    keys_to_pop = {uc, uc_limpa}

    if context_id is not None:
        keys_to_pop.add(str(context_id))
    if subject_id is not None:
        keys_to_pop.add(str(subject_id))

    try:
        folder = resolver_pasta_faiss_uc(uc_limpa)
        manifest_path = folder / MANIFEST_FILE
        if manifest_path.is_file():
            with manifest_path.open("r", encoding="utf-8") as f:
                manifest = json.load(f)
                m_context_id = manifest.get("context_id")
                m_subject_id = manifest.get("subject_id")
                if m_context_id is not None:
                    keys_to_pop.add(str(m_context_id))
                if m_subject_id is not None:
                    keys_to_pop.add(str(m_subject_id))
    except Exception as exc:
        logger.warning(
            "Falha ao resolver pasta ou ler manifest para invalidar caches: %s",
            exc,
        )

    id_keys = {str(value) for value in (context_id, subject_id) if value is not None}
    try:
        base_dir = Path(settings.base_faiss_dir)
        for folder in sorted(base_dir.iterdir(), key=lambda p: p.name.lower()):
            if not folder.is_dir():
                continue

            manifest_path = folder / MANIFEST_FILE
            if not manifest_path.is_file():
                continue

            with manifest_path.open("r", encoding="utf-8") as f:
                manifest = json.load(f)

            manifest_ids = {
                str(value)
                for value in (manifest.get("context_id"), manifest.get("subject_id"))
                if value is not None
            }
            if id_keys and manifest_ids.intersection(id_keys):
                keys_to_pop.add(folder.name)
                keys_to_pop.add(limpar_nome_uc(folder.name))
    except Exception as exc:
        logger.warning(
            "Falha ao procurar manifests duplicados para invalidar caches: %s",
            exc,
        )

    logger.info("A invalidar caches na memoria para as chaves: %s", keys_to_pop)
    for key in keys_to_pop:
        if key:
            faiss_cache.pop(key, None)
            bm25_cache.pop(key, None)
            docs_cache.pop(key, None)