core/utils.py

import re
import unicodedata
import logging
import datetime
import json
from pathlib import Path

# Reutilizar o logger configurado do sistema
logger = logging.getLogger("tuts")


def limpar_nome_uc(uc: str) -> str:
    """
    Normaliza o nome da UC para um formato seguro no sistema de ficheiros.
    Remove acentos, carateres especiais e limita o tamanho rigidamente.
    """
    sem_acentos = unicodedata.normalize("NFKD", uc)
    sem_acentos = sem_acentos.encode("ascii", "ignore").decode("ascii")
    limpo = "".join(x if x.isalnum() or x == "_" else "_" for x in sem_acentos)

    # Limite de 80 caracteres para prevenir erros no File System (Path Too Long)
    return re.sub(r"_+", "_", limpo).strip("_").lower()[:80]


def _base_faiss_dir() -> Path:
    from config import settings

    return Path(settings.base_faiss_dir)


def pasta_faiss_canonica_uc(uc: str) -> Path:
    """
    Caminho canonico para novas pastas FAISS.

    Novas ingestoes devem escrever sempre nesta pasta. Pastas antigas podem
    continuar a ser lidas atraves de resolver_pasta_faiss_uc().
    """
    return _base_faiss_dir() / limpar_nome_uc(uc)


def _manifest_updated_at(manifest: dict) -> datetime.datetime:
    raw = str(manifest.get("updated_at") or "").strip()
    if not raw:
        return datetime.datetime.min.replace(tzinfo=datetime.timezone.utc)

    try:
        return datetime.datetime.fromisoformat(raw.replace("Z", "+00:00"))
    except ValueError:
        return datetime.datetime.min.replace(tzinfo=datetime.timezone.utc)


def _manifest_name_matches(manifest: dict, nome_normalizado: str) -> bool:
    for key in ("uc", "canonical_uc", "canonical_name", "name", "subject_name"):
        value = manifest.get(key)
        if value is not None and limpar_nome_uc(str(value)) == nome_normalizado:
            return True

    aliases = manifest.get("aliases")
    if isinstance(aliases, list):
        for alias in aliases:
            if limpar_nome_uc(str(alias)) == nome_normalizado:
                return True

    return False


def _manifest_id_matches(manifest: dict, uc: str) -> bool:
    m_context_id = manifest.get("context_id")
    m_subject_id = manifest.get("subject_id")

    return (
        (m_context_id is not None and str(m_context_id) == str(uc))
        or (m_subject_id is not None and str(m_subject_id) == str(uc))
    )


def _read_manifest(path: Path) -> dict | None:
    manifest_path = path / "manifest.json"
    if not manifest_path.is_file():
        return None

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

    return manifest if isinstance(manifest, dict) else None


def resolver_pasta_faiss_uc(uc: str, preferred_uc: str | None = None) -> Path:
    """
    Resolve a pasta FAISS de uma UC com compatibilidade para nomes legacy.

    A ordem e:
    1. pasta canonica da UC preferida, quando fornecida;
    2. pasta canonica;
    3. primeira pasta existente em faiss_db cujo nome normalizado seja igual;
    4. correspondência por context_id ou subject_id no manifest.json;
    5. caminho canonico, mesmo que ainda nao exista.

    Esta funcao nunca renomeia nem apaga pastas.
    """
    nome_normalizado = limpar_nome_uc(uc)
    caminho_canonico = _base_faiss_dir() / nome_normalizado
    preferred_normalizado = limpar_nome_uc(preferred_uc or "")
    caminho_preferido = _base_faiss_dir() / preferred_normalizado if preferred_normalizado else None

    if caminho_preferido and caminho_preferido.exists():
        if preferred_normalizado != nome_normalizado:
            logger.info(
                "[TUTS][RAG][IndexResolve] direct UC folder preferred over manifest id mapping | subject_id=%s | uc=%s | folder=%s",
                uc,
                preferred_normalizado,
                caminho_preferido.name,
            )
        return caminho_preferido

    if caminho_canonico.exists():
        return caminho_canonico

    base = _base_faiss_dir()
    if base.exists():
        # 1. Correspondência exata por nome normalizado da pasta
        if preferred_normalizado:
            for pasta in sorted(base.iterdir(), key=lambda p: p.name.lower()):
                if pasta.is_dir() and limpar_nome_uc(pasta.name) == preferred_normalizado:
                    logger.info(
                        "[TUTS][RAG][IndexResolve] direct UC folder preferred over manifest id mapping | subject_id=%s | uc=%s | folder=%s",
                        uc,
                        preferred_normalizado,
                        pasta.name,
                    )
                    return pasta

        for pasta in sorted(base.iterdir(), key=lambda p: p.name.lower()):
            if pasta.is_dir() and limpar_nome_uc(pasta.name) == nome_normalizado:
                logger.warning(
                    "A usar pasta FAISS legacy para UC '%s': %s",
                    uc,
                    pasta,
                )
                return pasta

        # 2. Correspondência por ID da UC (context_id ou subject_id no manifest.json)
        manifest_matches: list[tuple[Path, dict]] = []
        for pasta in sorted(base.iterdir(), key=lambda p: p.name.lower()):
            if pasta.is_dir():
                manifest = _read_manifest(pasta)
                if manifest is not None and _manifest_id_matches(manifest, uc):
                    manifest_matches.append((pasta, manifest))

        if manifest_matches:
            if preferred_normalizado:
                preferred_matches = [
                    (pasta, manifest)
                    for pasta, manifest in manifest_matches
                    if limpar_nome_uc(pasta.name) == preferred_normalizado
                    or _manifest_name_matches(manifest, preferred_normalizado)
                ]

                if preferred_matches:
                    pasta, manifest = max(
                        preferred_matches,
                        key=lambda item: _manifest_updated_at(item[1]),
                    )
                    logger.info(
                        "[TUTS][RAG][IndexResolve] direct UC folder preferred over manifest id mapping | subject_id=%s | uc=%s | folder=%s",
                        uc,
                        preferred_normalizado,
                        pasta.name,
                    )
                    return pasta

            if len(manifest_matches) > 1:
                logger.warning(
                    "[TUTS][RAG][IndexResolve] duplicate subject_id mapping detected | subject_id=%s | folders=%s",
                    uc,
                    [pasta.name for pasta, _ in manifest_matches],
                )

            pasta, manifest = max(
                manifest_matches,
                key=lambda item: _manifest_updated_at(item[1]),
            )
            logger.info(
                "[TUTS][RAG][IndexResolve] Mapped ID '%s' to folder '%s' via manifest.json",
                uc,
                pasta.name,
            )
            return pasta

    return caminho_canonico


