scripts/desktop_mail_assistant.py — Desktop Mail Assistant

#!/usr/bin/env python3
"""Read and compose messages through the user's configured Mail accounts.

By Zero Width Lab (zerowidthlab.com)
"""

from __future__ import annotations

import argparse
import datetime
import hashlib
import json
import os
import subprocess
import sys
import unicodedata
from pathlib import Path
from typing import Any

__author__ = "Zero Width Lab"

SIGNATURE = "By Zero Width Lab (zerowidthlab.com)"
FAKE_RESPONSES = "DESKTOP_MAIL_FAKE_RESPONSES"
FAKE_ACK = "DESKTOP_MAIL_FAKE_TRANSPORT_TEST_ACK"
TIMEOUT = 45
BODY_BYTES = 400_000
FIXTURE_BYTES = 2_097_152

# This source is constant. Every value supplied by a user is passed after --.
JXA = r'''function run(argv) {
  const app = Application('Mail');
  function fail(message) { throw new Error(message); }
  function account(name) {
    const found = app.accounts().filter(a => a.name() === name);
    if (found.length !== 1) fail('account not found or ambiguous');
    return found[0];
  }
  function mailbox(a, name) {
    const found = a.mailboxes().filter(m => m.name() === name);
    if (found.length !== 1) fail('mailbox not found or ambiguous');
    return found[0];
  }
  function value(fn) { try { return fn(); } catch (_) { return null; } }
  function item(m) { return {id: String(m.id()), subject: String(value(() => m.subject()) || ''), sender: String(value(() => m.sender()) || ''), date: String(value(() => m.dateReceived()) || '')}; }
  const op = argv[0];
  if (op === 'accounts') return JSON.stringify({accounts: app.accounts().map(a => ({name: String(a.name()), addresses: value(() => a.emailAddresses()) || []}))});
  if (op === 'mailboxes') return JSON.stringify({mailboxes: account(argv[1]).mailboxes().map(m => String(m.name()))});
  if (op === 'recent' || op === 'search' || op === 'read') {
    const messages = mailbox(account(argv[1]), argv[2]).messages();
    if (op === 'read') {
      const found = messages.filter(m => String(m.id()) === argv[3]);
      if (found.length !== 1) fail('message id not found or ambiguous');
      const result = item(found[0]); result.body = String(found[0].content());
      return JSON.stringify({message: result});
    }
    const limit = Number(argv[3]);
    const matching = op === 'search' ? messages.filter(m => String(value(() => argv[4] === 'subject' ? m.subject() : m.sender()) || '').toLowerCase().includes(argv[5].toLowerCase())) : messages;
    matching.sort((a, b) => new Date(value(() => b.dateReceived()) || 0).getTime() - new Date(value(() => a.dateReceived()) || 0).getTime());
    const selected = matching.slice(0, limit + 1);
    return JSON.stringify({messages: selected.slice(0, limit).map(item), incomplete: selected.length > limit, skipped: selected.length > limit ? ['additional matching messages'] : []});
  }
  if (op === 'send' || op === 'draft') {
    const payload = JSON.parse(argv[1]);
    const matches = app.accounts().filter(a => (value(() => a.emailAddresses()) || []).some(x => String(x).toLowerCase() === payload.from.toLowerCase()));
    if (matches.length !== 1) fail('sender address does not select exactly one configured account');
    const message = app.OutgoingMessage({subject: payload.subject, content: payload.body, sender: payload.from, visible: false});
    app.outgoingMessages.push(message);
    payload.to.forEach(x => message.toRecipients.push(app.ToRecipient({address: x})));
    payload.cc.forEach(x => message.ccRecipients.push(app.CcRecipient({address: x})));
    payload.bcc.forEach(x => message.bccRecipients.push(app.BccRecipient({address: x})));
    if (op === 'draft') { message.save(); return JSON.stringify({draft_saved: true}); }
    const startedAt = new Date().toISOString();
    const accepted = message.send();
    return JSON.stringify({send_accepted: accepted === true, started_at: startedAt});
  }
  if (op === 'verify') {
    const payload = JSON.parse(argv[1]);
    const matches = app.accounts().filter(a => (value(() => a.emailAddresses()) || []).some(x => String(x).toLowerCase() === payload.from.toLowerCase()));
    if (matches.length !== 1) fail('sender account unavailable for verification');
    const sent = matches[0].sentMailbox();
    if (!sent) fail('Sent mailbox unavailable');
    function addresses(recipients) { return recipients.map(r => String(r.address()).toLowerCase()).sort().join('|'); }
    const found = sent.messages().slice(0, 100).filter(m =>
      String(value(() => m.subject()) || '') === payload.subject &&
      String(m.content()) === payload.body &&
      String(value(() => m.sender()) || '').toLowerCase().includes(payload.from.toLowerCase()) &&
      addresses(m.toRecipients()) === payload.to.map(x => x.toLowerCase()).sort().join('|') &&
      addresses(m.ccRecipients()) === payload.cc.map(x => x.toLowerCase()).sort().join('|') &&
      addresses(m.bccRecipients()) === payload.bcc.map(x => x.toLowerCase()).sort().join('|') &&
      new Date(value(() => m.dateSent()) || 0).getTime() >= new Date(payload.started_at).getTime() - 5000);
    return JSON.stringify({verified: found.length > 0, matching_sent_messages: found.length});
  }
  fail('unsupported operation');
}'''


