SJ/JEI_Import_Documents-App
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: capture stock-code descriptions from PDF line text

Browse source 
- Two PDF layouts supported: description-before-stock-code and description-after
- Clean up trailing inch-quote artifacts and multiple-comma noise
- Importer PATCHes existing records when new description is better
- Stock codes tab now shows real goods descriptions, not just codes
This commit is contained in:
Mavis 2026-09-17 23:49:54 +02:00
parent 22ddfe82bf
commit b18cb13a6c
3 changed files with 206 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

102
scripts/extract_invoices.py
View file

@ -29,6 +29,9 @@ 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()
@ -142,20 +145,79 @@ def parse_invoice_page(page_text: str, source_filename: str) -> dict | None:
if not scm:
prev_end = m.end()
continue
# 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")
stock_code = scm.group(1).rstrip(".,")
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
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
# 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]
@ -163,10 +225,10 @@ def parse_invoice_page(page_text: str, source_filename: str) -> dict | None:
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,
"description": description,
"quantity": qty,
"unit_price": unit_price,
"net_amount": net_amount,
@ -224,16 +286,37 @@ def main():
unparseable_files.append(f"{pdf_path.name}: {e}")
print(f"\n--- Parsed {len(records)} invoice records ---")
# Debug: show types of first few lines from first record
if records:
first = records[0]
print(f"DEBUG first record invoice_no={first.get('invoice_no')} lines_count={len(first.get('lines', []))}")
for i, l in enumerate(first.get("lines", [])[:3]):
print(f" line[{i}] type={type(l).__name__} value={repr(l)[:120]}")
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"])
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 ---")
@ -243,8 +326,11 @@ def main():
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}")
json.dump({
"records": records,
"stock_code_dates": {k: list(v) for k, v in stock_code_dates.items()},
}, f, indent=2)
print(f"\nWrote {len(records)} records + {len(stock_code_dates)} stock-code metadata to {ALL_JSON}")
if unparseable_files:
with open(UNPARSEABLE, "w", encoding="utf-8") as f:

68
scripts/import_to_pb.py
View file

@ -114,17 +114,58 @@ def find_stock_code(code: str):
return None
def upsert_stock_code(rec: dict, dry_run: bool) -> tuple[str, str]:
"""Returns (stock_code_id, action) where action is 'created' or 'exists'."""
def upsert_stock_code(rec: dict, dry_run: bool, sc_meta: dict | None = None) -> tuple[str, str]:
"""Returns (stock_code_id, action) where action is 'created', 'updated', or 'exists'.
sc_meta (optional) provides {first_seen_date, last_seen_date, description_goods}
derived from aggregating across all invoices.
"""
code = rec["stock_code"]
# sc_meta from extract_invoices.py is a list [first_seen, last_seen, description]
if isinstance(sc_meta, list) and len(sc_meta) >= 3:
first_seen = sc_meta[0]
last_seen = sc_meta[1]
description = sc_meta[2] or code
else:
first_seen = None
last_seen = None
description = rec.get("description") or code
existing = find_stock_code(code)
if existing:
# PATCH if any field is better than what's stored
needs_update = False
patch: dict = {}
old_desc = (existing.get("description_goods") or "").strip()
new_desc = (description or "").strip()
# Always update if: new is longer, OR old is just the code, OR old has artifact (trailing quote, etc.)
old_is_default = (old_desc == existing.get("code", "")) or not old_desc
old_has_artifact = old_desc.endswith('"') or old_desc.endswith('"') or " " in old_desc or old_desc.endswith(",")
new_is_better = (
(new_desc and old_is_default) or
(new_desc and len(new_desc) > len(old_desc) and not old_desc.startswith(new_desc[:20])) or
(new_desc and old_has_artifact and len(new_desc) >= len(old_desc) - 2)
)
if new_is_better and new_desc != old_desc:
patch["description_goods"] = new_desc
needs_update = True
if first_seen and not existing.get("first_seen_date"):
patch["first_seen_date"] = first_seen
needs_update = True
if last_seen and not existing.get("last_seen_date"):
patch["last_seen_date"] = last_seen
needs_update = True
if needs_update and not dry_run:
r = pb_req("PATCH", f"/api/collections/JEI_stock_codes/records/{existing['id']}", patch)
if isinstance(r, dict) and r.get("id"):
return r["id"], "updated"
return existing["id"], "exists"
body = {
"code": code,
"description_goods": rec.get("description_goods") or code,
"first_seen_date": rec.get("first_seen_date") or None,
"last_seen_date": rec.get("last_seen_date") or None,
"description_goods": description,
"first_seen_date": first_seen or None,
"last_seen_date": last_seen or None,
}
if dry_run:
return "DRY-" + code, "would-create"
@ -202,16 +243,21 @@ def main():
sys.exit(f"Run extract_invoices.py first. Missing: {ALL_JSON}")
with open(ALL_JSON) as f:
records = json.load(f)
data = json.load(f)
if isinstance(data, dict) and "records" in data:
records = data["records"]
sc_meta_map = data.get("stock_code_dates", {})
else:
records = data
sc_meta_map = {}
print(f"Importing {len(records)} records (dry_run={args.dry_run})")
counts = {"invoices_created": 0, "invoices_exists": 0, "invoices_fail": 0,
"stock_codes_created": 0, "stock_codes_exists": 0,
"stock_codes_created": 0, "stock_codes_updated": 0, "stock_codes_exists": 0,
"lines_created": 0, "lines_fail": 0}
for rec in records:
# Skip records with missing critical fields
if rec.get("total_net") is None or rec.get("invoice_total") is None:
print(f"SKIP {rec['invoice_no']}: missing total")
continue
@ -228,11 +274,13 @@ def main():
counts["invoices_fail"] += 1
continue
# Lines
for idx, line in enumerate(rec["lines"], start=1):
sc_id, sc_action = upsert_stock_code(line, args.dry_run)
sc_meta = sc_meta_map.get(line["stock_code"])
sc_id, sc_action = upsert_stock_code(line, args.dry_run, sc_meta)
if sc_action == "created":
counts["stock_codes_created"] += 1
elif sc_action == "updated":
counts["stock_codes_updated"] += 1
elif sc_action == "exists":
counts["stock_codes_exists"] += 1
if sc_id is None:

