de8aa83ef1
The state file was written before the documents were classified, so a --dry-run marked every message as seen. A new PO seen first by a dry run would then be skipped by every later real run -- never filed, never flagged for review. Now a message is remembered only after it has been classified, and only on a real run; the outcome (filed, already_filed, duplicate_number, needs_review) is stored with it. A dry run leaves the state file untouched, and the report now says how many messages were already seen so an idle hour is not mistaken for a failure.
514 lines
18 KiB
Python
Executable file
514 lines
18 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"}
|
|
|
|
|
|
# --------------------------------------------------------------------------- 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),
|
|
"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
|
|
|
|
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",
|
|
}
|
|
)
|
|
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_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)
|