61 lines
2.3 KiB
Python
61 lines
2.3 KiB
Python
"""Orquestra a extração de um arquivo enviado: OpenAI Vision -> fallback local.
|
|
|
|
Reutiliza `lernotafiscal.extraction` (normalização de PDF/imagem, render de
|
|
páginas via PyMuPDF, OCR Tesseract e a detecção heurística) como caminho de
|
|
contingência quando a IA está indisponível ou falha.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from lernotafiscal.extraction import (
|
|
DetectedDocumentCandidate,
|
|
detect_documents,
|
|
normalize_file,
|
|
)
|
|
|
|
from .ai_extraction import RawExtraction, extract_with_ai
|
|
|
|
|
|
def _candidate_to_raw(candidate: DetectedDocumentCandidate) -> RawExtraction:
|
|
fields = candidate.field_confidence or {}
|
|
uncertain = [name for name, level in fields.items() if level == "low"]
|
|
for name, val in (
|
|
("fornecedor", candidate.supplier_name),
|
|
("data_compra", candidate.purchase_date),
|
|
("valor_pago", candidate.total_paid),
|
|
):
|
|
if val is None and name not in uncertain:
|
|
uncertain.append(name)
|
|
legible = candidate.confidence != "low" and len(uncertain) < 2
|
|
return RawExtraction(
|
|
supplier_name=candidate.supplier_name,
|
|
purchase_date_raw=candidate.purchase_date,
|
|
total_paid=candidate.total_paid,
|
|
legible=legible,
|
|
uncertain_fields=uncertain,
|
|
extractor="local",
|
|
raw_text=(candidate.raw_text or "")[:4000],
|
|
)
|
|
|
|
|
|
def extract_file(stored_path: Path, original_name: str) -> list[RawExtraction]:
|
|
"""Retorna os documentos extraídos de um arquivo (>=0). Nunca levanta exceção
|
|
de extração para o chamador — em último caso devolve lista vazia."""
|
|
pages = normalize_file(stored_path, original_name)
|
|
|
|
# Caminho preferencial: enviar as imagens (páginas renderizadas / imagem) à IA.
|
|
# Só aceitamos o resultado da IA quando ela retorna ao menos um documento.
|
|
# Uma resposta vazia (nada encontrado OU shape inesperado) cai no fallback
|
|
# local — evita "zero documentos" silencioso por uma resposta malformada.
|
|
image_paths = [p.image_path for p in pages if p.image_path is not None]
|
|
if image_paths:
|
|
ai_result = extract_with_ai(image_paths)
|
|
if ai_result:
|
|
return ai_result
|
|
|
|
# Fallback: OCR/heurística sobre o texto já normalizado.
|
|
candidates = detect_documents(pages, original_name)
|
|
return [_candidate_to_raw(c) for c in candidates]
|