84 lines
2.9 KiB
Python
Executable File
84 lines
2.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Deterministic IMAP triage runner.
|
|
|
|
Collects unprocessed unread messages via imap_triage.py, marks routine legitimate
|
|
mail processed, moves high-confidence spam, and prints only important items.
|
|
Silent on no actionable items.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
TRIAGE = "/root/.hermes/scripts/imap_triage.py"
|
|
|
|
|
|
def run(args: list[str]) -> tuple[int, str, str]:
|
|
p = subprocess.run([sys.executable, TRIAGE, *args], capture_output=True, text=True, timeout=90)
|
|
return p.returncode, p.stdout, p.stderr
|
|
|
|
|
|
def classify(msg: dict) -> tuple[str, str]:
|
|
sender = (msg.get("from") or "").lower()
|
|
subject = (msg.get("subject") or "").lower()
|
|
hints = set(msg.get("heuristic_hints") or [])
|
|
amounts = msg.get("detected_amounts") or []
|
|
|
|
# User-blocked marketing/loan domains are spam/promotional.
|
|
if "user_blocked_spam_domain" in hints:
|
|
return "suspected_spam", "user-blocked sender domain / promotional offer"
|
|
|
|
# Clear marketing/newsletter categories.
|
|
marketing_senders = ["zillow", "opentable", "upgrade"]
|
|
if any(s in sender for s in marketing_senders):
|
|
if "bank of america" not in sender:
|
|
return "legit", "routine newsletter/marketing notification"
|
|
|
|
# Bank/security/money notifications should surface to Germaine.
|
|
if "bankofamerica.com" in sender or "bank of america" in sender:
|
|
return "bill_or_invoice", "bank/debit card activity notification"
|
|
|
|
if "bill_keywords" in hints and amounts:
|
|
return "bill_or_invoice", "payment/bill-like message with amount"
|
|
|
|
if "spam_keywords" in hints or any(h.startswith("possible_") for h in hints):
|
|
return "suspected_spam", "spam/phishing heuristic matched"
|
|
|
|
return "uncertain", "no deterministic action"
|
|
|
|
|
|
def main() -> int:
|
|
code, out, err = run(["--collect", "--max", "20"])
|
|
if code != 0:
|
|
print(f"IMAP triage collect failed: {err or out}".strip())
|
|
return code or 1
|
|
data = json.loads(out or "{}")
|
|
messages = data.get("messages") or []
|
|
actionable: list[str] = []
|
|
|
|
for msg in messages:
|
|
uid = str(msg.get("uid"))
|
|
decision, reason = classify(msg)
|
|
if not uid:
|
|
continue
|
|
if decision == "suspected_spam":
|
|
run(["--move", uid, "--reason", reason])
|
|
elif decision in ("legit", "uncertain"):
|
|
run(["--mark", uid, decision, reason])
|
|
elif decision == "bill_or_invoice":
|
|
# Mark processed and report once.
|
|
run(["--mark", uid, decision, reason])
|
|
actionable.append(
|
|
f"{msg.get('from','Unknown')} — {msg.get('subject','(no subject)')} — {reason}"
|
|
)
|
|
|
|
if actionable:
|
|
print("Email triage: important item(s) found:\n" + "\n".join(f"- {x}" for x in actionable))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|