Files
pastor-online/backend/app/corpus.py
T
omp 2043d5ddcf refactor: troca PDF+OCR por corpus USFX em domínio público
- corpus/: Almeida (João Ferreira de Almeida) em USFX, domínio público,
  66 livros / 1189 capítulos / 31098 versículos, versionado com SHA-256
- app/corpus.py: parser USFX (texto no tail de <v/>), nomes canônicos dos
  livros, aliases (abreviaturas e numerais romanos), VerseChunk sem página
- indexer: descobre *.xml em CORPUS_DIR, progresso em versículos, lote
  configurável, manifesto incremental por SHA-256
- remove extraction.py, parallel.py, bible_parser.py, Tesseract e PyMuPDF
- campo page removido de Source/SearchResult/StatusResponse e do frontend
- embedder: propriedade name; status reporta o embedder em uso, não o do .env
- create_app constrói o embedder ANTES do store (o default HashEmbedder do
  VectorStore impedia o modelo configurado de ser carregado em produção)
- fallback do modelo agora aparece em /api/status.errors
- torch 2.6.0 + transformers 5.22.1 pinados (transformers 5.x exige torch>=2.6)
2026-11-02 11:16:07 -03:00

264 lines
9.8 KiB
Python

"""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."""
book: str
chapter: int
verse: int
file: str
text: 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 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) -> Iterator[VerseChunk]:
"""Percorre um arquivo USFX produzindo um chunk por versículo.
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
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"
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
yield VerseChunk(
book=book_name,
chapter=chapter,
verse=verse,
file=file_name,
text=text,
)
if skipped:
logger.info("%s: %s versículo(s) ignorado(s) sem texto", file_name, skipped)