pastor-online/backend/app/corpus.py

"""Leitura do corpus bíblico em USFX.

O texto indexado vem de XML estruturado (USFX), não de PDF: cada versículo já
chega com livro, capítulo e número próprios, então não há OCR, páginas, nem
reconstrução de capítulos por marcadores `[N]`. Isso torna a citação exata por
construção — o que a validação confere é se o LLM citou um versículo que existe
no corpus recuperado.

Formato de entrada (ver `corpus/README.md`):

    <book id="GEN">
      <h>Gênesis</h>
      <c id="1"/>
      <v id="1"/>No princípio criou Deus os céus e a terra.   <ve/>
    </book>

Dois detalhes estruturais importam: `<v/>` é auto-fechado, então o texto do
versículo é o **tail** do elemento (não um filho), e o número do versículo
reinicia a cada `<c/>`.
"""

from __future__ import annotations

import hashlib
import logging
import re
import unicodedata
from dataclasses import dataclass
from pathlib import Path
from typing import Iterator
from xml.etree import ElementTree as ET

logger = logging.getLogger(__name__)


class CorpusParseError(RuntimeError):
    """O corpus não pôde ser lido, ou não contém nenhum livro."""


# Nomes canônicos na ordem e na grafia declaradas pelo próprio corpus. É essa
# grafia que aparece nas citações exibidas ao usuário ("1 Samuel", "Cânticos").
BOOKS: tuple[str, ...] = (
    # Antigo Testamento
    "Gênesis", "Êxodo", "Levítico", "Números", "Deuteronômio", "Josué", "Juízes",
    "Rute", "1 Samuel", "2 Samuel", "1 Reis", "2 Reis", "1 Crônicas",
    "2 Crônicas", "Esdras", "Neemias", "Ester", "Jó", "Salmos", "Provérbios",
    "Eclesiastes", "Cânticos", "Isaías", "Jeremias", "Lamentações", "Ezequiel",
    "Daniel", "Oséias", "Joel", "Amós", "Obadias", "Jonas", "Miquéias", "Naum",
    "Habacuque", "Sofonias", "Ageu", "Zacarias", "Malaquias",
    # Novo Testamento
    "Mateus", "Marcos", "Lucas", "João", "Atos", "Romanos", "1 Coríntios",
    "2 Coríntios", "Gálatas", "Efésios", "Filipenses", "Colossenses",
    "1 Tessalonicenses", "2 Tessalonicenses", "1 Timóteo", "2 Timóteo", "Tito",
    "Filemom", "Hebreus", "Tiago", "1 Pedro", "2 Pedro", "1 João", "2 João",
    "3 João", "Judas", "Apocalipse",
)

# Abreviaturas usuais em português (estilo SBB) → nome canônico.
BOOK_ABBREVIATIONS: dict[str, str] = {
    "Gn": "Gênesis", "Ex": "Êxodo", "Êx": "Êxodo", "Lv": "Levítico",
    "Nm": "Números", "Dt": "Deuteronômio", "Js": "Josué", "Jz": "Juízes",
    "Rt": "Rute", "1Sm": "1 Samuel", "2Sm": "2 Samuel", "1Rs": "1 Reis",
    "2Rs": "2 Reis", "1Cr": "1 Crônicas", "2Cr": "2 Crônicas", "Ed": "Esdras",
    "Ne": "Neemias", "Et": "Ester", "Jó": "Jó", "Sl": "Salmos",
    "Pv": "Provérbios", "Ec": "Eclesiastes", "Ct": "Cânticos", "Is": "Isaías",
    "Jr": "Jeremias", "Lm": "Lamentações", "Ez": "Ezequiel", "Dn": "Daniel",
    "Os": "Oséias", "Jl": "Joel", "Am": "Amós", "Ob": "Obadias", "Jn": "Jonas",
    "Mq": "Miquéias", "Na": "Naum", "Hc": "Habacuque", "Sf": "Sofonias",
    "Ag": "Ageu", "Zc": "Zacarias", "Ml": "Malaquias", "Mt": "Mateus",
    "Mc": "Marcos", "Lc": "Lucas", "Jo": "João", "At": "Atos", "Rm": "Romanos",
    "1Co": "1 Coríntios", "2Co": "2 Coríntios", "Gl": "Gálatas",
    "Ef": "Efésios", "Fp": "Filipenses", "Cl": "Colossenses",
    "1Ts": "1 Tessalonicenses", "2Ts": "2 Tessalonicenses",
    "1Tm": "1 Timóteo", "2Tm": "2 Timóteo", "Tt": "Tito", "Fm": "Filemom",
    "Hb": "Hebreus", "Tg": "Tiago", "1Pe": "1 Pedro", "2Pe": "2 Pedro",
    "1Jo": "1 João", "2Jo": "2 João", "3Jo": "3 João", "Jd": "Judas",
    "Ap": "Apocalipse",
}