class CliError(Exception):
    """A controlled CLI failure."""


def display_text(value: Any) -> str:
    parts = []
    for character in str(value):
        code = ord(character)
        if unicodedata.category(character).startswith("C") or code in (0x2028, 0x2029):
            parts.append(f"\\u{code:04x}" if code <= 0xFFFF else f"\\U{code:08x}")
        else:
            parts.append(character)
    return "".join(parts)


def bounded_int(maximum: int):
    def parse(value: str) -> int:
        try:
            number = int(value)
        except ValueError as exc:
            raise argparse.ArgumentTypeError("must be an integer") from exc
        if not 1 <= number <= maximum:
            raise argparse.ArgumentTypeError(f"must be between 1 and {maximum}")
        return number
    return parse


def require_platform() -> None:
    if sys.platform != "darwin":
        raise CliError("Desktop Mail Assistant supports macOS only; no file or Mail I/O was performed")


def bounded_utf8(path: str, maximum: int, label: str) -> str:
    require_platform()
    try:
        with Path(path).open("rb") as stream:
            data = stream.read(maximum + 1)
        if len(data) > maximum:
            raise CliError(f"{label} exceeds {maximum} bytes")
        return data.decode("utf-8")
    except (OSError, UnicodeError) as exc:
        raise CliError(f"cannot read {label}: {exc}") from exc


def strict_json(text: str) -> Any:
    def reject_constant(value: str) -> None:
        raise ValueError(f"non-finite JSON constant {value} is not permitted")
    return json.loads(text, parse_constant=reject_constant)


def validate_response(operation: str, response: Any) -> dict[str, Any]:
    def fail() -> None:
        raise CliError(f"Mail returned malformed {operation} response")
    def strings(value: Any) -> bool:
        return isinstance(value, list) and all(isinstance(item, str) for item in value)
    def message(value: Any, body: bool = False) -> bool:
        fields = ("id", "subject", "sender") + (("body",) if body else ("date",))
        return isinstance(value, dict) and all(isinstance(value.get(key), str) for key in fields)
    if not isinstance(response, dict):
        fail()
    if operation == "accounts":
        accounts = response.get("accounts")
        if not isinstance(accounts, list) or not all(isinstance(a, dict) and isinstance(a.get("name"), str) and strings(a.get("addresses")) for a in accounts):
            fail()
    elif operation == "mailboxes":
        if not strings(response.get("mailboxes")):
            fail()
    elif operation in ("recent", "search"):
        messages = response.get("messages")
        if not isinstance(messages, list) or len(messages) > 100 or not all(message(m) for m in messages) or type(response.get("incomplete")) is not bool or not strings(response.get("skipped")) or bool(response["skipped"]) != response["incomplete"]:
            fail()
    elif operation == "read":
        if not message(response.get("message"), body=True):
            fail()
    elif operation == "draft":
        if type(response.get("draft_saved")) is not bool:
            fail()
    elif operation == "send":
        if type(response.get("send_accepted")) is not bool:
            fail()
        try:
            stamp = datetime.datetime.fromisoformat(response["started_at"].replace("Z", "+00:00"))
            if stamp.tzinfo is None:
                fail()
        except (KeyError, AttributeError, TypeError, ValueError):
            fail()
    elif operation == "verify":
        count = response.get("matching_sent_messages")
        if type(response.get("verified")) is not bool or type(count) is not int or not 0 <= count <= 100 or response["verified"] != (count > 0):
            fail()
    else:
        fail()
    return response


