diff --git a/docs/n8n/po-email-intake.md b/docs/n8n/po-email-intake.md index 95889de..12c83af 100644 --- a/docs/n8n/po-email-intake.md +++ b/docs/n8n/po-email-intake.md @@ -7,6 +7,48 @@ Source: [`po-email-intake.workflow.js`](./po-email-intake.workflow.js) (rebuilda --- +## Hourly check (running now) — `scripts/po_email_intake.py` + +There is a **scheduled task that checks the mailbox every hour**: *MEGA PO email intake (hourly)*, +cron `7 * * * *` Africa/Johannesburg, running in this repo. + +It runs `python3 scripts/po_email_intake.py`, which: + +1. reads `ai@megainfra.co.za` through Mail.app (AppleScript — no Full Disk Access needed); +2. pulls the PDF documents out of any message it has not seen before; +3. reads the PO number, site, supplier and dates out of the document + (`pdftotext -layout`); +4. **checks for duplicates before filing anything** — same file hash at the same site, or the PO + number already in the system; +5. files what is new into the same `purchase_orders` pending queue the browser upload uses, so + the existing MEGA OCR workflow finalises it; +6. flags anything it cannot place, and reports the run. + +State and logs live in `scripts/data/po-email-intake/` (git-ignored): + +| File | What it is | +| --- | --- | +| `state.json` | every message already looked at, so nothing is processed twice | +| `last-run.json` | the last run's outcome, in full | +| `runs.jsonl` | one line per run, appended | +| `sites.json` | MEGA's sites, refreshed daily | + +Run it by hand any time: + +```bash +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 +``` + +It needs the Mac to be awake and Mail.app reachable, and it needs the **Automation** permission +for Mail (granted once — System Settings → Privacy & Security → Automation). + +> The n8n workflow below does the same job server-side, with no dependency on this Mac. The +> script is the route that works today, while the workflow waits on its Microsoft credential. + +--- + ## What it does Every 5 minutes the workflow: diff --git a/scripts/data/po-email-intake/.gitignore b/scripts/data/po-email-intake/.gitignore new file mode 100644 index 0000000..44b2cf9 --- /dev/null +++ b/scripts/data/po-email-intake/.gitignore @@ -0,0 +1,3 @@ +# Runtime state for scripts/po_email_intake.py — local only, never committed. +* +!.gitignore diff --git a/scripts/po_email_intake.py b/scripts/po_email_intake.py new file mode 100755 index 0000000..5ffe3f3 --- /dev/null +++ b/scripts/po_email_intake.py @@ -0,0 +1,489 @@ +#!/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"} + + +# --------------------------------------------------------------------------- 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) + + 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*:") + 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/]+)"), + "text": text, + } + + +# --------------------------------------------------------------------------- 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 resolve_site(text: str, sites: list[dict]) -> dict | None: + """Find the MEGA site named in a PO's subject line.""" + flat = norm(text) + if not flat: + return None + for site in sites: + key = norm(site.get("code")) + if len(key) >= 3 and re.search(rf"\b{re.escape(site.get('code',''))}\b", text, re.I): + return site + for site in sites: + name = norm(site.get("name")) + if len(name) >= 4 and name in flat: + return site + for site in sites: + key = norm(site.get("code")) + if len(key) >= 4 and key in flat: + return site + 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), + "filed": [], + "already_filed": [], + "duplicate_number": [], + "needs_review": [], + "skipped": [], + } + + for msg in messages: + if not args.force and msg["id"] in state["seen"]: + 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) + state["seen"][msg["id"]] = { + "message_id": message_id, + "subject": msg["subject"], + "seen_at": now_iso(), + } + + if not attachments: + report["skipped"].append({"subject": msg["subject"], "reason": "no PDF attachment"}) + 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", + } + ) + 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} + ) + 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", + } + ) + continue + + site = resolve_site(parsed["subject_line"], sites) or resolve_site( + msg["subject"], sites + ) + if not site: + report["needs_review"].append( + { + "po_number": po_number, + "subject": msg["subject"], + "file": filename, + "reason": "no MEGA site found in the PO subject line or the email", + } + ) + 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_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, + } + ) + 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, + } + ) + mark_handled(msg["id"]) + + 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"{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)