Unstructured document data is one of the most common hurdles in enterprise automation. PDFs like invoices, construction estimates, or purchase orders contain critical transactional information, but they are traditionally locked away behind visual layouts that machine learning models can struggle to parse reliably.
When you need precise, deterministic data extraction without the overhead of massive OCR engines, an algorithmic pipeline combining pdfplumber and Regular Expressions (Regex) is the gold standard.
In this tutorial, we will review a production-grade PDF data extraction script built to run as a serverless automation workflow within the Cyberoak platform. This script handles layout complexities, extracts messy line items, buffers dynamic text notes, and outputs a clean, serialized JSON file.
The Automation Stack
To implement this pipeline without heavy external system dependencies, we use a lean Python footprint:
1. pdfplumber: Unlike standard layout tools, pdfplumber offers detailed control over text extraction, allowing us to accurately pull layout-aware strings from complex documents.
2. re (Regular Expressions): A built-in engine used to pinpoint the exact boundaries where a itemized list starts, as well as where metrics like quantities and unit costs live.
3. tempfile & os: Essential safe-handling built-ins to process incoming dynamic streaming objects securely within isolated Cyberoak runner states.
The Production Blueprint
Here is the fully functional automation code designed to ingest an incoming binary stream variable (input_pdf), parse line items conditionally, and save the resulting JSON payload.
preview.py
import pdfplumber
import re
import json
import tempfile
import os
import logging
from decimal import Decimal
# 1. Helper to handle Decimal serialization for JSON
def clean_for_json(data):
if isinstance(data, list):
return [clean_for_json(x) for x in data]
elif isinstance(data, dict):
return {k: clean_for_json(v) for k, v in data.items()}
elif isinstance(data, Decimal):
return str(data)
return data
# 2. Regex Patterns
item_start_regex = re.compile(r"^(\d+)\.\s+")
number_unit_regex = re.compile(
r'([\d,]+(?:\.\d+)?)(?:\s*([A-Za-z]{1,4}))\s+((?:[\d,]+(?:\.\d+)?\s*){1,10})'
)
# 3. Parsing Logic
def extract_line_items_from_text(text):
lines = text.split("\n")
items = []
current_item = None
notes_buffer = []
skip_keywords = ["PAGE:", "EMPIRE CONTRACTORS", "LINH_TRAN_EMPIRE"]
for line in lines:
line = line.strip()
if not line or any(skip in line.upper() for skip in skip_keywords):
continue
if item_start_regex.match(line):
if current_item:
current_item["notes"] = " ".join(notes_buffer).strip()
items.append(current_item)
notes_buffer = []
match = number_unit_regex.search(line)
if not match:
continue
quantity = match.group(1).replace(",", "")
unit = match.group(2) if match.group(2) else ""
numbers_str = match.group(3)
numbers = re.findall(r"[\d,]+(?:\.\d+)?", numbers_str)
numbers = [num.replace(",", "") for num in numbers]
total = numbers[-1] if numbers else "0.00"
unit_price = numbers[0] if numbers else "0.00"
desc_start = item_start_regex.match(line).end()
desc_end = match.start(1)
description = line[desc_start:desc_end].strip()
current_item = {
"description": description,
"unit": unit,
"quantity": quantity,
"unit_price": unit_price,
"total": total,
"notes": ""
}
else:
if current_item:
# Contextual Lookahead: Identify if unnumbered lines belong to the description or notes
if len(line.split()) <= 4 and not any(punct in line for punct in [".", ":", ";", "("]):
current_item["description"] += " " + line.strip()
else:
notes_buffer.append(line.strip())
if current_item:
current_item["notes"] = " ".join(notes_buffer).strip()
items.append(current_item)
return items
def parse_pdf_line_items(file_obj):
all_items = []
temp_file_path = None
try:
# Secure file-streaming inside Cyberoak sandbox
with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as temp_file:
file_obj.seek(0)
temp_file.write(file_obj.read())
temp_file_path = temp_file.name
with pdfplumber.open(temp_file_path) as pdf:
for page in pdf.pages:
text = page.extract_text()
if text:
items = extract_line_items_from_text(text)
if items:
all_items.extend(items)
except Exception as e:
logging.error(f"Error parsing PDF: {str(e)}")
return {"error": str(e)}
finally:
if temp_file_path and os.path.exists(temp_file_path):
os.remove(temp_file_path) # Clean up filesystem environment
return clean_for_json(all_items)
# Cyberoak Execution Gateway
if "input_pdf" in locals() or "input_pdf" in globals():
logging.info("Starting PDF extraction...")
extracted_data = parse_pdf_line_items(input_pdf)
json_output = json.dumps(extracted_data, indent=4)
# Save raw structured artifact to Cyberoak File Storage
file_url = save_file("extracted_line_items.json", json_output)
logging.info(f"Extraction complete. File saved at: {file_url}")
else:
logging.warning("No 'input_pdf' variable found in context.")