# Variantes de nome que a normalização sozinha não resolve.
BOOK_ALIASES: dict[str, str] = {
    "Abdias": "Obadias",
    "Atos dos Apóstolos": "Atos",
    "Cântico dos Cânticos": "Cânticos",
    "Cântico de Salomão": "Cânticos",
    "Cantares": "Cânticos",
    "Cantares de Salomão": "Cânticos",
    "Filemon": "Filemom",
    "Lamentações de Jeremias": "Lamentações",
    "Salmo": "Salmos",
    "Salmos de Davi": "Salmos",
    "Thiago": "Tiago",
    "Apocalipse de João": "Apocalipse",
    "Provérbios de Salomão": "Provérbios",
}

_ROMAN_PREFIXES = {"1": "I", "2": "II", "3": "III"}
_LEADING_NUMBER_RE = re.compile(r"^\s*(\d+)")
_DIGITS_RE = re.compile(r"(\d+)")


def _normalize(text: str) -> str:
    """Remove acentos, colapsa espaços e ignora maiúsculas, para comparação."""
    decomposed = unicodedata.normalize("NFKD", text)
    stripped = "".join(ch for ch in decomposed if not unicodedata.combining(ch))
    return re.sub(r"\s+", " ", stripped).strip().casefold()


def _build_lookup() -> dict[str, str]:
    """Índice `{nome normalizado: nome canônico}` com todas as variantes.

    Cobre o nome exato, as abreviaturas (com e sem espaço após o número) e os
    prefixos em algarismos romanos, que ainda aparecem em citações antigas.
    """
    lookup: dict[str, str] = {}

    for canonical in BOOKS:
        lookup[_normalize(canonical)] = canonical

    for variant, canonical in BOOK_ALIASES.items():
        lookup[_normalize(variant)] = canonical

    for abbreviation, canonical in BOOK_ABBREVIATIONS.items():
        lookup[_normalize(abbreviation)] = canonical
        # "1 Sm" e "1Sm" normalizam diferente; ambos precisam resolver.
        spaced = _DIGITS_RE.sub(r"\1 ", abbreviation, count=1)
        if spaced != abbreviation:
            lookup[_normalize(spaced)] = canonical

    for canonical in BOOKS:
        match = _LEADING_NUMBER_RE.match(canonical)
        if not match:
            continue
        roman = _ROMAN_PREFIXES.get(match.group(1))
        if roman is None:
            continue
        rest = canonical[match.end():].strip()
        lookup[_normalize(f"{roman} {rest}")] = canonical
        lookup[_normalize(f"{roman}{rest}")] = canonical
        abbreviation = BOOK_ABBREVIATIONS.get(
            _DIGITS_RE.sub("", canonical, count=1).strip()
        )
        if abbreviation:
            short = _DIGITS_RE.sub("", abbreviation, count=1).strip()
            lookup[_normalize(f"{roman} {short}")] = canonical
            lookup[_normalize(f"{roman}{short}")] = canonical

    return lookup


_NORMALIZED_BOOKS: dict[str, str] = _build_lookup()


def canonical_book(name: str) -> str | None:
    """Devolve o nome canônico do livro, ou None se não reconhecido."""
    if not name or not name.strip():
        return None
    return _NORMALIZED_BOOKS.get(_normalize(name))


