feat(po-email): hourly mailbox check that files new POs into MEGA
scripts/po_email_intake.py reads ai@megainfra.co.za through Mail.app, pulls the PO documents out of unseen messages, reads the PO number, site and supplier from the document, checks for duplicates by file hash and by PO number, and files what is new into the same purchase_orders pending queue the browser upload uses so the existing MEGA - Purchase Order Ingest workflow finalises it. Anything it cannot place is flagged for review instead of being filed against a guessed site. A scheduled task runs it every hour (7 * * * *, Africa/Johannesburg).
This commit is contained in:
DEls
parent
15b9618122
commit
e6a8f97767
3 changed files with 534 additions and 0 deletions
|
|
@ -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:
|
||||
|
|
|
|||
3
scripts/data/po-email-intake/.gitignore
vendored
Normal file
3
scripts/data/po-email-intake/.gitignore
vendored
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
# Runtime state for scripts/po_email_intake.py — local only, never committed.
|
||||
*
|
||||
!.gitignore
|
||||
489
scripts/po_email_intake.py
Executable file
489
scripts/po_email_intake.py
Executable file
|
|
@ -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)
|
||||
Loading…
Reference in a new issue