def sanitizar_input(texto: str, max_chars: int = 4000) -> str:
    """
    Prepara e limpa o texto inserido pelo aluno para prevenir injeções básicas
    e limitar o consumo de recursos.
    """
    # Remove Null Bytes que podem quebrar funções nativas de C no Python ou SQL
    texto = (texto or "").replace("\x00", "")

    # Remove as tags XML que usamos internamente para isolar a prompt no RAG
    texto = re.sub(r"</?pergunta_aluno>", "", texto, flags=re.IGNORECASE)
    texto = texto.strip()

    # Truncamento de segurança para evitar Prompt Stuffing e DoS de Tokens
    return texto[:max_chars]


def cosine_similarity(v1: list[float], v2: list[float]) -> float:
    """
    Calcula a similaridade de cosseno de forma segura.
    Valida as dimensões dos vetores antes de calcular para evitar exceções matemáticas.
    """
    import numpy as np

    a, b = np.asarray(v1, dtype=np.float32), np.asarray(v2, dtype=np.float32)

    # Prevenção de ValueError no dot product
    if a.shape != b.shape:
        logger.warning(
            "[UTILS] Erro Matemático Prevenido: Vetores com dimensões incompatíveis (%s vs %s)",
            a.shape,
            b.shape
        )
        return 0.0

    denom = np.linalg.norm(a) * np.linalg.norm(b)
    return float(np.dot(a, b) / denom) if denom > 0 else 0.0


