Spaces:
Sleeping
Sleeping
File size: 2,052 Bytes
db4ba8d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | """
TradeFlow AI — Preprocessing Node (Step 2.1)
"""
import structlog
from ...services.ingest_svc import get_storage_service
from ...services.ocr_engine_svc import ocr_engine_service
from ..state import ExtractionGraphState
log = structlog.get_logger()
async def preprocess_documents_node(state: ExtractionGraphState) -> dict:
"""
Step 2.1: Document Preprocessing Node
- Checks document quality
- Converts PDFs to images if necessary
- Sets quality score
"""
log.info("Running preprocess_documents_node", batch_id=state["batch_id"])
updated_docs = []
for doc in state["documents"]:
storage_path = doc.get("storage_path")
if not storage_path:
updated_docs.append({
**doc,
"quality_score": 0.0,
"pages": [],
"ocr_candidates": {},
"error": "Document missing storage_path",
})
continue
try:
file_bytes = await get_storage_service().download_document(storage_path)
prepared = await ocr_engine_service.prepare_document(
doc_id=doc["doc_id"],
storage_path=storage_path,
filename=doc.get("original_name") or storage_path,
file_bytes=file_bytes,
)
except Exception as exc:
log.exception(
"Document preprocessing/OCR failed",
batch_id=state["batch_id"],
doc_id=doc.get("doc_id"),
error=str(exc),
)
updated_docs.append({
**doc,
"quality_score": 0.0,
"pages": [],
"ocr_candidates": {},
"error": str(exc),
})
continue
updated_docs.append({
**doc,
**prepared,
"ocr_method": "+".join(prepared["ocr_candidates"].keys()) or None,
})
return {
"documents": updated_docs,
"steps": ["preprocess"]
}
|