From 7a0396422fabc092a3f8dc9683aa20a57f6f2c83 Mon Sep 17 00:00:00 2001 From: omp Date: Sun, 1 Nov 2026 19:24:20 -0300 Subject: [PATCH] feat(backend): add FastAPI routes with SSE chat and reindex guard --- backend/app/config.py | 3 + backend/app/main.py | 337 +++++++++++++++++++++++++++++++++++++ backend/tests/test_main.py | 215 +++++++++++++++++++++++ 3 files changed, 555 insertions(+) create mode 100644 backend/app/main.py create mode 100644 backend/tests/test_main.py diff --git a/backend/app/config.py b/backend/app/config.py index 718058e..2393b80 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -60,6 +60,9 @@ class Settings(BaseSettings): # Limites de ingestão max_pdf_bytes: int = 200 * 1024 * 1024 + # Indexação automática na inicialização + auto_index: bool = True + @property def ocr_cache_dir(self) -> Path: return self.vector_db_dir / "ocr_cache" diff --git a/backend/app/main.py b/backend/app/main.py new file mode 100644 index 0000000..3a13191 --- /dev/null +++ b/backend/app/main.py @@ -0,0 +1,337 @@ +"""API HTTP do Pastor Online. + +Rotas: + +* `GET /api/health` — verificação de saúde +* `GET /api/status` — estado da indexação (PDFs lidos e vetores indexados) +* `POST /api/reindex` — força a reindexação dos PDFs +* `POST /api/chat` — chat em streaming (SSE) +* `GET /api/sources` — últimos trechos recuperados (inspeção) +""" + +from __future__ import annotations + +import asyncio +import json +import logging +from contextlib import asynccontextmanager +from dataclasses import dataclass +from typing import AsyncIterator + +from fastapi import FastAPI, Header, HTTPException, Request +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import StreamingResponse + +from .bible_parser import BibleParser +from .chat import ChatService +from .config import Settings, get_settings +from .embedding import Embedder, HashEmbedder, SentenceTransformerEmbedder +from .extraction import PdfExtractor +from .indexer import Indexer +from .llm import LLMProvider, LLMProviderError, build_provider +from .models import ( + ChatRequest, + HealthResponse, + ReindexResponse, + Source, + SourcesResponse, + StatusResponse, +) +from .vector_store import InMemoryVectorStore, VectorStore + +logger = logging.getLogger(__name__) + + +@dataclass +class Services: + """Dependências de longa duração da aplicação.""" + + settings: Settings + store: VectorStore + embedder: Embedder + provider: LLMProvider + indexer: Indexer + chat: ChatService + last_sources: list[Source] + + +def _build_embedder(settings: Settings) -> Embedder: + """Cria o embedder conforme o modelo configurado. + + `hash` e modelos desconhecidos usam o embedder determinístico; os demais + usam sentence-transformers. + """ + model = settings.embedding_model.strip().lower() + if model in ("", "hash", "hash-test", "none"): + return HashEmbedder() + try: + return SentenceTransformerEmbedder( + settings.embedding_model, batch_size=settings.embedding_batch_size + ) + except Exception as exc: # pragma: no cover - falha de download do modelo + logger.error( + "Não foi possível carregar o modelo de embedding %s: %s. " + "Usando embedder determinístico.", + settings.embedding_model, + exc, + ) + return HashEmbedder() + + +def create_app( + settings: Settings | None = None, + store: VectorStore | None = None, + embedder: Embedder | None = None, + provider: LLMProvider | None = None, + indexer: Indexer | None = None, +) -> FastAPI: + """Cria a aplicação FastAPI. As dependências podem ser injetadas nos testes.""" + settings = settings or get_settings() + + if store is None: + store = VectorStore( + persist_dir=settings.vector_db_dir, + collection_name=settings.collection_name, + embedder=embedder, + ) + if embedder is None: + embedder = getattr(store, "embedder", None) or _build_embedder(settings) + if provider is None: + provider = build_provider(settings) + + if indexer is None: + extractor = PdfExtractor( + cache_dir=settings.ocr_cache_dir, + dpi=settings.ocr_dpi, + ocr_lang=settings.ocr_language, + ocr_timeout=settings.ocr_timeout, + min_legible_ratio=settings.min_legible_ratio, + ) + indexer = Indexer( + settings=settings, + extractor=extractor, + parser=BibleParser(), + embedder=embedder, + store=store, + ) + + chat = ChatService( + settings=settings, store=store, embedder=embedder, provider=provider + ) + + services = Services( + settings=settings, + store=store, + embedder=embedder, + provider=provider, + indexer=indexer, + chat=chat, + last_sources=[], + ) + reindex_lock = asyncio.Lock() + + @asynccontextmanager + async def lifespan(app: FastAPI) -> AsyncIterator[None]: + """Dispara a indexação inicial sem bloquear o startup.""" + if settings.auto_index: + asyncio.create_task(_run_initial_index(services)) + yield + close = getattr(provider, "aclose", None) + if close is not None: + await close() + + app = FastAPI( + title="Pastor Online", + description="Sistema RAG para consulta bíblica com agente pastoral.", + version="1.0.0", + lifespan=lifespan, + ) + app.state.services = services + + app.add_middleware( + CORSMiddleware, + allow_origins=settings.cors_origin_list, + allow_credentials=False, + allow_methods=["GET", "POST"], + allow_headers=["Content-Type", "X-Reindex-Token"], + ) + + # ------------------------------------------------------------------ # + # Rotas + # ------------------------------------------------------------------ # + + @app.get("/api/health", response_model=HealthResponse) + async def health() -> HealthResponse: + """Verificação de saúde do serviço.""" + state = services.indexer.status().state + return HealthResponse( + status="ok", + llm_provider=settings.llm_provider, + indexing="indexing" if state == "indexing" else "ready", + ) + + @app.get("/api/status", response_model=StatusResponse) + async def status() -> StatusResponse: + """Estado da indexação: PDFs lidos e vetores indexados.""" + current = services.indexer.status() + return StatusResponse( + state=current.state, + pdfs_total=current.pdfs_total, + pdfs_indexed=current.pdfs_indexed, + chunks=current.chunks, + model=current.model, + last_indexed_at=current.last_indexed_at, + errors=current.errors, + ) + + @app.post("/api/reindex", response_model=ReindexResponse, status_code=202) + async def reindex( + x_reindex_token: str | None = Header(default=None), + ) -> ReindexResponse: + """Força a releitura e reindexação dos PDFs.""" + if settings.reindex_token: + if x_reindex_token != settings.reindex_token: + raise HTTPException(status_code=401, detail="Token inválido") + + if reindex_lock.locked(): + raise HTTPException( + status_code=409, detail="Uma reindexação já está em andamento" + ) + + # O lock é adquirido e liberado pela tarefa de background: a rota apenas + # agenda o trabalho. Liberá-lo aqui permitiria reindexações concorrentes. + await reindex_lock.acquire() + asyncio.create_task(_run_reindex(services, reindex_lock)) + + return ReindexResponse(state="indexing") + + @app.post("/api/chat") + async def chat(request: ChatRequest) -> StreamingResponse: + """Responde a uma pergunta em streaming (SSE).""" + if len(request.message) > settings.max_message_length: + raise HTTPException(status_code=422, detail="Mensagem muito longa") + + history = [ + {"role": entry.role, "content": entry.content} for entry in request.history + ] + + async def event_stream() -> AsyncIterator[str]: + try: + result = await services.chat.answer(request.message, history) + except LLMProviderError as exc: + logger.error("Falha no provedor de LLM: %s", exc) + yield _sse("error", {"message": "Provedor de LLM indisponível."}) + return + except Exception as exc: # pragma: no cover - erro inesperado + logger.exception("Erro ao processar a pergunta") + yield _sse("error", {"message": "Erro interno ao processar a pergunta."}) + return + + sources = [ + { + "reference": source.reference, + "text": source.text, + "book": source.book, + "chapter": source.chapter, + "verse": source.verse, + "page": source.page, + "file": source.file, + "score": round(source.score, 4), + } + for source in result.sources + ] + services.last_sources = result.sources + + yield _sse("sources", {"sources": sources}) + yield _sse("token", {"text": result.text}) + yield _sse("done", {"citations": result.citations}) + + return StreamingResponse( + event_stream(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", + }, + ) + + @app.get("/api/sources", response_model=SourcesResponse) + async def sources() -> SourcesResponse: + """Trechos consultados na última resposta.""" + return SourcesResponse( + sources=[ + Source( + reference=source.reference, + text=source.text, + book=source.book, + chapter=source.chapter, + verse=source.verse, + page=source.page, + file=source.file, + score=source.score, + ) + for source in services.last_sources + ] + ) + + return app + + +def _sse(event: str, data: dict) -> str: + """Serializa um evento SSE.""" + return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n" + + +async def _run_initial_index(services: Services) -> None: + """Indexa os PDFs na inicialização, sem bloquear o startup.""" + await asyncio.to_thread(services.indexer.reindex) + + +async def _run_reindex(services: Services, lock: asyncio.Lock) -> None: + """Executa a reindexação em background, segurando o lock durante o trabalho.""" + try: + await asyncio.to_thread(services.indexer.reindex, True) + finally: + if lock.locked(): + lock.release() + + +def _build_default_app() -> FastAPI: + """Cria a aplicação padrão, tolerando falha no carregamento do Chroma. + + O Uvicorn importa este módulo no arranque. Se o banco vetorial não puder ser + aberto (volume ainda não montado, por exemplo), a aplicação precisa subir e + reportar o erro em `/api/status`, não recusar o arranque. + """ + try: + return create_app() + except Exception as exc: # pragma: no cover - depende do ambiente + logger.error("Falha ao inicializar a aplicação: %s", exc) + settings = get_settings() + embedder = _build_embedder(settings) + store = InMemoryVectorStore(embedder=embedder) + provider = build_provider(settings) + indexer = Indexer( + settings=settings, + extractor=PdfExtractor( + cache_dir=settings.ocr_cache_dir, + dpi=settings.ocr_dpi, + ocr_lang=settings.ocr_language, + ocr_timeout=settings.ocr_timeout, + min_legible_ratio=settings.min_legible_ratio, + ), + parser=BibleParser(), + embedder=embedder, + store=store, + ) + return create_app( + settings=settings, + store=store, + embedder=embedder, + provider=provider, + indexer=indexer, + ) + + +app = _build_default_app() diff --git a/backend/tests/test_main.py b/backend/tests/test_main.py new file mode 100644 index 0000000..5ca17b2 --- /dev/null +++ b/backend/tests/test_main.py @@ -0,0 +1,215 @@ +"""Testes da API FastAPI.""" + +from __future__ import annotations + +import json +import sys +import threading +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from app.config import Settings # noqa: E402 +from app.embedding import HashEmbedder # noqa: E402 +from app.main import create_app # noqa: E402 +from app.vector_store import InMemoryVectorStore # noqa: E402 + + +class StubProvider: + async def stream(self, messages, system): + yield "Resposta ancorada em Gênesis 1:1." + + async def aclose(self): + return None + + +class FakeIndexer: + """Indexador controlado para os testes de API. + + `release` permite segurar a reindexação em background, de forma que o teste + do 409 seja determinístico. + """ + + def __init__(self, block: bool = False): + self.calls: list[bool] = [] + self.release = threading.Event() + self.release.set() + if block: + self.release.clear() + + def reindex(self, force: bool = False): + self.calls.append(force) + self.release.wait(timeout=10) + from app.indexer import IndexStatus + + return IndexStatus(state="ready", pdfs_total=1, pdfs_indexed=1, chunks=42) + + def status(self): + from app.indexer import IndexStatus + + return IndexStatus(state="ready", pdfs_total=1, pdfs_indexed=1, chunks=42) + + +@pytest.fixture +def settings(tmp_path): + return Settings( + pdf_dir=tmp_path / "pdfs", + vector_db_dir=tmp_path / "vector_db", + embedding_model="hash-test", + llm_provider="stub", + cors_origins="http://localhost:8080", + ) + + +@pytest.fixture +def client(settings): + embedder = HashEmbedder(dimensions=64) + store = InMemoryVectorStore(embedder=embedder) + app = create_app( + settings=settings, + store=store, + embedder=embedder, + provider=StubProvider(), + indexer=FakeIndexer(), + ) + return TestClient(app) + + +def test_health_returns_ok(client): + response = client.get("/api/health") + + assert response.status_code == 200 + body = response.json() + assert body["status"] == "ok" + assert body["llm_provider"] == "stub" + + +def test_status_reports_indexation(client): + response = client.get("/api/status") + + assert response.status_code == 200 + body = response.json() + assert body["state"] == "ready" + assert body["chunks"] == 42 + + +def test_reindex_returns_202(client): + response = client.post("/api/reindex") + + assert response.status_code == 202 + assert response.json()["state"] == "indexing" + + +def test_reindex_requires_token_when_configured(settings): + embedder = HashEmbedder(dimensions=64) + store = InMemoryVectorStore(embedder=embedder) + settings.reindex_token = "segredo" + app = create_app(settings=settings, store=store, embedder=embedder, provider=StubProvider()) + client = TestClient(app) + + assert client.post("/api/reindex").status_code == 401 + + authorized = client.post("/api/reindex", headers={"X-Reindex-Token": "segredo"}) + assert authorized.status_code == 202 + + +def test_concurrent_reindex_is_rejected_with_409(settings): + settings.auto_index = False + embedder = HashEmbedder(dimensions=64) + store = InMemoryVectorStore(embedder=embedder) + indexer = FakeIndexer(block=True) + app = create_app( + settings=settings, + store=store, + embedder=embedder, + provider=StubProvider(), + indexer=indexer, + ) + + # O context manager mantém um único event loop entre as requisições; sem ele, + # o TestClient fecha o loop a cada chamada e a tarefa de background seria + # cancelada, liberando o lock. + with TestClient(app) as client: + try: + first = client.post("/api/reindex") + assert first.status_code == 202 + + second = client.post("/api/reindex") + assert second.status_code == 409 + finally: + indexer.release.set() + + +def test_chat_rejects_empty_message(client): + response = client.post("/api/chat", json={"message": ""}) + + assert response.status_code == 422 + + +def test_chat_rejects_message_above_the_limit(client): + response = client.post("/api/chat", json={"message": "a" * 5000}) + + assert response.status_code == 422 + + +def test_chat_streams_sources_token_and_done(client): + from app.bible_parser import VerseChunk + + embedder = HashEmbedder(dimensions=64) + store = client.app.state.services.store + chunk = VerseChunk( + book="Gênesis", + chapter=1, + verse=1, + page=6, + file="biblia.pdf", + text="No princípio criou Deus os céus e a terra.", + ) + store.upsert([chunk], embedder.encode([chunk.text])) + + response = client.post( + "/api/chat", + json={"message": "Quem criou o mundo?", "history": []}, + ) + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/event-stream") + + body = response.text + assert "event: sources" in body + assert "event: token" in body + assert "event: done" in body + + sources_block = body.split("event: sources")[1].split("event:")[0] + payload = json.loads(sources_block.split("data:")[1].strip()) + assert payload["sources"][0]["reference"] == "Gênesis 1:1" + + +def test_chat_without_evidence_returns_the_fixed_message(client): + response = client.post("/api/chat", json={"message": "fale sobre ações", "history": []}) + + assert response.status_code == 200 + assert "event: token" in response.text + assert "não aborda diretamente" in response.text + + +def test_cors_is_restricted_to_configured_origin(client): + response = client.get("/api/health", headers={"Origin": "http://localhost:8080"}) + + assert response.headers.get("access-control-allow-origin") == "http://localhost:8080" + + +def test_cors_rejects_unknown_origin(client): + response = client.get("/api/health", headers={"Origin": "http://malvado.example"}) + + assert "access-control-allow-origin" not in response.headers + + +def test_sources_endpoint_returns_a_list(client): + response = client.get("/api/sources") + + assert response.status_code == 200 + assert "sources" in response.json()