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"", "", 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"]