breakpointsoftware/document-parser
0
1from __future__ import annotations2 3import base644import json5import logging6from pathlib import Path7from typing import Any8 9 10logger = logging.getLogger(__name__)11 12 13EXTRACTION_INSTRUCTIONS = (14 "You extract purchase receipt information from a single document. "15 "Return one JSON object that matches the schema exactly. "16 "Use null for missing values, empty arrays for missing line items, and numbers for amounts. "17 "Prefer the receipt's own values over guesses. "18 "Extract only these fields: fecha, fecha_vencimiento, cuit_proveedor, description_proveedor, moneda, subtotal, taxes, total. "19 "If the document is not a purchase receipt, set document_type to unknown but still capture any purchase-related hints you can find. "20 "Do not add markdown, prose, or extra keys."21)22 23 24def build_user_prompt(file_name: str, source_type: str) -> str:25 return (26 f"File name: {file_name}\n\n"27 f"Source type: {source_type}\n"28 "Important rules:\n"29 "- Read the full document before answering.\n"30 "- Copy amounts as numbers, not strings.\n"31 "- Use ISO 8601 for dates when possible.\n"32 "- Extract only: fecha, fecha_vencimiento, cuit_proveedor, description_proveedor, moneda, subtotal, taxes, total.\n"33 "- If a field cannot be found confidently, set it to null.\n"34 "- Keep output compact and do not include extra fields."35 )36 37 38def build_schema() -> dict[str, Any]:39 return {40 "name": "purchase_receipt_extraction",41 "schema": {42 "type": "object",43 "additionalProperties": False,44 "properties": {45 "source_file": {"type": "string"},46 "document_type": {"type": "string", "enum": ["purchase_receipt", "unknown"]},47 "fecha": {"type": ["string", "null"], "description": "ISO 8601 date when possible."},48 "fecha_vencimiento": {"type": ["string", "null"], "description": "ISO 8601 date when possible."},49 "cuit_proveedor": {"type": ["string", "null"]},50 "description_proveedor": {"type": ["string", "null"]},51 "moneda": {"type": ["string", "null"]},52 "subtotal": {"type": ["number", "null"]},53 "taxes": {"type": ["number", "null"]},54 "total": {"type": ["number", "null"]},55 "confidence": {"type": ["number", "null"], "minimum": 0, "maximum": 1},56 },57 "required": [58 "source_file",59 "document_type",60 "fecha",61 "fecha_vencimiento",62 "cuit_proveedor",63 "description_proveedor",64 "moneda",65 "subtotal",66 "taxes",67 "total",68 "confidence",69 ],70 },71 "strict": True,72 }73 74 75def extract_receipt_json(client: Any, model: str, file_path: Path, text: str) -> dict[str, Any]:76 logger.debug("extract_receipt_json called: file=%s, model=%s, text_length=%s", file_path.name, model, len(text))77 78 try:79 logger.debug("Sending text extraction request to OpenAI for %s", file_path.name)80 response = client.responses.create(81 model=model,82 instructions=EXTRACTION_INSTRUCTIONS,83 input=f"{build_user_prompt(file_path.name, 'text')}\n\nDocument text:\n{text[:20000]}",84 text={85 "format": {86 "type": "json_schema",87 "name": "purchase_receipt_extraction",88 "schema": build_schema()["schema"],89 "strict": True,90 }91 },92 )93 94 logger.debug("Received response from OpenAI for %s", file_path.name)95 96 output_text = getattr(response, "output_text", None)97 if output_text:98 logger.debug("Parsing output_text from response")99 result = json.loads(output_text)100 logger.debug("Successfully parsed JSON: document_type=%s, total=%s, confidence=%s", 101 result.get("document_type"), result.get("total"), result.get("confidence"))102 return result103 104 if response.output and response.output[0].content:105 logger.debug("Parsing response.output content")106 result = json.loads(response.output[0].content[0].text)107 logger.debug("Successfully parsed JSON: document_type=%s, total=%s, confidence=%s", 108 result.get("document_type"), result.get("total"), result.get("confidence"))109 return result110 111 logger.error("No JSON returned for %s - no output_text and no response.output", file_path.name)112 raise RuntimeError(f"No JSON returned for {file_path.name}")113 114 except json.JSONDecodeError as e:115 logger.error("JSON decode error for %s: %s", file_path.name, e)116 raise117 except Exception as e:118 logger.error("Error in extract_receipt_json for %s: %s", file_path.name, e)119 raise120 121 122def extract_receipt_json_from_image(client: Any, model: str, file_path: Path, image_data_uri: str) -> dict[str, Any]:123 logger.debug("extract_receipt_json_from_image called: file=%s, model=%s, image_size=%s bytes", 124 file_path.name, model, len(image_data_uri))125 126 try:127 logger.debug("Sending image extraction request to OpenAI for %s", file_path.name)128 response = client.responses.create(129 model=model,130 instructions=EXTRACTION_INSTRUCTIONS,131 input=[132 {133 "role": "user",134 "content": [135 {136 "type": "input_text",137 "text": build_user_prompt(file_path.name, "image"),138 },139 {140 "type": "input_image",141 "image_url": image_data_uri,142 "detail": "high",143 },144 ],145 }146 ],147 text={148 "format": {149 "type": "json_schema",150 "name": "purchase_receipt_extraction",151 "schema": build_schema()["schema"],152 "strict": True,153 }154 },155 )156 157 logger.debug("Received response from OpenAI for image %s", file_path.name)158 159 output_text = getattr(response, "output_text", None)160 if output_text:161 logger.debug("Parsing output_text from image response")162 result = json.loads(output_text)163 logger.debug("Successfully parsed image JSON: document_type=%s, total=%s, confidence=%s", 164 result.get("document_type"), result.get("total"), result.get("confidence"))165 return result166 167 if response.output and response.output[0].content:168 logger.debug("Parsing response.output content from image")169 result = json.loads(response.output[0].content[0].text)170 logger.debug("Successfully parsed image JSON: document_type=%s, total=%s, confidence=%s", 171 result.get("document_type"), result.get("total"), result.get("confidence"))172 return result173 174 logger.error("No JSON returned for image %s - no output_text and no response.output", file_path.name)175 raise RuntimeError(f"No JSON returned for {file_path.name}")176 177 except json.JSONDecodeError as e:178 logger.error("JSON decode error for image %s: %s", file_path.name, e)179 raise180 except Exception as e:181 logger.error("Error in extract_receipt_json_from_image for %s: %s", file_path.name, e)182 raise183 184 185def extract_receipt_json_from_pdf(client: Any, model: str, file_path: Path) -> dict[str, Any]:186 """Upload PDF to OpenAI Files API and extract structured data."""187 logger.debug("extract_receipt_json_from_pdf called: file=%s, model=%s", file_path.name, model)188 189 file_id = None190 try:191 # 1. Upload PDF to Files API192 logger.debug("Uploading PDF to OpenAI Files API: %s", file_path.name)193 with open(file_path, "rb") as pdf_file:194 uploaded_file = client.files.create(195 file=pdf_file,196 purpose="user_data"197 )198 file_id = uploaded_file.id199 logger.debug("PDF uploaded successfully, file_id: %s", file_id)200 201 # 2. Send file_id to model for processing202 logger.debug("Sending PDF to OpenAI via Files API for %s", file_path.name)203 response = client.responses.create(204 model=model,205 instructions=EXTRACTION_INSTRUCTIONS,206 input=[207 {208 "role": "user",209 "content": [210 {211 "type": "input_file",212 "file_id": file_id213 },214 {215 "type": "input_text",216 "text": build_user_prompt(file_path.name, "pdf")217 }218 ]219 }220 ],221 text={222 "format": {223 "type": "json_schema",224 "name": "purchase_receipt_extraction",225 "schema": build_schema()["schema"],226 "strict": True,227 }228 },229 )230 231 logger.debug("Received response from OpenAI for PDF %s", file_path.name)232 233 output_text = getattr(response, "output_text", None)234 if output_text:235 logger.debug("Parsing output_text from PDF response")236 result = json.loads(output_text)237 logger.debug("Successfully parsed PDF JSON: document_type=%s, total=%s, confidence=%s", 238 result.get("document_type"), result.get("total"), result.get("confidence"))239 return result240 241 if response.output and response.output[0].content:242 logger.debug("Parsing response.output content from PDF")243 result = json.loads(response.output[0].content[0].text)244 logger.debug("Successfully parsed PDF JSON: document_type=%s, total=%s, confidence=%s", 245 result.get("document_type"), result.get("total"), result.get("confidence"))246 return result247 248 logger.error("No JSON returned for PDF %s - no output_text and no response.output", file_path.name)249 raise RuntimeError(f"No JSON returned for {file_path.name}")250 251 except json.JSONDecodeError as e:252 logger.error("JSON decode error for PDF %s: %s", file_path.name, e)253 raise254 except Exception as e:255 logger.error("Error in extract_receipt_json_from_pdf for %s: %s", file_path.name, e)256 raise257 finally:258 # Clean up uploaded file259 if file_id:260 try:261 logger.debug("Deleting uploaded file: %s", file_id)262 client.files.delete(file_id)263 except Exception as e:264 logger.warning("Failed to delete file %s: %s", file_id, e)265 266 267def extract_receipt_json_from_document(client: Any, model: str, file_path: Path) -> dict[str, Any]:268 """Send document type to OpenAI by extracting text (DOCX, TXT, etc)."""269 logger.debug("extract_receipt_json_from_document called: file=%s, model=%s", file_path.name, model)270 271 try:272 logger.debug("Extracting text from document: %s", file_path.name)273 274 # Extract text based on file type275 if file_path.suffix.lower() == ".txt":276 text = file_path.read_text(encoding="utf-8", errors="ignore")277 logger.debug("Read TXT file: %s characters", len(text))278 279 elif file_path.suffix.lower() == ".docx":280 from docx import Document281 document = Document(str(file_path))282 paragraphs = [paragraph.text for paragraph in document.paragraphs]283 text = "\n".join(paragraphs)284 logger.debug("Extracted DOCX file: %s characters", len(text))285 286 else:287 logger.warning("Unsupported document format: %s, attempting generic text read", file_path.suffix)288 text = file_path.read_text(encoding="utf-8", errors="ignore")289 logger.debug("Read document file: %s characters", len(text))290 291 # Send extracted text to OpenAI292 logger.debug("Sending extracted document text to OpenAI for %s", file_path.name)293 response = client.responses.create(294 model=model,295 instructions=EXTRACTION_INSTRUCTIONS,296 input=f"{build_user_prompt(file_path.name, 'document')}\n\nDocument text:\n{text[:20000]}",297 text={298 "format": {299 "type": "json_schema",300 "name": "purchase_receipt_extraction",301 "schema": build_schema()["schema"],302 "strict": True,303 }304 },305 )306 307 logger.debug("Received response from OpenAI for document %s", file_path.name)308 309 output_text = getattr(response, "output_text", None)310 if output_text:311 logger.debug("Parsing output_text from document response")312 result = json.loads(output_text)313 logger.debug("Successfully parsed document JSON: document_type=%s, total=%s, confidence=%s", 314 result.get("document_type"), result.get("total"), result.get("confidence"))315 return result316 317 if response.output and response.output[0].content:318 logger.debug("Parsing response.output content from document")319 result = json.loads(response.output[0].content[0].text)320 logger.debug("Successfully parsed document JSON: document_type=%s, total=%s, confidence=%s", 321 result.get("document_type"), result.get("total"), result.get("confidence"))322 return result323 324 logger.error("No JSON returned for document %s - no output_text and no response.output", file_path.name)325 raise RuntimeError(f"No JSON returned for {file_path.name}")326 327 except json.JSONDecodeError as e:328 logger.error("JSON decode error for document %s: %s", file_path.name, e)329 raise330 except Exception as e:331 logger.error("Error in extract_receipt_json_from_document for %s: %s", file_path.name, e)332 raise