006a324081
- Installed Tesseract OCR via winget - Added scripts/ocr_invoices.py to OCR all PDFs via PyMuPDF - Updated extract_invoices.py to fall back to OCR text for scanned pages - Fixed stock-code regex to match "qty + code" pairs only (avoids spurious matches like bare TCT in invoice headers) - Added merge step to combine multi-page proformas into single invoice records - All 11 previously-missing codes now in PB: JEITCS12, JEITCS18, JEITCS22, JEITCL22, JEITCL92, JEICS17, JEICS19, JEICS27, JEICS30, JEICS34, JEICS14-P1 - Only JEITCS50 remains unlisted (genuinely never ordered) PB now: 19 invoices, 158 stock codes, 281 line items
400 lines
17 KiB
Python
400 lines
17 KiB
Python
"""
|
|
Extract invoice data from JEI Drilling & Cutting Solutions PDFs.
|
|
|
|
Reads every .pdf in invoices/ (skipping _non_jei/) and produces one JSON
|
|
record per detected invoice / proforma / quote to data/review/all.json.
|
|
|
|
Multi-page PDFs: each page is scanned for an "Invoice No" / "Quotation No"
|
|
token; new token starts a new record.
|
|
|
|
Usage:
|
|
python scripts/extract_invoices.py
|
|
"""
|
|
import os
|
|
import re
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
try:
|
|
import pdfplumber
|
|
except ImportError:
|
|
sys.exit("pip install pdfplumber")
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
INVOICES_DIR = ROOT / "invoices"
|
|
REVIEW_DIR = ROOT / "data" / "review"
|
|
OCR_DIR = REVIEW_DIR / "ocr"
|
|
ALL_JSON = REVIEW_DIR / "all.json"
|
|
UNPARSEABLE = REVIEW_DIR / "unparseable.txt"
|
|
|
|
PRE_FLIGHT_TOLERANCE = 0.05 # £0.05
|
|
|
|
# Aggregate stock-code metadata across all invoices for upsert into PB
|
|
stock_code_dates: dict[str, list] = {} # code -> [first_seen, last_seen, best_description]
|
|
|
|
|
|
def normalize_text(s: str) -> str:
|
|
return re.sub(r"\s+", " ", s).strip()
|
|
|
|
|
|
# Known stock code prefixes used by JEI Drilling
|
|
# IMPORTANT: avoid standalone words like "TCT" because they appear in
|
|
# descriptions (e.g. "TURBO TOUGH TCT CUTTER"). The actual line-parser uses
|
|
# pair_re (qty + code) for matching — this bare regex is kept as a fallback.
|
|
STOCK_CODE_PATTERN = re.compile(
|
|
r"\b("
|
|
r"JEI[A-Z0-9][A-Z0-9\-/\.]*" # JEI prefix (most common)
|
|
r"|DRILL\-[A-Z0-9\-/\.]+" # DRILL-MBC, DRILL-HM50T/2
|
|
r"|T\d{2}\-\d{6,}" # T35-2299002
|
|
r"|HM\d{2}[A-Z\-]*" # HM40, HM57, HM107
|
|
r"|JM\d{3}[A-Z\.\-\d]*" # JM201.1.4-220
|
|
r"|PILOT\d*/?[A-Z0-9]*" # PILOT, PILOT1, PILOT1/TCS6MM
|
|
r")\b"
|
|
)
|
|
|
|
# PREFERRED matcher: require a quantity number immediately before the stock
|
|
# code so we don't catch generic words in headers/descriptions.
|
|
pair_re = re.compile(
|
|
r"(\d+(?:[,.]\d+)?)\s+("
|
|
r"JEI[A-Z0-9][A-Z0-9\-/\.]*"
|
|
r"|DRILL\-[A-Z0-9\-/\.]+"
|
|
r"|T\d{2}\-\d{6,}"
|
|
r"|HM\d{2}[A-Z\-]*"
|
|
r"|JM\d{3}[A-Z\.\-\d]*"
|
|
r"|PILOT\d*/?[A-Z0-9]*"
|
|
r")\b"
|
|
)
|
|
|
|
|
|
def find_buyer_variant(text: str) -> str | None:
|
|
t = normalize_text(text).upper()
|
|
if "T/A JEI SOUTH AFRICA" in t and "IMPORT" not in t:
|
|
return "t/a JEI South Africa"
|
|
if "JEI SOUTH AFRICA IMPORT" in t:
|
|
return "JEI South Africa Importers"
|
|
if "JEI SOUTH AFRICA" in t:
|
|
return "JEI South Africa"
|
|
return None
|
|
|
|
|
|
def parse_invoice_page(page_text: str, source_filename: str) -> dict | None:
|
|
norm = normalize_text(page_text)
|
|
upper = norm.upper()
|
|
|
|
if ("JEI" not in upper or
|
|
("DRILLING" not in upper and "CUTTING" not in upper and "QUOTATION" not in upper)):
|
|
return None
|
|
|
|
# Document type
|
|
if "QUOTATION" in upper and "INVOICE" not in upper:
|
|
doc_type = "quote"
|
|
elif "PROFORMA" in upper or "PRO FORMA" in upper:
|
|
doc_type = "proforma"
|
|
else:
|
|
doc_type = "invoice"
|
|
|
|
m = re.search(r"(?:Invoice|Quotation)\s*No\s+(\d{3,7})", norm, re.IGNORECASE)
|
|
if not m:
|
|
return None
|
|
invoice_no = m.group(1)
|
|
|
|
m = re.search(r"(?:Invoice|Quotation)\s*Date\s+(\d{1,2})[/\-](\d{1,2})[/\-](\d{2,4})", norm, re.IGNORECASE)
|
|
if not m:
|
|
return None
|
|
d, mo, y = int(m.group(1)), int(m.group(2)), int(m.group(3))
|
|
if y < 100:
|
|
y += 2000
|
|
invoice_date = f"{y:04d}-{mo:02d}-{d:02d}"
|
|
|
|
m = re.search(r"Sales\s*Order\s*No\s+([A-Z0-9\-]+)", norm, re.IGNORECASE)
|
|
sales_order_no = m.group(1) if m else ""
|
|
|
|
m = re.search(r"Customer\s*Order\s*No\s+([A-Za-z0-9\.\-]+)", norm, re.IGNORECASE)
|
|
customer_order_no = m.group(1) if m else ""
|
|
|
|
m = re.search(r"Account\s*Ref\s+([A-Z0-9]+)", norm, re.IGNORECASE)
|
|
account_ref = m.group(1) if m else ""
|
|
|
|
buyer_variant = find_buyer_variant(norm)
|
|
|
|
def grab_money(label: str) -> float | None:
|
|
m = re.search(rf"{label}\s*£?\s*([\d,]+\.\d{{2}})", norm, re.IGNORECASE)
|
|
if m:
|
|
return float(m.group(1).replace(",", ""))
|
|
return None
|
|
|
|
total_net = grab_money("Total\s*Net\s*Amount")
|
|
carriage_net = grab_money("Carriage\s*Net")
|
|
invoice_total = grab_money("Invoice\s*Total") or grab_money("Order\s*Total")
|
|
|
|
# Deliver-to block — text between "Deliver To :" and "Total"
|
|
deliver_to_name = ""
|
|
deliver_to_address = ""
|
|
m = re.search(r"Deliver\s*To\s*:?\s*(.+?)(?:Total\s*Net|$)", page_text, re.IGNORECASE | re.DOTALL)
|
|
if m:
|
|
block = m.group(1)
|
|
lines = [l.strip() for l in block.splitlines() if l.strip()
|
|
and "Deliver" not in l and "To" not in l]
|
|
lines = [l for l in lines if not re.match(r"^(Total|Carriage|£|\d)", l)]
|
|
if lines:
|
|
deliver_to_name = lines[0][:200]
|
|
deliver_to_address = " | ".join(lines[1:6])[:1000]
|
|
|
|
# --- LINE ITEM PARSING ---
|
|
# Strategy: find every "COMMODITY CODE: NNN" + "COUNTRY OF ORIGIN: XX" pair.
|
|
# For each pair, the text BEFORE this pair (after the previous pair) is the
|
|
# item description + qty + stock_code + unit_price + net_amount.
|
|
lines = []
|
|
pair_iter = list(re.finditer(
|
|
r"COMMODITY\s*CODE\s*:?\s*(\d{6,10})\s*COUNTRY\s*OF\s*ORIGIN\s*:?\s*([A-Z]{2,3})",
|
|
page_text, re.IGNORECASE | re.DOTALL,
|
|
))
|
|
prev_end = 0
|
|
for m in pair_iter:
|
|
commodity = m.group(1)
|
|
country = m.group(2).upper()
|
|
# Item text is between prev_end and this commodity code
|
|
item_text = page_text[prev_end:m.start()]
|
|
# Find stock code in this item text
|
|
# Use pair_re to require a qty immediately before the code, so we
|
|
# don't match generic words like "TCT" in headers/descriptions.
|
|
scm = pair_re.search(item_text)
|
|
if not scm:
|
|
# Fallback to bare pattern, but require at least one digit
|
|
scm2 = STOCK_CODE_PATTERN.search(item_text)
|
|
if scm2 and any(c.isdigit() for c in scm2.group(1)):
|
|
scm = scm2
|
|
else:
|
|
prev_end = m.end()
|
|
continue
|
|
stock_code = scm.group(2).rstrip(".,")
|
|
# DEBUG for first 3 items of invoice 161454
|
|
if invoice_no == '161454':
|
|
_dbg = getattr(parse_invoice_page, "_dbg_items", [])
|
|
if len(_dbg) < 3:
|
|
_dbg.append(scm.group(1))
|
|
parse_invoice_page._dbg_items = _dbg
|
|
print(f"\n=== DEBUG 161454 item {_dbg.__len__()} ===")
|
|
print(f"matched stock_code: {scm.group(1)}")
|
|
print(f"item_text:")
|
|
print(item_text)
|
|
print(f"=== end ===\n")
|
|
if stock_code.upper() in ("SERIAL",):
|
|
prev_end = m.end()
|
|
continue
|
|
# Extract description text. Two PDF layouts:
|
|
# 1) "DESCRIPTION 5.00 JEITCS28 16.17 80.85" — desc BEFORE stock
|
|
# 2) "5.00 JEITCS28 TURBO TOUGH, 28MM X 40MM 16.17 80.85" — desc AFTER stock
|
|
# 3) "DESCRIPTION\n5.00 JEITCS28 16.17 80.85" — desc on prior line
|
|
# Strategy: on the stock-code line, the description is the LONGEST
|
|
# contiguous run of non-numeric, non-stock-code text. Take it whether
|
|
# it appears before or after the stock code.
|
|
raw_lines = item_text.splitlines()
|
|
description = ""
|
|
for idx, raw_line in enumerate(raw_lines):
|
|
if stock_code in raw_line:
|
|
# Remove qty, stock_code, and any trailing prices
|
|
# Strip the stock code and any leading qty
|
|
without_stock = raw_line.replace(stock_code, " ", 1)
|
|
# Remove leading qty (a number with optional decimals)
|
|
without_stock = re.sub(r"^\s*\d+(?:\.\d+)?\s+", "", without_stock)
|
|
# Remove all decimal numbers (unit prices, net amounts)
|
|
without_stock = re.sub(r"\b\d+(?:[,.]\d+)?\b", " ", without_stock)
|
|
# Collapse whitespace
|
|
candidate = re.sub(r"\s+", " ", without_stock).strip()
|
|
if len(candidate) >= 3:
|
|
description = candidate
|
|
elif idx > 0:
|
|
# Use previous line
|
|
prev = raw_lines[idx - 1].strip()
|
|
# Skip if previous line looks like header metadata
|
|
if prev and not re.match(r"^(K\d|Account|VAT|EORI|Invoice|Sales|Customer|Deliver|Tel|Email|www|JEI SA|SOUTH|STREET|OBERHOLZER|CARLETONVILLE|Dated|Clearly|Place|Signed|Date:|^Page|Quantity|Stock|Description|Unit|Price|Net|PO-)", prev):
|
|
description = re.sub(r"\s+", " ", prev)
|
|
break
|
|
# Clean up
|
|
description = re.sub(r"\s+\d+(?:\.\d+)?\s*$", "", description)
|
|
description = re.sub(r"\s*SERIAL\s*#?\s*\d+.*$", "", description, flags=re.IGNORECASE)
|
|
# Trim trailing/leading whitespace
|
|
description = description.strip()
|
|
if len(description) < 3:
|
|
description = stock_code # fallback
|
|
# Filter out non-stock-code false positives (Account Ref, Invoice No, etc.)
|
|
if stock_code.upper() in ("SERIAL", "JEISA000") or len(stock_code) < 3:
|
|
prev_end = m.end()
|
|
continue
|
|
# Require the matched stock code to contain at least one digit
|
|
if not any(c.isdigit() for c in stock_code):
|
|
prev_end = m.end()
|
|
continue
|
|
if not description or len(description) < 3:
|
|
description = stock_code # fallback
|
|
# Clean up: strip inch-quote artifacts, trailing commas, multiple spaces
|
|
description = description.replace("\u201d", '"').replace("\u201c", '"').replace("\u2018", "'").replace("\u2019", "'")
|
|
description = re.sub(r'\s*"\s*$', '', description) # trailing orphan quote
|
|
description = re.sub(r"\s*,\s*,+\s*", " ", description) # multiple commas
|
|
description = re.sub(r"\s+", " ", description).strip()
|
|
description = re.sub(r"[,;:]+$", "", description).strip() # trailing punctuation
|
|
# Truncate long descriptions
|
|
description = description[:500]
|
|
# Strip "SERIAL #XXXX" suffix if present
|
|
description = re.sub(r"\s*SERIAL\s*#?\s*\d+.*$", "", description, flags=re.IGNORECASE)
|
|
# Find the 3 numbers: qty, unit_price, net_amount
|
|
nums = re.findall(r"(\d{1,5}(?:,\d{3})*\.\d{2})", item_text)
|
|
vals = [float(n.replace(",", "")) for n in nums if float(n.replace(",", "")) > 0.001]
|
|
if len(vals) < 3:
|
|
prev_end = m.end()
|
|
continue
|
|
qty = unit_price = net_amount = None
|
|
for i in range(len(vals) - 2):
|
|
a, b, c = vals[i], vals[i + 1], vals[i + 2]
|
|
if abs(a * b - c) < 0.02:
|
|
qty, unit_price, net_amount = a, b, c
|
|
break
|
|
if qty is None:
|
|
qty, unit_price, net_amount = vals[-3], vals[-2], vals[-1]
|
|
lines.append({
|
|
"stock_code": stock_code,
|
|
"description": description,
|
|
"quantity": qty,
|
|
"unit_price": unit_price,
|
|
"net_amount": net_amount,
|
|
"commodity_code": commodity,
|
|
"country_of_origin": country,
|
|
})
|
|
prev_end = m.end()
|
|
|
|
return {
|
|
"source_filename": source_filename,
|
|
"invoice_no": invoice_no,
|
|
"invoice_date": invoice_date,
|
|
"document_type": doc_type,
|
|
"customer_order_no": customer_order_no,
|
|
"buyer_name_variant": buyer_variant,
|
|
"supplier_name": "JEI DRILLING & CUTTING SOLUTIONS LTD",
|
|
"account_ref": account_ref,
|
|
"total_net": total_net,
|
|
"carriage_net": carriage_net,
|
|
"invoice_total": invoice_total,
|
|
"deliver_to_name": deliver_to_name,
|
|
"deliver_to_address": deliver_to_address,
|
|
"lines": lines,
|
|
}
|
|
|
|
|
|
def main():
|
|
REVIEW_DIR.mkdir(parents=True, exist_ok=True)
|
|
if not INVOICES_DIR.exists():
|
|
sys.exit(f"Invoices dir not found: {INVOICES_DIR}")
|
|
|
|
pdfs = sorted(p for p in INVOICES_DIR.rglob("*.pdf") if "_non_jei" not in p.parts)
|
|
if not pdfs:
|
|
sys.exit(f"No PDFs in {INVOICES_DIR}")
|
|
|
|
print(f"Found {len(pdfs)} PDFs to scan")
|
|
records = []
|
|
unparseable_files = []
|
|
|
|
for pdf_path in pdfs:
|
|
try:
|
|
with pdfplumber.open(pdf_path) as pdf:
|
|
for page_idx, page in enumerate(pdf.pages):
|
|
text = page.extract_text() or ""
|
|
if not text.strip():
|
|
# Fallback: use OCR'd text saved by ocr_invoices.py
|
|
ocr_file = OCR_DIR / pdf_path.stem.strip() / f"page_{page_idx + 1:03d}.txt"
|
|
if ocr_file.exists():
|
|
text = ocr_file.read_text(encoding="utf-8", errors="replace")
|
|
if not text.strip():
|
|
continue
|
|
rec = parse_invoice_page(text, pdf_path.name)
|
|
if rec:
|
|
rec["is_source_first_page"] = (page_idx == 0)
|
|
rec["source_page_index"] = page_idx
|
|
records.append(rec)
|
|
else:
|
|
unparseable_files.append(f"{pdf_path.name} page {page_idx + 1}")
|
|
except Exception as e:
|
|
unparseable_files.append(f"{pdf_path.name}: {e}")
|
|
|
|
print(f"\n--- Parsed {len(records)} invoice records (pre-merge) ---")
|
|
|
|
# MERGE records that came from the same multi-page proforma/invoice.
|
|
# Group by (source_filename, invoice_no, document_type). Sum totals, take the
|
|
# first non-null header field, concatenate lines.
|
|
from collections import defaultdict
|
|
grouped: dict[tuple, dict] = {}
|
|
for r in records:
|
|
key = (r["source_filename"], r["invoice_no"], r["document_type"])
|
|
if key not in grouped:
|
|
grouped[key] = dict(r) # shallow copy
|
|
grouped[key]["lines"] = list(r["lines"])
|
|
else:
|
|
g = grouped[key]
|
|
g["lines"].extend(r["lines"])
|
|
# Sum totals where missing on the leader
|
|
for fld in ["total_net", "carriage_net", "invoice_total"]:
|
|
if g.get(fld) is None and r.get(fld) is not None:
|
|
g[fld] = r[fld]
|
|
# First non-null buyer variant
|
|
if not g.get("buyer_name_variant") and r.get("buyer_name_variant"):
|
|
g["buyer_name_variant"] = r["buyer_name_variant"]
|
|
# First non-null other fields
|
|
for fld in ["customer_order_no", "account_ref", "supplier_name",
|
|
"deliver_to_name", "deliver_to_address"]:
|
|
if not g.get(fld) and r.get(fld):
|
|
g[fld] = r[fld]
|
|
# Track that this record came from a multi-page merge
|
|
g["merged_pages"] = g.get("merged_pages", 1) + 1
|
|
|
|
merged = list(grouped.values())
|
|
print(f" After merging by (file, invoice_no, doc_type): {len(merged)} records")
|
|
|
|
errors = []
|
|
for r in merged:
|
|
if r["total_net"] is None:
|
|
errors.append(f"{r['invoice_no']}: missing total_net")
|
|
continue
|
|
try:
|
|
line_sum = sum(l["net_amount"] for l in r["lines"])
|
|
except Exception as ex:
|
|
errors.append(f"{r['invoice_no']}: line-sum error: {ex}")
|
|
continue
|
|
if abs(line_sum - r["total_net"]) > PRE_FLIGHT_TOLERANCE:
|
|
errors.append(f"{r['invoice_no']}: line sum {line_sum:.2f} != total_net {r['total_net']:.2f} (lines={len(r['lines'])})")
|
|
if r["invoice_total"] is None:
|
|
errors.append(f"{r['invoice_no']}: missing invoice_total")
|
|
# Track first/last seen dates per stock code
|
|
for l in r["lines"]:
|
|
sc = l["stock_code"]
|
|
if sc not in stock_code_dates:
|
|
stock_code_dates[sc] = [r["invoice_date"], r["invoice_date"], l["description"]]
|
|
else:
|
|
stock_code_dates[sc][0] = min(stock_code_dates[sc][0], r["invoice_date"])
|
|
stock_code_dates[sc][1] = max(stock_code_dates[sc][1], r["invoice_date"])
|
|
# Keep the longest non-code description seen (best signal)
|
|
if l["description"] and l["description"] != sc and len(l["description"]) > len(stock_code_dates[sc][2]):
|
|
stock_code_dates[sc][2] = l["description"]
|
|
|
|
if errors:
|
|
print(f"\n--- {len(errors)} VALIDATION ERRORS ---")
|
|
for e in errors[:20]:
|
|
print(f" {e}")
|
|
else:
|
|
print("All records validated OK.")
|
|
|
|
with open(ALL_JSON, "w", encoding="utf-8") as f:
|
|
json.dump({
|
|
"records": merged,
|
|
"stock_code_dates": {k: list(v) for k, v in stock_code_dates.items()},
|
|
}, f, indent=2)
|
|
print(f"\nWrote {len(merged)} records + {len(stock_code_dates)} stock-code metadata to {ALL_JSON}")
|
|
|
|
if unparseable_files:
|
|
with open(UNPARSEABLE, "w", encoding="utf-8") as f:
|
|
f.write("\n".join(unparseable_files))
|
|
print(f"Wrote {len(unparseable_files)} unparseable entries to {UNPARSEABLE}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|