54
scripts/price_lookup.py Normal file
View file

@ -0,0 +1,54 @@
"""Price lookup helper — given a list of stock codes + quantities, return total cost."""
import os, sys, json, urllib.request, urllib.parse
PB_URL = "https://pb.agent.apphub.co.za"
PB_TOKEN = None
env_path = os.path.join(os.path.dirname(__file__), "..", ".env.local")
if os.path.exists(env_path):
for line in open(env_path):
if line.startswith("PB_TOKEN="):
PB_TOKEN = line.strip().split("=", 1)[1]
H = {"Authorization": PB_TOKEN}
def get(path):
r = urllib.request.Request(PB_URL + path, headers=H, method="GET")
with urllib.request.urlopen(r, timeout=30) as resp:
return json.loads(resp.read())
# (code, qty) — extracted from user's image
items = [
("JEITCS50", 1),
("JEITCL92", 1),
("JEICS19", 4),
("JEICS34", 1),
("JEICS14-P1", 2),
("JEICS27", 1),
("JEICS30", 2),
("JEICS17", 8),
]
print(f"{'Code':<15} {'Qty':>5} {'Avg£':>8} {'Min£':>8} {'Max£':>8} {'Lines':>6} {'Cost@avg':>10}")
print("-" * 70)
grand = 0.0
for code, qty in items:
f1 = 'code="' + code + '"'
sc = get('/api/collections/JEI_stock_codes/records?filter=' + urllib.parse.quote(f1) + '&perPage=1')
if not sc.get("items"):
print(f"{code:<15} {qty:>5} - NOT IN DB")
continue
sc_id = sc["items"][0]["id"]
f2 = 'stock_code="' + sc_id + '"'
lr = get('/api/collections/JEI_invoice_lines/records?filter=' + urllib.parse.quote(f2) + '&perPage=500')
lines = lr.get("items", [])
if not lines:
print(f"{code:<15} {qty:>5} - NO PURCHASES")
continue
prices = [l["unit_price"] for l in lines if l.get("unit_price")]
avg = sum(prices) / len(prices)
mn = min(prices)
mx = max(prices)
cost = qty * avg
grand += cost
print(f"{code:<15} {qty:>5} {avg:>8.2f} {mn:>8.2f} {mx:>8.2f} {len(lines):>6} {cost:>10.2f}")
print("-" * 70)
print(f"{'TOTAL @ avg':<45} {grand:>24.2f}")