def mail_call(operation: str, *values: str) -> dict[str, Any]:
    require_platform()
    fake_path = os.environ.get(FAKE_RESPONSES)
    fake_ack = os.environ.get(FAKE_ACK)
    if fake_path or fake_ack:
        if not fake_path or fake_ack != "1":
            raise CliError("fake transport requires both test environment variables")
        print("warning: synthetic osascript fixture; Mail was not contacted", file=sys.stderr)
        try:
            fixture = strict_json(bounded_utf8(fake_path, FIXTURE_BYTES, "synthetic fixture"))
            response = fixture[operation]
        except (OSError, ValueError, KeyError, TypeError, RecursionError) as exc:
            raise CliError(f"missing or invalid fake response for {operation}: {exc}") from exc
        return validate_response(operation, response)
    command = ["/usr/bin/osascript", "-l", "JavaScript", "-e", JXA, "--", operation, *values]
    try:
        completed = subprocess.run(command, capture_output=True, text=True, timeout=TIMEOUT, check=False)
    except subprocess.TimeoutExpired as exc:
        raise CliError("Mail automation timed out; the first macOS Automation permission prompt may still be open") from exc
    except (OSError, UnicodeError) as exc:
        raise CliError(f"could not launch or decode osascript response: {exc}") from exc
    if completed.returncode:
        raise CliError(f"Mail automation failed: {display_text(completed.stderr.strip())}")
    try:
        response = strict_json(completed.stdout)
    except (ValueError, RecursionError) as exc:
        raise CliError("Mail automation returned invalid JSON") from exc
    return validate_response(operation, response)


def body_from_file(path: str) -> str:
    body = bounded_utf8(path, BODY_BYTES, "body file")
    if len(body) > 100_000:
        raise CliError("body file exceeds 100000 characters")
    return body


def message_plan(args: argparse.Namespace) -> dict[str, Any]:
    if not args.from_address or not args.to or not args.subject:
        raise CliError("sender, recipient and subject are required")
    return {"operation": args.command, "from": args.from_address, "to": args.to,
            "cc": args.cc or [], "bcc": args.bcc or [], "subject": args.subject,
            "body": body_from_file(args.body_file)}


def digest(plan: dict[str, Any]) -> str:
    encoded = json.dumps(plan, sort_keys=True, ensure_ascii=True, separators=(",", ":")).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def print_message(plan: dict[str, Any], approval: str) -> None:
    for key in ("operation", "from", "to", "cc", "bcc", "subject", "body"):
        value = plan[key]
        if isinstance(value, list):
            value = ", ".join(value)
        print(f"{key}: {display_text(value)}")
    print(f"approval_digest: {approval}")