@dataclass(frozen=True)
class VerseChunk:
    """Um versículo indexável com seus metadados de origem.

    `text` é o versículo citado — é o que aparece ao usuário. `context` é a janela
    de versículos vizinhos usada na busca: um versículo isolado é curto demais
    para o embedding captar o tema (medidos: "estou com medo" contra "não temas"
    dava 0,19 de similaridade). A referência continua apontando para o versículo
    central, então a citação exibida não perde precisão.
    """

    book: str
    chapter: int
    verse: int
    file: str
    text: str
    context: str = ""

    @property
    def reference(self) -> str:
        """Referência no formato `Livro Capítulo:Versículo`."""
        return f"{self.book} {self.chapter}:{self.verse}"

    @property
    def embedding_text(self) -> str:
        """Texto usado para embeddar e para a busca por palavras-chave."""
        return self.context or self.text

    @property
    def chunk_id(self) -> str:
        """ID determinístico do chunk no banco vetorial.

        Depende só da coordenada (arquivo, livro, capítulo, versículo) — nunca do
        texto. Assim uma reindexação com o texto revisado mantém a identidade do
        verso e o upsert sobrescreve em vez de duplicar.
        """
        key = f"{self.file}|{self.book}|{self.chapter}|{self.verse}"
        return hashlib.sha1(key.encode("utf-8")).hexdigest()


def normalize_verse(text: str) -> str:
    """Normaliza o texto de um versículo.

    O corpus usa `\\xa0` (espaço não separável) e traz espaços múltiplos antes do
    marcador `<ve/>`. Tudo vira espaço simples, sem pontas soltas.
    """
    return re.sub(r"\s+", " ", text.replace("\xa0", " ")).strip()


def _is_indexable(text: str) -> bool:
    """Versículo com letra ou dígito.

    Algumas edições mantêm versículos colcheteados como um travessão solto; sem
    esse filtro eles viram chunks de ruído puro no índice vetorial.
    """
    return any(ch.isalnum() for ch in text)


def _parse_id(raw: str | None) -> int | None:
    """Extrai o número de um `id`, tolerando sufixos (`3-4`, `4a`)."""
    if not raw:
        return None
    match = _DIGITS_RE.search(raw)
    return int(match.group(1)) if match else None


def parse_usfx(
    path: Path, file_name: str | None = None, context_window: int = 2
) -> Iterator[VerseChunk]:
    """Percorre um arquivo USFX produzindo um chunk por versículo.

    Cada chunk carrega o versículo (`text`) e uma janela de `context_window`
    vizinhos para cada lado (`context`). A janela é montada **dentro do livro** —
    atravessa capítulos, porque a narrativa continua, mas nunca mistura livros.

    Levanta `CorpusParseError` se o arquivo não puder ser lido ou não contiver
    nenhum livro: um corpus vazio indexado em silêncio deixaria o assistente sem
    evidência e respondendo a partir do nada.
    """
    path = Path(path)
    file_name = file_name or path.name

    try:
        tree = ET.parse(path)
    except ET.ParseError as exc:
        raise CorpusParseError(f"{file_name}: XML inválido ({exc})") from exc
    except OSError as exc:
        raise CorpusParseError(f"{file_name}: não foi possível ler ({exc})") from exc

    root = tree.getroot()
    books = root.findall("book")
    if not books:
        raise CorpusParseError(f"{file_name}: nenhum livro encontrado no corpus")

    skipped = 0
    window = max(0, context_window)
    for book in books:
        header = book.find("h")
        book_name = (header.text or "").strip() if header is not None else ""
        if not book_name:
            book_id = book.get("id") or ""
            book_name = canonical_book(book_id) or book_id or "Desconhecido"

        entries: list[tuple[int, int, str]] = []
        chapter = 0
        for element in book:
            if element.tag == "c":
                chapter = _parse_id(element.get("id")) or 0
            elif element.tag == "v":
                verse = _parse_id(element.get("id")) or 0
                text = normalize_verse(element.tail or "")
                if chapter < 1 or verse < 1 or not _is_indexable(text):
                    skipped += 1
                    continue
                entries.append((chapter, verse, text))

        for index, (verse_chapter, verse_number, text) in enumerate(entries):
            if window:
                neighbours = entries[max(0, index - window) : index + window + 1]
                context = " ".join(neighbour[2] for neighbour in neighbours)
            else:
                context = text
            yield VerseChunk(
                book=book_name,
                chapter=verse_chapter,
                verse=verse_number,
                file=file_name,
                text=text,
                context=context,
            )

    if skipped:
        logger.info("%s: %s versículo(s) ignorado(s) sem texto", file_name, skipped)