wienen
5c0e345fd8
Prompts used since the previous commit: - Hieronder staat je oorspronkelijke plan. Volgens mij moet stap 'Add VPS automation' nog worden uitgevoerd, inclusief de testen. Special observations: - Added a testable IMAP sorting daemon, shared IMAP operation helpers, systemd service template, and SSH deployment script. - Local tests passed; no live VPS deployment or mailbox-affecting daemon test was run. - verplaats_log.json remains modified from the user's run and was intentionally not staged.
89 lines
2.8 KiB
Python
89 lines
2.8 KiB
Python
"""Shared IMAP operations used by mailcat sorters."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import email
|
|
import email.header
|
|
from email.message import Message
|
|
|
|
from imap_utils import quote_mailbox
|
|
from mail_routes import PREFIX, quarter_folder, route_from_subject
|
|
|
|
|
|
HEADER_FETCH = "(BODY.PEEK[HEADER.FIELDS (FROM SUBJECT DATE)])"
|
|
|
|
|
|
def decode_header_value(value) -> str:
|
|
"""Decode an RFC 2047 header value into display text."""
|
|
if not value:
|
|
return ""
|
|
try:
|
|
parts = email.header.decode_header(value)
|
|
out = []
|
|
for raw, charset in parts:
|
|
if isinstance(raw, bytes):
|
|
out.append(raw.decode(charset or "utf-8", errors="replace"))
|
|
else:
|
|
out.append(str(raw))
|
|
return " ".join(out).strip()
|
|
except Exception:
|
|
return str(value or "")
|
|
|
|
|
|
def fetch_body(data) -> bytes | None:
|
|
"""Return the first bytes payload from an IMAP FETCH response."""
|
|
if not data:
|
|
return None
|
|
for item in data:
|
|
if isinstance(item, tuple) and len(item) >= 2 and isinstance(item[1], bytes):
|
|
return item[1]
|
|
return None
|
|
|
|
|
|
def parse_header(raw_header: bytes) -> Message:
|
|
"""Parse an IMAP header payload into an email Message."""
|
|
return email.message_from_bytes(raw_header)
|
|
|
|
|
|
def destination_from_header(raw_header: bytes) -> str | None:
|
|
"""Return the full IMAP destination folder for a fetched header."""
|
|
msg = parse_header(raw_header)
|
|
from_raw = decode_header_value(msg.get("From", ""))
|
|
subject = decode_header_value(msg.get("Subject", ""))
|
|
date_str = msg.get("Date", "")
|
|
|
|
destination = route_from_subject(from_raw, subject)
|
|
if destination == "__FACTUUR_DATE__":
|
|
destination = quarter_folder(date_str)
|
|
if destination:
|
|
return PREFIX + destination
|
|
return None
|
|
|
|
|
|
def ensure_mailbox(mail, folder: str) -> bool:
|
|
"""Create a mailbox and its parents if they do not already exist."""
|
|
parts = folder.split(".")
|
|
for index in range(1, len(parts) + 1):
|
|
parent = ".".join(parts[:index])
|
|
status, _ = mail.create(quote_mailbox(parent))
|
|
if status not in {"OK", "NO"}:
|
|
return False
|
|
return True
|
|
|
|
|
|
def move_message(mail, uid: bytes, src: str, dst: str):
|
|
"""Copy a message to dst and mark it deleted in src."""
|
|
select_status, select_data = mail.select(quote_mailbox(src))
|
|
if select_status != "OK":
|
|
return False, ("select_source", select_status, select_data)
|
|
|
|
copy_status, copy_data = mail.uid("COPY", uid, quote_mailbox(dst))
|
|
if copy_status != "OK":
|
|
return False, ("copy", copy_status, copy_data)
|
|
|
|
store_status, store_data = mail.uid("STORE", uid, "+FLAGS", "\\Deleted")
|
|
if store_status != "OK":
|
|
return False, ("store_deleted", store_status, store_data)
|
|
|
|
return True, None
|