MEGAINFRA-WEBAPP/scripts/po_email_intake.py
DEls 4e5d13950a fix(po-email): book a PO against the site name, not the cost code
The project code on a PO's lines is a cost allocation, not a statement of
where the work is going. MEGCPO9550 is a Cedar Gate delivery -- its subject
says "(Cedar Gate)" and its email subject says "RE: Cedar Gate Paving &
Kerbs PO" -- yet four of its seven lines are billed to C26/08 (Castle Gate)
and three to C26/13 (Cedar Gate). Booking by code would have put a Cedar
Gate PO against Castle Gate, and that risk grows as sites are added.

The site is now read from the name written on the PO: its subject line
first, then its description, then the email subject. A PO naming two sites,
or none, is flagged for review instead of being booked against one of them.
The code is still read and kept on the filed row as po_cost_codes, because
the app's own finalisation step uses it as evidence.

MEGA does not always write a site name the way the app has it, so the name
is also matched without its instance qualifier ("Munyaka" for "Munyaka 2")
and through an explicit table of the spellings seen on real POs ("West Park"
for Wespark, MEGCPO9554; "Nellmpius" for Nellmapius Ext 25, MEGCPO9475).
Without those, POs naming a site would have stopped filing themselves.
2026-11-06 10:25:34 +02:00

575 lines
21 KiB
Python
Executable file

