#!/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()