"""Single-PDF invoice extractor. Prints JSON to stdout. Reads a single PDF, parses it via the same regex logic as extract_invoices.py, and emits a single JSON record (header + lines) to stdout. Usage: python scripts/extract_one.py Output JSON shape: { "ok": true, "invoice_no": "...", "invoice_date": "YYYY-MM-DD", "buyer_name_variant": "...", "account_ref": "...", "customer_order_no": "...", "sales_order_no": "...", "deliver_to_name": "...", "deliver_to_address": "...", "supplier_name": "...", "total_net": 1470.0, "carriage_net": 550.0, "invoice_total": 2020.0, "document_type": "invoice", "lines": [ {"stock_code": "DRILL-RDE36/2", "description": "...", "quantity": 1, "unit_price": 1385.0, "net_amount": 1385.0, "commodity_code": "84592900", "country_of_origin": "PL"}, ... ], "warnings": [] // any parsing anomalies the user should know about } On parse failure, emits {"ok": false, "error": "..."}. """ import json import re import sys from pathlib import Path # Import the shared parser from extract_invoices.py (same module SJ has been using) sys.path.insert(0, str(Path(__file__).resolve().parent)) from extract_invoices import parse_invoice_page # noqa: E402 import pdfplumber # noqa: E402 def extract(pdf_path: str) -> dict: pdf_path = str(pdf_path) if not Path(pdf_path).is_file(): return {"ok": False, "error": f"File not found: {pdf_path}"} # Combine text from all pages — invoices that span multiple pages need it page_texts = [] try: with pdfplumber.open(pdf_path) as pdf: for i, page in enumerate(pdf.pages): try: txt = page.extract_text() or "" except Exception as e: txt = "" page_texts = None return {"ok": False, "error": f"Failed to read page {i + 1}: {e}"} page_texts.append(txt) except Exception as e: return {"ok": False, "error": f"Failed to open PDF: {e}"} if not page_texts or all(not t.strip() for t in page_texts): return {"ok": False, "error": "No text could be extracted from PDF (scanned? needs OCR via ocr_invoices.py first)."} # Parse each page, then dedupe by (invoice_no, document_type) — same merge logic records = [parse_invoice_page(t, Path(pdf_path).name) for t in page_texts] # Group by (invoice_no, document_type) — multi-page proformas share these merged: dict[tuple, dict] = {} for r in records: key = (r.get("invoice_no", ""), r.get("document_type", "invoice")) if key not in merged: merged[key] = { "invoice_no": r.get("invoice_no"), "invoice_date": r.get("invoice_date"), "buyer_name_variant": r.get("buyer_name_variant"), "account_ref": r.get("account_ref"), "customer_order_no": r.get("customer_order_no"), "sales_order_no": r.get("sales_order_no"), "deliver_to_name": r.get("deliver_to_name"), "deliver_to_address": r.get("deliver_to_address"), "supplier_name": r.get("supplier_name"), "total_net": r.get("total_net"), "carriage_net": r.get("carriage_net"), "invoice_total": r.get("invoice_total"), "document_type": r.get("document_type"), "lines": [], } merged[key]["lines"].extend(r.get("lines", [])) if not merged: return {"ok": False, "error": "No invoice records found in PDF."} # Take the first record (multi-invoice PDFs aren't supported here — use split_invoices.py first) rec = next(iter(merged.values())) # Warnings: anything the user should review warnings = [] line_sum = sum(ln.get("net_amount", 0) or 0 for ln in rec["lines"]) total = rec.get("total_net") or 0 if rec["lines"] and abs(line_sum - total) > 0.05: warnings.append( f"Line net_amount sum ({line_sum:.2f}) doesn't match total_net ({total:.2f}) — review the line items." ) if not rec.get("lines"): warnings.append("No line items were parsed from this PDF.") if not rec.get("invoice_no"): warnings.append("Could not detect invoice_no from PDF.") # Format date as YYYY-MM-DD (PB date field expects this) raw_date = rec.get("invoice_date") or "" if raw_date and len(raw_date) >= 10: iso_date = raw_date[:10] else: iso_date = "" return { "ok": True, "invoice_no": rec.get("invoice_no") or "", "invoice_date": iso_date, "buyer_name_variant": rec.get("buyer_name_variant") or "", "account_ref": rec.get("account_ref") or "", "customer_order_no": rec.get("customer_order_no") or "", "sales_order_no": rec.get("sales_order_no") or "", "deliver_to_name": rec.get("deliver_to_name") or "", "deliver_to_address": rec.get("deliver_to_address") or "", "supplier_name": rec.get("supplier_name") or "", "total_net": rec.get("total_net") or 0, "carriage_net": rec.get("carriage_net") or 0, "invoice_total": rec.get("invoice_total") or 0, "document_type": rec.get("document_type") or "invoice", "lines": [ { "stock_code": ln.get("stock_code") or "", "description": ln.get("description") or "", "quantity": ln.get("quantity") or 0, "unit_price": ln.get("unit_price") or 0, "net_amount": ln.get("net_amount") or 0, "commodity_code": ln.get("commodity_code") or "", "country_of_origin": ln.get("country_of_origin") or "", } for ln in rec["lines"] ], "warnings": warnings, } if __name__ == "__main__": if len(sys.argv) != 2: print(json.dumps({"ok": False, "error": "Usage: python extract_one.py "})) sys.exit(1) try: result = extract(sys.argv[1]) except Exception as e: result = {"ok": False, "error": f"{type(e).__name__}: {e}"} print(json.dumps(result, ensure_ascii=False))