22854074ca
- Next.js 14 + Tailwind + TypeScript scaffolded - PocketBase v0.39.8 collections: JEI_invoices, JEI_invoice_lines, JEI_stock_codes (30 fields total, locked to superuser) - PDF extraction: scripts/extract_invoices.py parses 33 PDFs -> 38 records - Import script: scripts/import_to_pb.py idempotent push to PB - Imported: 24 invoices, 70+ stock codes, 118 invoice lines - Pages: /, /invoices, /invoices/[id], /stock-codes, /stock-codes/[code] - All 5 pages verified returning 200 with real data on port 6002
256 lines
9 KiB
Python
256 lines
9 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"
|
|
ALL_JSON = REVIEW_DIR / "all.json"
|
|
UNPARSEABLE = REVIEW_DIR / "unparseable.txt"
|
|
|
|
PRE_FLIGHT_TOLERANCE = 0.05 # £0.05
|
|
|
|
|
|
def normalize_text(s: str) -> str:
|
|
return re.sub(r"\s+", " ", s).strip()
|
|
|
|
|
|
# Known stock code prefixes used by JEI Drilling
|
|
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"|TCT\d*" # TCT variants
|
|
r"|PILOT\d*/?[A-Z0-9]*" # PILOT, PILOT1, PILOT1/TCS6MM
|
|
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
|
|
scm = STOCK_CODE_PATTERN.search(item_text)
|
|
if not scm:
|
|
prev_end = m.end()
|
|
continue
|
|
stock_code = scm.group(1).rstrip(".,")
|
|
if stock_code.upper() in ("SERIAL",):
|
|
prev_end = m.end()
|
|
continue
|
|
# Find the 3 numbers: qty, unit_price, net_amount
|
|
# They appear at the END of the item text, after the description
|
|
# Pattern: qty stock_code ... unit_price net_amount
|
|
# OR: stock_code ... 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
|
|
# Try combinations of last 3 numbers
|
|
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:
|
|
# Last 3 fallback
|
|
qty, unit_price, net_amount = vals[-3], vals[-2], vals[-1]
|
|
lines.append({
|
|
"stock_code": stock_code,
|
|
"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():
|
|
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 ---")
|
|
errors = []
|
|
for r in records:
|
|
if r["total_net"] is None:
|
|
errors.append(f"{r['invoice_no']}: missing total_net")
|
|
continue
|
|
line_sum = sum(l["net_amount"] for l in r["lines"])
|
|
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")
|
|
|
|
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, f, indent=2)
|
|
print(f"\nWrote {len(records)} records 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()
|