nixos-config/hosts/pinix/localServices/finance-agent/mailtool.py

#!/usr/bin/env python3
"""Strictly read-only IMAP helper for the finance agent -- search a mailbox
folder for messages with attachments, fetch one specific attachment to
disk, or list available folders. Never marks messages as read/seen, never
deletes or moves anything, never sends mail. The mailbox is opened in IMAP
read-only mode so even a bug here can't change server-side state.
"""
import argparse
import email
import imaplib
import os
import sys
from email.header import decode_header
from pathlib import Path


def login():
    host = os.environ["FINANCE_MAIL_HOST"]
    user = os.environ["FINANCE_MAIL_USER"]
    password = Path(os.environ["FINANCE_MAIL_PASSWORD_FILE"]).read_text(
        encoding="utf-8"
    ).strip()
    connection = imaplib.IMAP4_SSL(host)
    connection.login(user, password)
    return connection


def connect(folder="INBOX"):
    connection = login()
    # readonly=True: IMAP SELECT (not EXAMINE) is what marks messages seen;
    # readonly mode uses EXAMINE instead, so browsing never changes server
    # state.
    connection.select(f'"{folder}"', readonly=True)
    return connection


def decode_value(raw):
    if raw is None:
        return ""
    parts = decode_header(raw)
    decoded = []
    for text, charset in parts:
        if isinstance(text, bytes):
            decoded.append(text.decode(charset or "utf-8", errors="replace"))
        else:
            decoded.append(text)
    # Long headers (e.g. filenames) can arrive folded across lines per RFC
    # 2822/5322; collapse that back to plain whitespace instead of leaking
    # a literal newline into search output.
    return " ".join("".join(decoded).split())


# Real invoice/receipt PDFs are often sent with Content-Disposition: inline
# rather than attachment (confirmed live 2026-08-01 -- a filter on
# disposition=="attachment" missed genuine invoices like "RE 6:26, Rahel
# Jacob .pdf"), so any named part is a candidate regardless of disposition.
# Content-ID is the reliable signal instead: parts referenced from an HTML
# body (cid:...), e.g. an embedded signature logo, always carry one; real
# user-sent documents don't. Shared by search and fetch so they can't drift
# apart the way disposition-based filtering already did once.
def document_parts(message):
    for part in message.walk():
        if part.get("Content-ID"):
            continue
        filename = part.get_filename()
        if filename:
            yield decode_value(filename), part


def attachment_names(message):
    return [name for name, _part in document_parts(message)]


def cmd_folders(args):
    connection = login()
    try:
        status, folders = connection.list()
        if status != "OK":
            print(f"IMAP-LIST fehlgeschlagen: {status}", file=sys.stderr)
            return 1
        for entry in folders:
            print(entry.decode(errors="replace"))
        return 0
    finally:
        connection.logout()


def cmd_search(args):
    connection = connect(args.folder)
    try:
        since = (
            __import__("datetime")
            .date.today()
            .fromordinal(__import__("datetime").date.today().toordinal() - args.days)
        )
        criterion = f'(SINCE "{since.strftime("%d-%b-%Y")}")'
        status, data = connection.search(None, criterion)
        if status != "OK":
            print(f"IMAP-Suche fehlgeschlagen: {status}", file=sys.stderr)
            return 1
        uids = data[0].split()
        # Newest first, capped -- keeps a single reply well under the chat
        # length budget even on a mailbox with a lot of recent traffic.
        uids = list(reversed(uids))[: args.limit]

        found_any = False
        for uid in uids:
            status, msg_data = connection.fetch(uid, "(RFC822)")
            if status != "OK" or not msg_data or msg_data[0] is None:
                continue
            message = email.message_from_bytes(msg_data[0][1])
            names = attachment_names(message)
            if not names:
                continue
            found_any = True
            print(f"UID {uid.decode()}")
            print(f"  Von: {decode_value(message.get('From'))}")
            print(f"  Datum: {message.get('Date', '')}")
            print(f"  Betreff: {decode_value(message.get('Subject'))}")
            print(f"  Anhänge: {', '.join(names)}")
        if not found_any:
            print(
                f"Keine Nachrichten mit Anhang in den letzten {args.days} Tagen "
                f"in „{args.folder}“."
            )
        return 0
    finally:
        connection.logout()


def cmd_fetch(args):
    out_path = Path(args.out)
    if out_path.exists():
        print(f"Zieldatei existiert bereits, breche ab: {out_path}", file=sys.stderr)
        return 1

    connection = connect(args.folder)
    try:
        status, msg_data = connection.fetch(args.uid.encode(), "(RFC822)")
        if status != "OK" or not msg_data or msg_data[0] is None:
            print(f"Nachricht mit UID {args.uid} nicht gefunden.", file=sys.stderr)
            return 1
        message = email.message_from_bytes(msg_data[0][1])

        for filename, part in document_parts(message):
            if filename != args.attachment:
                continue
            payload = part.get_payload(decode=True)
            out_path.parent.mkdir(parents=True, exist_ok=True)
            out_path.write_bytes(payload)
            print(f"Anhang gespeichert: {out_path}")
            return 0

        print(
            f"Anhang „{args.attachment}“ nicht in Nachricht {args.uid} gefunden.",
            file=sys.stderr,
        )
        return 1
    finally:
        connection.logout()


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    subparsers = parser.add_subparsers(dest="command", required=True)

    folders_parser = subparsers.add_parser(
        "folders", help="List available IMAP folders"
    )
    folders_parser.set_defaults(func=cmd_folders)

    search_parser = subparsers.add_parser(
        "search", help="List recent messages in one folder that have attachments"
    )
    search_parser.add_argument("--folder", default="INBOX")
    search_parser.add_argument("--days", type=int, default=90)
    search_parser.add_argument("--limit", type=int, default=30)
    search_parser.set_defaults(func=cmd_search)

    fetch_parser = subparsers.add_parser(
        "fetch", help="Save one attachment from one message to disk"
    )
    fetch_parser.add_argument("--folder", default="INBOX")
    fetch_parser.add_argument("--uid", required=True)
    fetch_parser.add_argument("--attachment", required=True)
    fetch_parser.add_argument("--out", required=True)
    fetch_parser.set_defaults(func=cmd_fetch)

    args = parser.parse_args()
    sys.exit(args.func(args))


if __name__ == "__main__":
    main()