def avaliar_suficiencia_contexto(pergunta: str, textos_chunks: list[str]) -> dict:
    """Avalia a suficiência do contexto recuperado para responder à pergunta."""
    def normalizar_texto(valor: str) -> str:
        sem_acentos = unicodedata.normalize("NFKD", valor or "")
        ascii_texto = sem_acentos.encode("ascii", "ignore").decode("ascii").lower()
        return re.sub(r"\s+", " ", re.sub(r"[^a-z0-9_]+", " ", ascii_texto)).strip()

    tokens_originais = re.findall(r"\b[\w-]+\b", pergunta or "")
    pergunta_norm = normalizar_texto(pergunta)
    contexto_norm = normalizar_texto(" ".join(str(t) for t in (textos_chunks or [])))
    if not pergunta_norm or not contexto_norm:
        return {
            "suficiente": False,
            "razao": "pergunta_ou_contexto_vazio",
            "cobertura": 0.0,
            "termos": [],
            "marcadores_presentes": [],
            "marcadores_ausentes": []
        }

    def correspondem(w1: str, w2: str) -> bool:
        if w1 == w2:
            return True
        
        # Plural simples
        if w1 + "s" == w2 or w2 + "s" == w1:
            return True
        if w1 + "es" == w2 or w2 + "es" == w1:
            return True
            
        # Plurais especiais em Português (al/ais, el/eis, il/is, ol/ois, ul/uis)
        for suffix_sing, suffix_plur in [("al", "ais"), ("el", "eis"), ("il", "is"), ("ol", "ois"), ("ul", "uis")]:
            if w1.endswith(suffix_sing) and w2.endswith(suffix_plur) and w1[:-2] == w2[:-3]:
                return True
            if w2.endswith(suffix_sing) and w1.endswith(suffix_plur) and w2[:-2] == w1[:-3]:
                return True
                
        # Mapeamentos bilingues / técnicos comuns
        pairs = [
            ("deploy", "deployment"),
            ("dominio", "domain"),
            ("aplicacao", "application"),
            ("modulo", "module"),
            ("ficheiro", "file"),
            ("signo", "sinais"),
            ("signo", "sinal"),
            ("signo", "signos"),
        ]
        for p1, p2 in pairs:
            if (w1 == p1 and w2 == p2) or (w1 == p2 and w2 == p1):
                return True
                
        # Substring prefix matching com bloqueio de falsos positivos
        if len(w1) >= 4 and len(w2) >= 4:
            if w1.startswith(w2) or w2.startswith(w1):
                bad_pairs = [("objeto", "objetivo"), ("comunidade", "comunicacao")]
                if not any((w1.startswith(b1) and w2.startswith(b2)) or (w1.startswith(b2) and w2.startswith(b1)) for b1, b2 in bad_pairs):
                    return True
                    
        return False

    # 1. Detetar termos fora de âmbito
    termos_fora_escopo = [
        "deploy", "vercel", "netlify", "docker", "hosting", "hospedar",
        "aws", "amplify", "stripe", "figma", "photoshop", "inpi", "patente",
        "dns", "cpa", "roi", "ads", "google ads", "hostinger", "cloudflare", "mockup"
    ]
    termos_suspeitos = [t for t in termos_fora_escopo if t in pergunta_norm]
    if termos_suspeitos:
        # Se nenhum dos termos suspeitos na pergunta tiver correspondência no contexto
        if not any(any(correspondem(t, c) for c in contexto_norm.split()) for t in termos_suspeitos):
            termos_pergunta = {t for t in pergunta_norm.split() if len(t) >= 4 or t in ["js", "css", "html", "dom", "ads", "roi", "cpa", "pc"]}
            termos_contexto = {t for t in contexto_norm.split() if len(t) >= 3}
            correspondencias = {termo for termo in termos_pergunta if any(correspondem(termo, candidato) for candidato in termos_contexto)}
            cobertura = len(correspondencias) / max(1, len(termos_pergunta))
            marcadores = {
                normalizar_texto(token) for indice, token in enumerate(tokens_originais)
                if len(token) >= 3 and (token.isupper() or any(ch.isupper() for ch in token[1:]) or (indice > 0 and token[0].isupper()))
            }
            presentes = sorted(m for m in marcadores if m and any(correspondem(m, c) for c in termos_contexto))
            return {
                "suficiente": False,
                "razao": "fora_de_ambito",
                "cobertura": round(cobertura, 4),
                "termos": sorted(correspondencias),
                "marcadores_presentes": presentes,
                "marcadores_ausentes": sorted(marcadores - set(presentes)),
            }

    # 2. Análise lexical padrão com plurais/géneros e verbos operacionais
    termos_pergunta = {t for t in pergunta_norm.split() if len(t) >= 4 or t in ["js", "css", "html", "dom", "ads", "roi", "cpa", "pc"]}
    termos_contexto = {t for t in contexto_norm.split() if len(t) >= 3}

    correspondencias = {termo for termo in termos_pergunta if any(correspondem(termo, candidato) for candidato in termos_contexto)}
    cobertura = len(correspondencias) / max(1, len(termos_pergunta))
    
    marcadores = {
        normalizar_texto(token) for indice, token in enumerate(tokens_originais)
        if len(token) >= 3 and (token.isupper() or any(ch.isupper() for ch in token[1:]) or (indice > 0 and token[0].isupper()))
    }
    presentes = sorted(m for m in marcadores if m and any(correspondem(m, c) for c in termos_contexto))
    marcadores_ausentes = sorted(marcadores - set(presentes))

    # Verificar verbos operacionais
    operacionais = set(re.findall(r"\b(?:configur\w*|cri\w*|regist\w*|calcul\w*|anim\w*|export\w*|instal\w*|execut\w*|public\w*|edit\w*)\b", pergunta_norm))
    operacao_suportada = not operacionais or any(any(op[:4] in c for c in termos_contexto) for op in operacionais)

    entidade_especifica_sem_evidencia = bool(marcadores) and not presentes
    cobertura_minima = 0.25 if (presentes and marcadores_ausentes) else 0.15
    suficiente = cobertura >= cobertura_minima and not entidade_especifica_sem_evidencia and operacao_suportada

    return {
        "suficiente": suficiente,
        "razao": "evidencia_recuperada_suficiente" if suficiente else ("entidade_especifica_ausente_no_contexto" if entidade_especifica_sem_evidencia else "cobertura_conceitual_insuficiente"),
        "cobertura": round(cobertura, 4),
        "termos": sorted(correspondencias),
        "marcadores_presentes": presentes,
        "marcadores_ausentes": marcadores_ausentes,
    }

def verificar_necessidade_recusa_fora_escopo(pergunta: str, textos_chunks: list[str]) -> bool:
    return not avaliar_suficiencia_contexto(pergunta, textos_chunks)["suficiente"]