def run(args: argparse.Namespace) -> int:
    require_platform()
    if args.command in ("send", "draft"):
        plan = message_plan(args)
        approval = digest(plan)
        if args.dry_run:
            print_message(plan, approval)
            return 0
        if args.approve != approval:
            raise CliError("live write requires the matching --approve digest from --dry-run")
        try:
            result = mail_call(args.command, json.dumps(plan, ensure_ascii=True))
        except CliError as exc:
            print(f"incomplete: Mail write outcome is uncertain: {display_text(exc)}; skipped verification; inspect Mail before retrying", file=sys.stderr)
            return 2
        if args.command == "draft":
            if result.get("draft_saved") is not True:
                raise CliError("Mail did not confirm draft save")
            print("draft_saved: true")
            return 0
        if result.get("send_accepted") is not True:
            raise CliError("Mail did not confirm send acceptance")
        started_at = result.get("started_at")
        if not isinstance(started_at, str) or not started_at:
            print("incomplete: Mail accepted send but gave no verification timestamp; skipped Sent confirmation; do not retry blindly", file=sys.stderr)
            return 2
        try:
            verification = mail_call("verify", json.dumps({**plan, "started_at": started_at}, ensure_ascii=True))
        except CliError as exc:
            print(f"incomplete: Sent verification failed: {display_text(exc)}; skipped confirmation; inspect Mail before retrying", file=sys.stderr)
            return 2
        if verification.get("verified") is not True:
            print("incomplete: Mail accepted send but no matching message was found in Sent; skipped confirmed-delivery claim; do not retry blindly", file=sys.stderr)
            return 2
        print("sent_verified: true")
        return 0
    if args.command == "accounts":
        result = mail_call("accounts")
    elif args.command == "mailboxes":
        result = mail_call("mailboxes", args.account)
    elif args.command in ("recent", "search"):
        extra = (args.search_field, args.search_text) if args.command == "search" else ()
        result = mail_call(args.command, args.account, args.mailbox, str(args.limit), *extra)
    else:
        result = mail_call("read", args.account, args.mailbox, args.message_id)
        message = result.get("message")
        if not isinstance(message, dict) or not isinstance(message.get("body"), str):
            raise CliError("Mail returned no readable message body")
        body = message["body"]
        if len(body) > args.max_body_chars:
            message["body"] = body[:args.max_body_chars]
            result["incomplete"] = True
            result["skipped"] = [f"{len(body) - args.max_body_chars} body characters"]
    for key, value in result.items():
        print(f"{key}: {display_text(json.dumps(value, ensure_ascii=True))}")
    if result.get("incomplete") and not getattr(args, "allow_partial", False):
        return 2
    return 0


def parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(description="Use Apple's Mail app through macOS Automation.", epilog=SIGNATURE)
    sub = p.add_subparsers(dest="command", required=True)
    sub.add_parser("accounts", epilog=SIGNATURE)
    boxes = sub.add_parser("mailboxes", epilog=SIGNATURE)
    boxes.add_argument("--account", required=True)
    for name in ("recent", "search", "read"):
        command = sub.add_parser(name, epilog=SIGNATURE)
        command.add_argument("--account", required=True)
        command.add_argument("--mailbox", required=True)
        command.add_argument("--allow-partial", action="store_true")
        if name in ("recent", "search"):
            command.add_argument("--limit", type=bounded_int(100), default=20)
        if name == "search":
            group = command.add_mutually_exclusive_group(required=True)
            group.add_argument("--subject")
            group.add_argument("--sender")
        if name == "read":
            command.add_argument("--message-id", required=True)
            command.add_argument("--max-body-chars", type=bounded_int(100_000), default=10_000)
    for name in ("send", "draft"):
        command = sub.add_parser(name, epilog=SIGNATURE)
        command.add_argument("--from", dest="from_address", required=True)
        command.add_argument("--to", action="append", required=True)
        command.add_argument("--cc", action="append")
        command.add_argument("--bcc", action="append")
        command.add_argument("--subject", required=True)
        command.add_argument("--body-file", required=True)
        command.add_argument("--dry-run", action="store_true")
        command.add_argument("--approve")
    return p


def main(argv: list[str] | None = None) -> int:
    try:
        args = parser().parse_args(argv)
        if args.command == "search":
            args.search_field = "subject" if args.subject is not None else "sender"
            args.search_text = args.subject if args.subject is not None else args.sender
        return run(args)
    except CliError as exc:
        print(f"error: {display_text(exc)}", file=sys.stderr)
        return 1
    except KeyboardInterrupt:
        print("error: interrupted", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

License: MIT (read it).

Back to the listing