#!/usr/bin/env python3
"""Hourly PO email intake for the MEGA mailbox.
Reads ai@megainfra.co.za through Mail.app (AppleScript), finds purchase-order
documents in new messages, checks each one against the purchase_orders collection
for duplicates, and files the ones that are new into the same pending queue the
browser upload uses -- so the "MEGA - Purchase Order Ingest" n8n workflow OCRs and
finalises them.
Safe to run as often as you like:
* every message that has been looked at is remembered in the state file;
* every PO is checked by number and by file hash before anything is created;
* a PO whose site cannot be resolved is never filed -- it is reported for review.
Usage:
python3 scripts/po_email_intake.py # check and file
python3 scripts/po_email_intake.py --dry-run # report only, create nothing
python3 scripts/po_email_intake.py --force # re-check every inbox message
Run it from anywhere; the paths resolve from this file.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import re
import subprocess
import sys
import time
import urllib.parse
import uuid
from datetime import datetime, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
from pb_env import PB, http, superuser_token # noqa: E402
MAIL_ACCOUNT = "ai@megainfra.co.za"
INBOX = "Inbox"
STATE_DIR = Path(__file__).resolve().parent / "data" / "po-email-intake"
STATE_FILE = STATE_DIR / "state.json"
LAST_RUN_FILE = STATE_DIR / "last-run.json"
RUN_LOG = STATE_DIR / "runs.jsonl"
SITES_FILE = STATE_DIR / "sites.json"
PDF_BIN = "/opt/homebrew/bin/pdftotext"
DOC_TYPES = {"application/pdf"}
# A trailing instance qualifier on a site name: "Munyaka 2", "Nellmapius Ext 25".
QUALIFIER = re.compile(r"\s+(?:ext\.?\s*\d*|\d+)$", re.I)
# MEGA does not always write a site's name the way the app has it. These are
# the spellings seen on real POs, mapped to the site they mean:
# "West Park EXT6" -> Wespark (MEGCPO9554, 2026-11-06)
# "Nellmpius EXT25" -> Nellmapius Ext 25 (MEGCPO9475, 2026-11-05)
SITE_ALIASES = {
"WESTPARK": "Wespark",
"NELLMPIUS": "Nellmapius Ext 25",
}
# --------------------------------------------------------------------------- helpers
def now_iso() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.000Z")
def norm(value: str) -> str:
"""Uppercase and strip everything that is not a letter or a digit."""
return re.sub(r"[^A-Z0-9]", "", (value or "").upper())
def osa(script: str, timeout: int = 180) -> str:
"""Run an AppleScript and return stdout."""
proc = subprocess.run(
["osascript", "-e", script],
capture_output=True,
text=True,
timeout=timeout,
)
if proc.returncode != 0:
raise RuntimeError(f"osascript failed: {proc.stderr.strip()[:400]}")
return proc.stdout
def ensure_mail_running() -> None:
running = subprocess.run(
["pgrep", "-x", "Mail"], capture_output=True, text=True
).returncode == 0
if running:
return
subprocess.run(["open", "-a", "Mail"], check=False)
for _ in range(30):
time.sleep(2)
if subprocess.run(["pgrep", "-x", "Mail"], capture_output=True).returncode == 0:
time.sleep(5) # let the Exchange account finish syncing
return
raise RuntimeError("Mail.app did not start")
# --------------------------------------------------------------------------- mail
def inbox_messages() -> list[dict]:
"""List the inbox: id, subject, sender, received date."""
script = f'''
tell application "Mail"
set acct to account "{MAIL_ACCOUNT}"
set out to ""
set n to count of messages of mailbox "{INBOX}" of acct
repeat with i from 1 to n
set m to message i of mailbox "{INBOX}" of acct
set out to out & (message id of m) & "\\t" & ((date received of m) as string) & "\\t" & (sender of m) & "\\t" & (subject of m) & linefeed
end repeat
return out
end tell
'''
raw = osa(script)
messages = []
for line in raw.splitlines():
if not line.strip():
continue
parts = line.split("\t")
if len(parts) < 4:
continue
messages.append(
{
"id": parts[0].strip(),
"received": parts[1].strip(),
"sender": parts[2].strip(),
"subject": parts[3].strip(),
}
)
return messages
def raw_source(message_id: str) -> bytes:
"""Raw RFC822 source of one message (attachments included, base64)."""
script = f'''
tell application "Mail"
set mb to mailbox "{INBOX}" of account "{MAIL_ACCOUNT}"
set m to (first message of mb whose message id is "{message_id}")
return source of m
end tell
'''
return osa(script).encode("utf-8", "replace")
def mark_handled(message_id: str) -> None:
"""Mark the message read and flagged so it reads as handled in Mail and Outlook."""
script = f'''
tell application "Mail"
set mb to mailbox "{INBOX}" of account "{MAIL_ACCOUNT}"
set m to (first message of mb whose message id is "{message_id}")
set read status of m to true
set flagged status of m to true
end tell
'''
try:
osa(script)
except Exception as exc: # noqa: BLE001 - a failed flag must not lose the PO
print(f" ! could not flag message {message_id}: {exc}")
# --------------------------------------------------------------------------- documents
def pdf_attachments(source: bytes) -> list[tuple[str, bytes]]:
"""PDF attachments from a raw MIME message, as (filename, bytes)."""
import email
msg = email.message_from_bytes(source)
out = []
for part in msg.walk():
if part.get_content_type() not in DOC_TYPES:
continue
if part.get_filename() is None and part.get("Content-Disposition") is None:
continue # inline part, not a document
data = part.get_payload(decode=True) or b""
name = part.get_filename() or "document.pdf"
out.append((name, data))
return out
def header_of(source: bytes, name: str) -> str:
import email
msg = email.message_from_bytes(source)
return (msg.get(name) or "").strip()
def parse_pdf(pdf: bytes) -> dict:
"""Pull the header fields out of a MEGA PO document."""
tmp = STATE_DIR / f"tmp-{uuid.uuid4().hex}.pdf"
tmp.write_bytes(pdf)
try:
text = subprocess.run(
[PDF_BIN, "-layout", str(tmp), "-"],
capture_output=True,
text=True,
timeout=120,
).stdout
finally:
tmp.unlink(missing_ok=True)
parsed = parse_po_text(text)
parsed["text"] = text
return parsed
def parse_po_text(text: str) -> dict:
"""Pull the header fields out of the text of a MEGA PO document."""
def grab(pattern: str, flags=re.I | re.S) -> str:
m = re.search(pattern, text, flags)
return " ".join(m.group(1).split()) if m else ""
po_number = grab(r"Purchase Order\(s\)\s*#\s*([A-Za-z0-9\-]+)")
# The subject line carries the site, e.g. "MEGCREQ11928 a CO on MEGCREQ11927 (Nellmapius EXT25)"
subject_segment = grab(r"Subject\s*:\s*(.*?)\s*Description\s*:")
if not subject_segment:
subject_segment = grab(r"Subject\s*:\s*(.*?)\s*To\s*:")
# MEGA's own project reference on each line (e.g. "C26/27, 0001"). Only the
# line-item block is scanned, so a code quoted in the letterhead -- or the
# "Head office" in the contact block -- cannot be mistaken for the site's code.
line_items = text.split("Cost Code", 1)[-1] if "Cost Code" in text else text
cost_codes = sorted({m.group(1) for m in re.finditer(r"\b([A-Z]\d{2}/\d{2})\b", line_items)})
return {
"po_number": po_number.upper(),
"subject_line": subject_segment,
"supplier": grab(r"\bTo\s*:\s*(.*?)\s*(?:PO Creation Date|Code\s*:|Req No)"),
"description": grab(r"Description\s*:\s*(.*?)\s*(?:PO Creation Date|To\s*:|Req No)"),
"order_date": grab(r"PO Creation Date\s*:\s*([0-9/]+)"),
"required_by": grab(r"Required By\s*:\s*([0-9/]+)"),
"cost_codes": cost_codes,
}
# --------------------------------------------------------------------------- sites
def load_sites(token: str) -> list[dict]:
"""MEGA's sites (id, name, project code) — cached for a day."""
if SITES_FILE.exists():
try:
cached = json.loads(SITES_FILE.read_text())
if time.time() - cached.get("fetched_at", 0) < 86400:
return cached["sites"]
except Exception: # noqa: BLE001
pass
code, body = http(
"GET",
"/api/collections/sites/records?perPage=200&fields=id,name,project_code",
token=token,
)
if code != 200:
return []
sites = [
{"id": r.get("id"), "name": r.get("name") or "", "code": r.get("project_code") or ""}
for r in body.get("items", [])
]
SITES_FILE.parent.mkdir(parents=True, exist_ok=True)
SITES_FILE.write_text(json.dumps({"fetched_at": time.time(), "sites": sites}, indent=1))
return sites
def name_variants(name: str) -> list[str]:
"""The site name, and the same name without its instance qualifier.
MEGA writes "Munyaka 2" but a PO may say only "Munyaka", and "Nellmapius
Ext 25" may appear as "Nellmapius". The site is still named.
"""
out = [name.strip()]
stem = name.strip()
while True:
stripped = QUALIFIER.sub("", stem)
if stripped == stem or not stripped:
break
stem = stripped
out.append(stem)
return out
def resolve_site_by_name(text: str, sites: list[dict]) -> dict | None:
"""The MEGA site whose name is written in the text.
This is what a PO is booked against. The project code on the PO's lines is
not used: it is a cost allocation, not a statement of where the work is
going -- MEGCPO9550 is a Cedar Gate delivery with four of its seven lines
billed to Castle Gate's project code. The name is written on the PO
itself, in its subject line or its description.
A PO that names two sites is ambiguous, and is left for a person rather
than booked against one of them.
"""
flat = norm(text)
if not flat:
return None
matched: list[dict] = []
for site in sites:
for variant in name_variants(site.get("name") or ""):
key = norm(variant)
if len(key) >= 4 and key in flat:
matched.append(site)
break
for alias, target in SITE_ALIASES.items():
key = norm(alias)
if len(key) >= 4 and key in flat:
for site in sites:
if site.get("name") == target:
matched.append(site)
break
unique = {site.get("id"): site for site in matched}
if len(unique) == 1:
return next(iter(unique.values()))
return None
# --------------------------------------------------------------------------- pocketbase
def pb_find(token: str, flt: str, fields: str = "") -> list[dict]:
query = {"filter": flt, "perPage": "10"}
if fields:
query["fields"] = fields
code, body = http(
"GET",
"/api/collections/purchase_orders/records?" + urllib.parse.urlencode(query),
token=token,
)
if code != 200:
return []
return body.get("items", [])
def multipart(fields: dict, file_field: str, filename: str, blob: bytes, mime: str) -> tuple[bytes, str]:
boundary = f"----pobound{uuid.uuid4().hex}"
out = bytearray()
for key, value in fields.items():
out += f"--{boundary}\r\n".encode()
out += f'Content-Disposition: form-data; name="{key}"\r\n\r\n'.encode()
out += f"{value}\r\n".encode()
out += f"--{boundary}\r\n".encode()
out += (
f'Content-Disposition: form-data; name="{file_field}"; filename="{filename}"\r\n'
f"Content-Type: {mime}\r\n\r\n"
).encode()
out += blob
out += f"\r\n--{boundary}--\r\n".encode()
return bytes(out), f"multipart/form-data; boundary={boundary}"
def pb_create_po(token: str, pdf: bytes, site_id: str, file_hash: str, meta: dict) -> dict:
number = "PENDING-" + file_hash[:12].upper()
fields = {
"purchase_order_number": number,
"site": site_id,
"processing_status": "pending",
"attempts": "0",
"is_verified": "false",
"file_hash": file_hash,
"date_uploaded": now_iso(),
"raw_ocr_data": json.dumps(meta),
}
body, content_type = multipart(
fields, "purchase_order_file", f"{number}.pdf", pdf, "application/pdf"
)
req = urllib.request.Request(
f"{PB}/api/collections/purchase_orders/records",
data=body,
method="POST",
)
req.add_header("Authorization", token)
req.add_header("Content-Type", content_type)
try:
with urllib.request.urlopen(req, timeout=120) as resp:
return json.loads(resp.read().decode() or "{}")
except urllib.error.HTTPError as exc:
detail = exc.read().decode()[:400]
raise RuntimeError(f"create failed: {exc.code} {detail}") from exc
# --------------------------------------------------------------------------- main
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--dry-run", action="store_true", help="report only, create nothing")
parser.add_argument("--force", action="store_true", help="re-check every inbox message")
args = parser.parse_args()
STATE_DIR.mkdir(parents=True, exist_ok=True)
state = {"seen": {}}
if STATE_FILE.exists():
try:
state = json.loads(STATE_FILE.read_text())
except Exception: # noqa: BLE001
state = {"seen": {}}
state.setdefault("seen", {})
token = superuser_token()
sites = load_sites(token)
ensure_mail_running()
messages = inbox_messages()
report = {
"ran_at": now_iso(),
"dry_run": args.dry_run,
"messages_seen": len(messages),
"already_seen": [],
"filed": [],
"already_filed": [],
"duplicate_number": [],
"needs_review": [],
"skipped": [],
}
for msg in messages:
if not args.force and msg["id"] in state["seen"]:
report["already_seen"].append(msg["subject"])
continue
try:
source = raw_source(msg["id"])
except Exception as exc: # noqa: BLE001
report["skipped"].append({"subject": msg["subject"], "error": str(exc)[:200]})
continue
message_id = header_of(source, "Message-ID")
attachments = pdf_attachments(source)
# The message is only remembered once it has been classified. A dry run must
# not consume it -- otherwise a new PO seen first by --dry-run would be
# skipped by every later real run and never filed, never flagged.
outcomes: list[str] = []
if not attachments:
outcomes.append("no_pdf")
report["skipped"].append({"subject": msg["subject"], "reason": "no PDF attachment"})
if not args.dry_run:
state["seen"][msg["id"]] = {
"message_id": message_id,
"subject": msg["subject"],
"seen_at": now_iso(),
"outcome": outcomes,
}
continue
for filename, blob in attachments:
file_hash = hashlib.sha256(blob).hexdigest()
parsed = parse_pdf(blob)
po_number = parsed["po_number"]
if not po_number:
report["needs_review"].append(
{
"subject": msg["subject"],
"file": filename,
"reason": "no PO number found in the document",
}
)
outcomes.append("needs_review")
continue
# 1. same document, same site -> already filed
same = pb_find(token, f'file_hash="{file_hash}" && site!=""')
if any(r.get("file_hash") == file_hash for r in same):
report["already_filed"].append(
{"po_number": po_number, "subject": msg["subject"], "file": filename}
)
outcomes.append("already_filed")
continue
# 2. same PO number already in the system -> a human must look
clash = pb_find(token, f'purchase_order_number="{po_number}"')
if clash:
report["duplicate_number"].append(
{
"po_number": po_number,
"subject": msg["subject"],
"existing_id": clash[0].get("id"),
"existing_site": clash[0].get("site"),
"reason": "PO number already in the system",
}
)
outcomes.append("duplicate_number")
continue
# The site comes from the name written on the PO: its subject line
# first, then its description, then the email subject. The project
# code on the PO's lines is a cost allocation, not a statement of
# where the work is going, so it never decides the site.
site = (
resolve_site_by_name(parsed["subject_line"], sites)
or resolve_site_by_name(parsed["description"], sites)
or resolve_site_by_name(msg["subject"], sites)
)
if not site:
report["needs_review"].append(
{
"po_number": po_number,
"subject": msg["subject"],
"file": filename,
"reason": "no MEGA site name found on the PO or in the email",
}
)
outcomes.append("needs_review")
continue
meta = {
"source": "mail-intake",
"message_id": message_id,
"subject": msg["subject"],
"from": msg["sender"],
"received": msg["received"],
"site_code": site.get("code"),
"site_name": site.get("name"),
"po_cost_codes": parsed["cost_codes"],
"po_number_hint": po_number,
"file_name": filename,
"backfilled_at": now_iso(),
}
if args.dry_run:
report["filed"].append(
{
"po_number": po_number,
"site": site.get("name"),
"supplier": parsed["supplier"],
"description": parsed["description"],
"value": None,
"dry_run": True,
}
)
outcomes.append("would_file")
continue
created = pb_create_po(token, blob, site["id"], file_hash, meta)
report["filed"].append(
{
"po_number": po_number,
"id": created.get("id"),
"site": site.get("name"),
"supplier": parsed["supplier"],
"description": parsed["description"],
"file": filename,
}
)
outcomes.append("filed")
mark_handled(msg["id"])
if not args.dry_run:
state["seen"][msg["id"]] = {
"message_id": message_id,
"subject": msg["subject"],
"seen_at": now_iso(),
"outcome": outcomes,
}
if not args.dry_run:
STATE_FILE.write_text(json.dumps(state, indent=1))
LAST_RUN_FILE.write_text(json.dumps(report, indent=1))
with RUN_LOG.open("a") as fh:
fh.write(json.dumps({k: v for k, v in report.items() if k != "skipped"}) + "\n")
filed = len(report["filed"])
print(
f"PO intake: {report['messages_seen']} message(s) in inbox | "
f"{len(report['already_seen'])} already seen | "
f"{filed} filed | {len(report['already_filed'])} already filed | "
f"{len(report['duplicate_number'])} duplicate number | "
f"{len(report['needs_review'])} need review"
)
for item in report["filed"]:
print(f" FILED {item['po_number']} -> {item.get('site')} | {item.get('supplier')}")
for item in report["already_filed"]:
print(f" SKIP {item['po_number']} already filed")
for item in report["duplicate_number"]:
print(f" REVIEW {item['po_number']} number already in the system")
for item in report["needs_review"]:
print(f" REVIEW {item.get('po_number', '?')} {item['reason']}")
return 0
if __name__ == "__main__":
try:
sys.exit(main())
except Exception as exc: # noqa: BLE001
print(f"PO INTAKE FAILED: {exc}", file=sys.stderr)
sys.exit(1)