Extract shared mail routing and IMAP folder helpers

Prompt since previous commit: execute step 1

Centralized route targets, source-folder policy, and modified UTF-7 mailbox quoting. Verification: py_compile, route/folder consistency, and git diff --check passed.
This commit is contained in:
2026-07-04 12:23:14 +02:00
parent 537c87e472
commit eb53bc123a
8 changed files with 329 additions and 522 deletions
+4 -14
View File
@@ -26,6 +26,7 @@ from pathlib import Path
from collections import defaultdict
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from imap_utils import list_folders, quote_mailbox
# Mappen die NIET in het overzicht komen (systeem + inbox zelf)
UITSLUITINGEN = {
@@ -86,14 +87,6 @@ def leesbare_map(imap_naam: str) -> str:
return imap_naam.removeprefix("INBOX.").replace(".", " ")
def parse_folder_name(raw: bytes) -> str | None:
decoded = raw.decode("utf-8", errors="replace")
m = re.match(r'\(.*?\)\s+".*?"\s+(.*)', decoded)
if m:
return m.group(1).strip().strip('"')
return None
# ── IMAP ophalen ──────────────────────────────────────────────────────────────
def haal_berichten_op(mail: imaplib.IMAP4_SSL, zoekdatum: date) -> dict[str, list[dict]]:
@@ -104,13 +97,10 @@ def haal_berichten_op(mail: imaplib.IMAP4_SSL, zoekdatum: date) -> dict[str, lis
since = imap_date(zoekdatum)
before = imap_date(zoekdatum + timedelta(days=1))
status, raw_folders = mail.list()
if status != "OK":
alle_mappen = list_folders(mail)
if not alle_mappen:
return {}
alle_mappen = [parse_folder_name(f) for f in raw_folders if f]
alle_mappen = [f for f in alle_mappen if f]
resultaat: dict[str, list[dict]] = defaultdict(list)
for folder in sorted(alle_mappen):
@@ -119,7 +109,7 @@ def haal_berichten_op(mail: imaplib.IMAP4_SSL, zoekdatum: date) -> dict[str, lis
if UITSLUIT_PATRONEN.match(folder):
continue
status, data = mail.select(f'"{folder}"', readonly=True)
status, data = mail.select(quote_mailbox(folder), readonly=True)
if status != "OK":
continue
+4 -19
View File
@@ -13,23 +13,13 @@ import re
import sys
import getpass
from pathlib import Path
from imap_utils import list_folders, quote_mailbox
IMAP_HOST = "australius.nl"
IMAP_PORT = 993
LOCAL_DIR = Path(__file__).parent / "mailbox"
def parse_folder_name(line: bytes) -> str | None:
"""Haal mapnaam op uit IMAP LIST-response."""
decoded = line.decode("utf-8", errors="replace")
# Formaat: (\Flags) "delimiter" "naam" of (\Flags) "delimiter" naam
match = re.match(r'\(.*?\)\s+".*?"\s+(.*)', decoded)
if not match:
return None
name = match.group(1).strip().strip('"')
return name if name else None
def safe_path(name: str) -> str:
"""Zet mapnaam om naar een veilige directorynaam."""
return re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", name)
@@ -49,15 +39,11 @@ def download():
print("Ingelogd.\n")
# Alle mappen ophalen
status, raw_folders = mail.list()
if status != "OK":
folders = list_folders(mail)
if not folders:
print("Kan mappenlijst niet ophalen.")
sys.exit(1)
folders = [parse_folder_name(f) for f in raw_folders if f]
folders = [f for f in folders if f]
print(f"Gevonden mappen ({len(folders)}):")
for f in folders:
print(f" {f}")
@@ -69,8 +55,7 @@ def download():
for folder_idx, folder in enumerate(folders, 1):
folder_prefix = f"[map {folder_idx}/{num_folders}] {folder}"
quoted = f'"{folder}"'
status, data = mail.select(quoted, readonly=True)
status, data = mail.select(quote_mailbox(folder), readonly=True)
if status != "OK":
print(f"{folder_prefix}: kan niet openen overgeslagen.")
continue
+96
View File
@@ -0,0 +1,96 @@
"""Shared IMAP mailbox-name helpers."""
import base64
import re
def to_mutf7(s: str) -> str:
"""Encode a mailbox name as IMAP modified UTF-7."""
res = []
buf = []
for c in s:
if 0x20 <= ord(c) <= 0x7E and c != "&":
if buf:
b64 = (
base64.b64encode("".join(buf).encode("utf-16-be"))
.decode("ascii")
.rstrip("=")
.replace("/", ",")
)
res.append(f"&{b64}-")
buf.clear()
res.append(c)
elif c == "&":
if buf:
b64 = (
base64.b64encode("".join(buf).encode("utf-16-be"))
.decode("ascii")
.rstrip("=")
.replace("/", ",")
)
res.append(f"&{b64}-")
buf.clear()
res.append("&-")
else:
buf.append(c)
if buf:
b64 = (
base64.b64encode("".join(buf).encode("utf-16-be"))
.decode("ascii")
.rstrip("=")
.replace("/", ",")
)
res.append(f"&{b64}-")
return "".join(res)
def from_mutf7(s: str) -> str:
"""Decode an IMAP modified UTF-7 mailbox name."""
res = []
i = 0
while i < len(s):
if s[i] == "&":
j = s.index("-", i + 1)
if j == i + 1:
res.append("&")
else:
encoded = s[i + 1 : j].replace(",", "/")
pad = (4 - len(encoded) % 4) % 4
res.append(
base64.b64decode(encoded + "=" * pad).decode("utf-16-be")
)
i = j + 1
else:
res.append(s[i])
i += 1
return "".join(res)
def quote_mailbox(folder: str) -> str:
"""Return an IMAP-safe quoted mailbox name."""
escaped = to_mutf7(folder).replace("\\", "\\\\").replace('"', r"\"")
return f'"{escaped}"'
def parse_folder_name(raw: bytes) -> str | None:
"""Extract and decode the mailbox name from an IMAP LIST response."""
decoded = raw.decode("ascii", errors="replace")
match = re.match(r'\(.*?\)\s+".*?"\s+(.*)', decoded)
if not match:
return None
name = match.group(1).strip().strip('"')
if not name:
return None
try:
return from_mutf7(name)
except Exception:
return name
def list_folders(mail) -> list[str]:
"""Return decoded folder names from an IMAP connection."""
status, raw_folders = mail.list()
if status != "OK" or not raw_folders:
return []
folders = [parse_folder_name(item) for item in raw_folders if item]
return sorted(folder for folder in folders if folder)
+6 -14
View File
@@ -24,6 +24,7 @@ import time
import config
from pathlib import Path
from datetime import datetime, timezone
from imap_utils import list_folders, quote_mailbox
LOG_FILE = Path(__file__).parent / "backup_log.json"
@@ -33,13 +34,6 @@ SKIP_MAPPEN = {"INBOX.Trash", "INBOX.Spam"}
# ── helpers ───────────────────────────────────────────────────────────────────
def parse_folder_name(raw: bytes) -> str | None:
decoded = raw.decode("utf-8", errors="replace")
m = re.match(r'\(.*?\)\s+".*?"\s+(.*)', decoded)
if m:
return m.group(1).strip().strip('"')
return None
def parse_flags(flag_str: str) -> list[str]:
"""Haal standaard IMAP-vlaggen op uit fetch-response."""
@@ -60,7 +54,7 @@ def parse_internaldate(response: bytes) -> str | None:
def haal_bestaande_message_ids(mail: imaplib.IMAP4_SSL, folder: str) -> set[str]:
"""Geef alle Message-IDs terug die al in een doelmap staan."""
status, data = mail.select(f'"{folder}"', readonly=True)
status, data = mail.select(quote_mailbox(folder), readonly=True)
if status != "OK" or int(data[0]) == 0:
return set()
@@ -100,7 +94,7 @@ def kopieer_map(
Geeft (gekopieerd, overgeslagen) terug.
"""
# Selecteer bronmap
status, data = bron.select(f'"{folder}"', readonly=True)
status, data = bron.select(quote_mailbox(folder), readonly=True)
if status != "OK":
print(f" Kan bronmap niet openen: {folder}")
return 0, 0
@@ -116,7 +110,7 @@ def kopieer_map(
ids = data[0].split()
# Zorg dat doelmap bestaat
doel.create(f'"{folder}"') # negeer fout als al bestaat
doel.create(quote_mailbox(folder)) # negeer fout als al bestaat
# Welke Message-IDs staan al in de doelmap?
bestaande = haal_bestaande_message_ids(doel, folder)
@@ -166,7 +160,7 @@ def kopieer_map(
# Schrijf naar doelmap
flag_str = "(" + " ".join(flags) + ")" if flags else "()"
status, _ = doel.append(
f'"{folder}"',
quote_mailbox(folder),
flag_str,
internaldate,
raw_message,
@@ -242,9 +236,7 @@ def run():
sys.exit(1)
# Mappenlijst van bron
status, raw_folders = bron.list()
alle_mappen = [parse_folder_name(f) for f in raw_folders if f]
alle_mappen = [f for f in alle_mappen if f and f not in SKIP_MAPPEN]
alle_mappen = [folder for folder in list_folders(bron) if folder not in SKIP_MAPPEN]
alle_mappen.sort()
log = load_log()
+6 -198
View File
@@ -5,191 +5,15 @@ Bestaande mappen worden overgeslagen.
Gebruik: python3 maak_mappen.py
"""
import imaplib
import base64
import re
import sys
import imaplib
import config
# Sommige Dovecot-servers eisen "INBOX." als prefix voor submappen.
# Zet op "" als de server mappen zonder prefix accepteert.
PREFIX = "INBOX."
# ── mUTF-7 helpers (RFC 3501) ─────────────────────────────────────────────────
def to_mutf7(s: str) -> str:
"""Codeer een mapnaam naar Modified UTF-7 voor IMAP-aanroepen (RFC 3501).
Verschillen met standaard base64: geen padding, '/' vervangen door ','."""
res = []
buf = []
for c in s:
if 0x20 <= ord(c) <= 0x7e and c != "&":
if buf:
b64 = (
base64.b64encode("".join(buf).encode("utf-16-be"))
.decode("ascii")
.rstrip("=")
.replace("/", ",")
)
res.append(f"&{b64}-")
buf.clear()
res.append(c)
elif c == "&":
if buf:
b64 = (
base64.b64encode("".join(buf).encode("utf-16-be"))
.decode("ascii")
.rstrip("=")
.replace("/", ",")
)
res.append(f"&{b64}-")
buf.clear()
res.append("&-")
else:
buf.append(c)
if buf:
b64 = (
base64.b64encode("".join(buf).encode("utf-16-be"))
.decode("ascii")
.rstrip("=")
.replace("/", ",")
)
res.append(f"&{b64}-")
return "".join(res)
def from_mutf7(s: str) -> str:
"""Decodeer een mUTF-7-mapnaam (van server) terug naar Unicode."""
res = []
i = 0
while i < len(s):
if s[i] == "&":
j = s.index("-", i + 1)
if j == i + 1:
res.append("&")
else:
encoded = s[i + 1 : j].replace(",", "/")
pad = (4 - len(encoded) % 4) % 4
res.append(
base64.b64decode(encoded + "=" * pad).decode("utf-16-be")
)
i = j + 1
else:
res.append(s[i])
i += 1
return "".join(res)
from imap_utils import list_folders, quote_mailbox
from mail_routes import PREFIX, destination_folders
def mappen() -> list[str]:
"""Alle te maken mappen, zonder prefix (prefix wordt automatisch toegevoegd)."""
folders = [
# Nieuwsbrieven
"Nieuwsbrieven",
"Nieuwsbrieven.Leren",
"Nieuwsbrieven.Leren.AI Report",
"Nieuwsbrieven.Leren.Ben Tiggelaar",
"Nieuwsbrieven.Leren.Brian Keating",
"Nieuwsbrieven.Leren.Coursera",
"Nieuwsbrieven.Leren.Elizabeth Butler",
"Nieuwsbrieven.Leren.Eva Keiffenheim",
"Nieuwsbrieven.Leren.MIT Open Learning",
"Nieuwsbrieven.Leren.MIT Technology Review",
"Nieuwsbrieven.Leren.Nick Milo",
"Nieuwsbrieven.Leren.Storytelling with Data",
"Nieuwsbrieven.Nieuws & Vakbladen",
"Nieuwsbrieven.Nieuws & Vakbladen.Computable",
"Nieuwsbrieven.Nieuws & Vakbladen.The Presentation Guru",
"Nieuwsbrieven.Nieuws & Vakbladen.Vakblad Crisismanager",
"Nieuwsbrieven.Overig",
"Nieuwsbrieven.Overig.Lendahand",
"Nieuwsbrieven.Overig.The Good Roll",
# Financieel
"Financieel",
"Financieel.Bank",
"Financieel.Bank.ABN AMRO",
"Financieel.Bank.ICS",
"Financieel.Bank.NPEX",
"Financieel.Bank.SnelStart",
"Financieel.Bank.Tellow",
"Financieel.Verzekering",
"Financieel.Verzekering.Univé",
"Financieel.Verzekering.VEZA",
"Financieel.Facturen",
# Facturen per kwartaal: 2020 t/m huidig jaar + 1
*[
f"Financieel.Facturen.{jaar}.{kwartaal}"
for jaar in range(2020, 2027)
for kwartaal in ["Q1", "Q2", "Q3", "Q4"]
],
# Werk
"Werk",
"Werk.Opdrachten",
"Werk.Opdrachten.Circle8",
"Werk.Opdrachten.Flextender",
"Werk.Opdrachten.FutureXL",
"Werk.Opdrachten.Get There",
"Werk.Opdrachten.Huckr",
"Werk.Opdrachten.IMMENS",
"Werk.Opdrachten.Jamie",
"Werk.Opdrachten.OneStopSourcing",
"Werk.Overig",
"Werk.Overig.BOIP",
"Werk.Overig.Gemeente Groningen",
# Diensten
"Diensten",
"Diensten.Hosting",
"Diensten.Hosting.Backblaze",
"Diensten.Hosting.Lets Encrypt",
"Diensten.Hosting.STRATO",
"Diensten.Hosting.TransIP",
"Diensten.Software",
"Diensten.Software.1Password",
"Diensten.Software.CogSci Apps",
"Diensten.Software.Claude",
"Diensten.Software.Cursor",
"Diensten.Software.Kahoot",
"Diensten.Software.Matter",
"Diensten.Software.Microsoft Teams",
"Diensten.Software.MindNode",
"Diensten.Software.Papers",
"Diensten.Software.Perplexity",
"Diensten.Software.Sophos",
"Diensten.Telecom",
"Diensten.Telecom.KPN",
"Diensten.Telecom.Vodafone",
# Mobiliteit
"Mobiliteit",
"Mobiliteit.EV",
"Mobiliteit.EV.Alfen",
"Mobiliteit.EV.Eneco eMobility",
"Mobiliteit.EV.FastNed",
"Mobiliteit.EV.Mitsubishi",
"Mobiliteit.EV.OPnGO",
"Mobiliteit.EV.Shell Recharge",
"Mobiliteit.EV.Tesla",
"Mobiliteit.OV & Reizen",
"Mobiliteit.OV & Reizen.Booking",
"Mobiliteit.OV & Reizen.Flitsmeister",
"Mobiliteit.OV & Reizen.NS",
# Bestellingen
"Bestellingen",
"Bestellingen.123inkt",
"Bestellingen.Allekabels",
"Bestellingen.bol",
"Bestellingen.Coolblue",
"Bestellingen.DHL",
"Bestellingen.GLS",
"Bestellingen.Makro",
"Bestellingen.Office Centre",
"Bestellingen.PostNL",
"Bestellingen.Sligro",
"Bestellingen.Smartphonehoesjes",
"Bestellingen.Viking",
# Technisch
"Technisch",
"Technisch.DMARC",
]
return folders
return destination_folders()
def run():
@@ -204,21 +28,7 @@ def run():
print(f"Verbinding mislukt: {e}")
sys.exit(1)
# Bestaande mappen ophalen en decoderen vanuit mUTF-7
status, raw = mail.list()
existing = set()
for item in raw:
if not item:
continue
# Serverresponse is ASCII (mUTF-7); decodeer naar leesbare naam
raw_str = item.decode("ascii", errors="replace")
m = re.search(r'"?" (.*?)$', raw_str)
if m:
raw_name = m.group(1).strip('"')
try:
existing.add(from_mutf7(raw_name))
except Exception:
existing.add(raw_name)
existing = set(list_folders(mail))
te_maken = mappen()
print(f"\n{len(te_maken)} mappen te controleren ({len(existing)} bestaan al op server)\n")
@@ -233,9 +43,7 @@ def run():
print(f" ✓ bestaat {folder}")
overgeslagen += 1
continue
# Codeer naar mUTF-7 voor IMAP (puur ASCII, ook voor &, é, enz.)
gecodeerd = to_mutf7(volledige_naam)
status, data = mail.create(f'"{gecodeerd}"')
status, data = mail.create(quote_mailbox(volledige_naam))
if status == "OK":
print(f" + aangemaakt {folder}")
aangemaakt += 1
+152
View File
@@ -0,0 +1,152 @@
"""Shared mail routing policy for mailcat."""
import email.utils
import re
from datetime import datetime
PREFIX = "INBOX."
DOMAIN_ROUTES: list[tuple[list[str], str]] = [
(["nl.abnamro.com", "abnamro.nl"], "Financieel.Bank.ABN AMRO"),
(["icscards.nl", "icsmarketing.icscards.nl", "service.icscards.nl"], "Financieel.Bank.ICS"),
(["npex.nl"], "Financieel.Bank.NPEX"),
(["snelstart.nl"], "Financieel.Bank.SnelStart"),
(["tellow.nl"], "Financieel.Bank.Tellow"),
(["unive.nl", "nieuwsbrief.unive.nl"], "Financieel.Verzekering.Univé"),
(["heinenoord.nl"], "Financieel.Verzekering.VEZA"),
(["flextender.nl"], "Werk.Opdrachten.Flextender"),
(["onestopsourcing.nl"], "Werk.Opdrachten.OneStopSourcing"),
(["immens.nu"], "Werk.Opdrachten.IMMENS"),
(["circle8.nl"], "Werk.Opdrachten.Circle8"),
(["getthere.nl"], "Werk.Opdrachten.Get There"),
(["huckr.ai"], "Werk.Opdrachten.Huckr"),
(["hey.meetjamie.ai"], "Werk.Opdrachten.Jamie"),
(["futurexl.nl"], "Werk.Opdrachten.FutureXL"),
(["boip.int"], "Werk.Overig.BOIP"),
(["groningen.nl", "groningenbereikbaar.nl"], "Werk.Overig.Gemeente Groningen"),
(["eneco-emobility.com", "eneco.nl"], "Mobiliteit.EV.Eneco eMobility"),
(["shellrecharge.com"], "Mobiliteit.EV.Shell Recharge"),
(["fastned.nl"], "Mobiliteit.EV.FastNed"),
(["opngo.com"], "Mobiliteit.EV.OPnGO"),
(["tesla.com"], "Mobiliteit.EV.Tesla"),
(["mmsn.nl"], "Mobiliteit.EV.Mitsubishi"),
(["alfen.com"], "Mobiliteit.EV.Alfen"),
(["ns.nl", "email.ns.nl"], "Mobiliteit.OV & Reizen.NS"),
(["booking.com"], "Mobiliteit.OV & Reizen.Booking"),
(["flitsmeister.nl"], "Mobiliteit.OV & Reizen.Flitsmeister"),
(["postnl.nl", "edm.postnl.nl", "notificatie.postnl.nl"], "Bestellingen.PostNL"),
(["dhlparcel.nl", "dhlecommerce.nl"], "Bestellingen.DHL"),
(["gls-netherlands.com"], "Bestellingen.GLS"),
(["bol.com", "feedback.bol.com"], "Bestellingen.bol"),
(["coolblue.nl", "coolblue.eu"], "Bestellingen.Coolblue"),
(["allekabels.nl"], "Bestellingen.Allekabels"),
(["vikingdirect.nl", "delivery.vikingdirect.nl", "online.vikingdirect.nl"], "Bestellingen.Viking"),
(["123inkt.nl", "m.123inkt.nl"], "Bestellingen.123inkt"),
(["officecentre.nl"], "Bestellingen.Office Centre"),
(["sligro.nl"], "Bestellingen.Sligro"),
(["makro.nl"], "Bestellingen.Makro"),
(["smartphonehoesjes.nl"], "Bestellingen.Smartphonehoesjes"),
(["strato.com", "news.strato.com", "marketingradar.strato.com"], "Diensten.Hosting.STRATO"),
(["transip.nl"], "Diensten.Hosting.TransIP"),
(["letsencrypt.org"], "Diensten.Hosting.Lets Encrypt"),
(["backblaze.com"], "Diensten.Hosting.Backblaze"),
(["kpn.com", "campagnes.kpn.com"], "Diensten.Telecom.KPN"),
(["zakelijk.vodafone.nl", "vodafone.nl"], "Diensten.Telecom.Vodafone"),
(["cursor.com"], "Diensten.Software.Cursor"),
(["1password.com"], "Diensten.Software.1Password"),
(["home.sophos.com", "cleverbridge.com"], "Diensten.Software.Sophos"),
(["cogsciapps.com"], "Diensten.Software.CogSci Apps"),
(["claude.com"], "Diensten.Software.Claude"),
(["microsoft365.com"], "Diensten.Software.Microsoft Teams"),
(["getmatter.com"], "Diensten.Software.Matter"),
(["mindnode.com"], "Diensten.Software.MindNode"),
(["papersapp.com"], "Diensten.Software.Papers"),
(["team.kahoot.com"], "Diensten.Software.Kahoot"),
(["perplexity.ai"], "Diensten.Software.Perplexity"),
(["coursera.org", "email.coursera.org", "m.mail.coursera.org", "m.learn.coursera.org"], "Nieuwsbrieven.Leren.Coursera"),
(["technologyreview.com", "fulfillment.technologyreview.com"], "Nieuwsbrieven.Leren.MIT Technology Review"),
(["mit.edu"], "Nieuwsbrieven.Leren.MIT Open Learning"),
(["storytellingwithdata.com"], "Nieuwsbrieven.Leren.Storytelling with Data"),
(["linkingyourthinking.com"], "Nieuwsbrieven.Leren.Nick Milo"),
(["evakeiffenheim.com"], "Nieuwsbrieven.Leren.Eva Keiffenheim"),
(["tiggelaar.nl"], "Nieuwsbrieven.Leren.Ben Tiggelaar"),
(["briankeating.com"], "Nieuwsbrieven.Leren.Brian Keating"),
(["beehiiv.com", "mail.beehiiv.com"], "Nieuwsbrieven.Leren.AI Report"),
(["elizabethbutlermd.com"], "Nieuwsbrieven.Leren.Elizabeth Butler"),
(["crisismanager.nl"], "Nieuwsbrieven.Nieuws & Vakbladen.Vakblad Crisismanager"),
(["jaarbeurs.nl"], "Nieuwsbrieven.Nieuws & Vakbladen.Computable"),
(["presentation-guru.com"], "Nieuwsbrieven.Nieuws & Vakbladen.The Presentation Guru"),
(["lendahand.com"], "Nieuwsbrieven.Overig.Lendahand"),
(["thegoodroll.nl"], "Nieuwsbrieven.Overig.The Good Roll"),
(["dmarc.postmarkapp.com"], "Technisch.DMARC"),
]
DOMAIN_LOOKUP: dict[str, str] = {
domain.lower(): target
for domains, target in DOMAIN_ROUTES
for domain in domains
}
FACTUUR_KEYWORDS = [
"factuur", "invoice", "rekening", "nota", "betaalverzoek",
"receipt", "orderbevestiging", "order confirmation",
"betaling ontvangen", "payment received",
"uw factuur", "your invoice", "je factuur",
]
EXCLUDED_SOURCE_FOLDERS = {
"INBOX.Sent", "Sent",
"INBOX.Drafts", "Drafts",
"INBOX.Trash", "Trash",
"INBOX.Spam", "Spam",
}
ALREADY_SORTED_FOLDERS = {"INBOX.Facturen - verwerkt"}
def sender_domain(from_addr: str) -> str:
match = re.search(r"@([\w.\-]+)", from_addr)
return match.group(1).lower() if match else ""
def quarter_folder(date_str: str) -> str:
"""Return the invoice quarter target based on an email Date header."""
try:
dt = email.utils.parsedate_to_datetime(date_str)
year = dt.year
quarter = (dt.month - 1) // 3 + 1
return f"Financieel.Facturen.{year}.Q{quarter}"
except Exception:
year = datetime.now().year
return f"Financieel.Facturen.{year}.Q1"
def route_from_subject(from_addr: str, subject: str) -> str | None:
"""Return a destination folder, __FACTUUR_DATE__, or None."""
subj_lo = subject.lower()
if any(keyword in subj_lo for keyword in FACTUUR_KEYWORDS):
return "__FACTUUR_DATE__"
return DOMAIN_LOOKUP.get(sender_domain(from_addr))
def is_source_folder(folder: str) -> bool:
"""True if a folder should be scanned as a source mailbox."""
return folder not in EXCLUDED_SOURCE_FOLDERS and folder not in ALREADY_SORTED_FOLDERS
def destination_folders(start_year: int = 2020, end_year: int = 2026) -> list[str]:
"""Return all destination folders without the INBOX. prefix."""
folders = {target for _, target in DOMAIN_ROUTES}
folders.update(
f"Financieel.Facturen.{year}.Q{quarter}"
for year in range(start_year, end_year + 1)
for quarter in range(1, 5)
)
with_parents = set(folders)
for folder in list(folders):
parts = folder.split(".")
for i in range(1, len(parts)):
with_parents.add(".".join(parts[:i]))
return sorted(with_parents)
+51 -32
View File
@@ -22,8 +22,6 @@ Git workflow:
- Work should continue on `codex` unless the user says otherwise.
- Follow the users global workflow: initialize Git when needed, branch before work in existing repos, and commit after each user-prompted change using the required three-section commit message format.
Tracked files include Python scripts, generated reports, logs, caches, and project docs.
Ignored local data includes:
- `mailbox/`
@@ -58,28 +56,58 @@ Sieve:
- Sieve deployment is not viable with the provider.
- Future work should move operational automation to Python over IMAP/SMTP.
## Important Current Implementation Notes
## Current Implementation Notes
`verplaats_bestaand.py` has already been changed so:
Shared modules:
- `imap_utils.py` centralizes IMAP modified UTF-7 folder encoding/decoding, quoted mailbox names, LIST parsing, and folder listing.
- `mail_routes.py` centralizes `PREFIX`, domain routes, invoice keywords, source-folder exclusions, already-sorted folders, invoice quarter routing, and generated destination folders.
Scripts using shared IMAP folder handling:
- `verplaats_bestaand.py`
- `maak_mappen.py`
- `kopieer_naar_backup.py`
- `download_mailbox.py`
- `dagelijks_overzicht.py`
`verplaats_bestaand.py` current behavior:
- `MAP_RENAMES` is intentionally empty.
- Source folders are selected by `list_source_folders()`.
- Excluded source folders are `Sent`, `Drafts`, `Trash`, and `Spam`, with both bare and `INBOX.` names where relevant.
- `ALREADY_SORTED_FOLDERS = {"INBOX.Facturen - verwerkt"}`.
- Source folders are selected by shared `list_folders()` plus `mail_routes.is_source_folder()`.
- Step 2 routes all eligible source folders, not just `INBOX`.
- Move success now requires both `UID COPY` and `UID STORE +FLAGS \Deleted` to return `OK`.
- Move success requires both `UID COPY` and `UID STORE +FLAGS \Deleted` to return `OK`.
- Step 3 no longer treats `INBOX.Facturen - verwerkt` as a source.
- IMAP mailbox names are consistently quoted and encoded through `imap_utils.quote_mailbox()`.
`maak_mappen.py` has already been changed so:
`maak_mappen.py` current behavior:
- Destination folders are generated from `mail_routes.destination_folders()`.
- It no longer creates `Archief.*` destination folders.
- It creates `Technisch` and `Technisch.DMARC`.
- It creates all parent folders needed for route targets, including `Technisch` and `Technisch.DMARC`.
- IMAP mailbox names are consistently quoted and encoded through `imap_utils.quote_mailbox()`.
Last verification performed:
`kopieer_naar_backup.py` current behavior:
- Mirrors folders from one configured IMAP account to another.
- Skips `INBOX.Trash` and `INBOX.Spam`.
- Deduplicates by `Message-ID` within each destination folder.
- Uses shared folder listing and mailbox quoting/encoding for select/create/append operations.
## Last Verification
The latest verification performed:
- `python3 -m py_compile *.py`
- route/folder consistency check reported `missing_targets 0`
- archive target check reported `archief_targets 0`
- `python3 -c 'from mail_routes import DOMAIN_ROUTES,destination_folders; targets={target for _, target in DOMAIN_ROUTES}; folders=set(destination_folders()); print("routes", len(targets)); print("folders", len(folders)); print("missing_targets", sorted(targets-folders)); print("archief_targets", sorted(t for t in targets if t.startswith("Archief") or ".Archief" in t))'`
- `git diff --check`
Expected consistency result:
- `routes 72`
- `folders 127`
- `missing_targets []`
- `archief_targets []`
## Reports and Findings
@@ -96,32 +124,24 @@ Key earlier findings:
Recommended next work:
1. Refactor shared IMAP utilities:
- folder list parsing
- mUTF-7 encoding/decoding
- folder creation
- UID fetch
- safe copy/delete/expunge
- Message-ID dedupe
1. Harden move and mirror auditability:
- `kopieer_naar_backup.py` should not mark a folder complete if append failures occur.
- `verplaats_bestaand.py` should log failures with source folder, UID, destination, and server response.
- Add a dry-run audit mode showing planned moves by source and destination.
2. Refactor shared routing logic:
- extract the route table from `verplaats_bestaand.py`
- support both historical sorting and live sorting from one source of truth
- preserve invoice priority and quarter-folder routing
2. Add shared operation helpers where useful:
- UID fetch wrappers.
- safe copy/delete/expunge helper.
- Message-ID dedupe helper reused by live automation.
3. Harden scripts:
- `kopieer_naar_backup.py` should not mark a folder complete if append failures occur
- `verplaats_bestaand.py` should log failures with source folder, UID, destination, and server response
- add a dry-run audit mode showing planned moves by source and destination
4. Add VPS automation:
3. Add VPS automation:
- create an IMAP IDLE daemon for `backup@australius.nl`
- deploy via SSH to `vps.australius.nl`
- use a non-root sudo user
- install as a systemd service
- keep secrets in an ignored `config.json` with restrictive permissions
5. Test on `backup@australius.nl` before production:
4. Test on `backup@australius.nl` before production:
- mirror production mail into backup
- create target folders in backup
- run dry-run sorter on backup
@@ -131,4 +151,3 @@ Recommended next work:
## Persistent File Rule
Before every future commit in this project, rewrite this `restart_prompt.md` file so it describes the current state at that commit. Do not append. Replace the content with a fresh, accurate reconstruction prompt.
+10 -245
View File
@@ -18,90 +18,15 @@ Echt uitvoeren: python3 verplaats_bestaand.py --uitvoeren
Log: verplaats_log.json
"""
import imaplib
import base64
import email
import email.header
import email.utils
import re
import imaplib
import sys
import json
import config
from pathlib import Path
from datetime import datetime, timezone
PREFIX = "INBOX."
def to_mutf7(s: str) -> str:
"""Codeer mapnaam naar Modified UTF-7 voor IMAP-aanroepen (RFC 3501)."""
res = []
buf = []
for c in s:
if 0x20 <= ord(c) <= 0x7e and c != "&":
if buf:
b64 = (
base64.b64encode("".join(buf).encode("utf-16-be"))
.decode("ascii").rstrip("=").replace("/", ",")
)
res.append(f"&{b64}-")
buf.clear()
res.append(c)
elif c == "&":
if buf:
b64 = (
base64.b64encode("".join(buf).encode("utf-16-be"))
.decode("ascii").rstrip("=").replace("/", ",")
)
res.append(f"&{b64}-")
buf.clear()
res.append("&-")
else:
buf.append(c)
if buf:
b64 = (
base64.b64encode("".join(buf).encode("utf-16-be"))
.decode("ascii").rstrip("=").replace("/", ",")
)
res.append(f"&{b64}-")
return "".join(res)
def from_mutf7(s: str) -> str:
"""Decodeer een mUTF-7-mapnaam (van server) terug naar Unicode."""
res = []
i = 0
while i < len(s):
if s[i] == "&":
j = s.index("-", i + 1)
if j == i + 1:
res.append("&")
else:
encoded = s[i + 1 : j].replace(",", "/")
pad = (4 - len(encoded) % 4) % 4
res.append(
base64.b64decode(encoded + "=" * pad).decode("utf-16-be")
)
i = j + 1
else:
res.append(s[i])
i += 1
return "".join(res)
def parse_folder_name(raw: bytes) -> str | None:
"""Haal mapnaam op uit IMAP LIST-response en decodeer mUTF-7."""
decoded = raw.decode("ascii", errors="replace")
match = re.match(r'\(.*?\)\s+".*?"\s+(.*)', decoded)
if not match:
return None
name = match.group(1).strip().strip('"')
if not name:
return None
try:
return from_mutf7(name)
except Exception:
return name
from imap_utils import list_folders, quote_mailbox
from mail_routes import PREFIX, is_source_folder, quarter_folder, route_from_subject
LOG_FILE = Path(__file__).parent / "verplaats_log.json"
@@ -109,116 +34,6 @@ LOG_FILE = Path(__file__).parent / "verplaats_log.json"
DRY_RUN = "--uitvoeren" not in sys.argv
# ── routeringstabel ──────────────────────────────────────────────────────────
# (domain, doelmap) — zelfde logica als mailrules.sieve
# Facturen worden apart behandeld (kwartaalsortering).
DOMAIN_ROUTES: list[tuple[list[str], str]] = [
# Bank
(["nl.abnamro.com", "abnamro.nl"], "Financieel.Bank.ABN AMRO"),
(["icscards.nl", "icsmarketing.icscards.nl",
"service.icscards.nl"], "Financieel.Bank.ICS"),
(["npex.nl"], "Financieel.Bank.NPEX"),
(["snelstart.nl"], "Financieel.Bank.SnelStart"),
(["tellow.nl"], "Financieel.Bank.Tellow"),
# Verzekering
(["unive.nl", "nieuwsbrief.unive.nl"], "Financieel.Verzekering.Univé"),
(["heinenoord.nl"], "Financieel.Verzekering.VEZA"),
# Werk
(["flextender.nl"], "Werk.Opdrachten.Flextender"),
(["onestopsourcing.nl"], "Werk.Opdrachten.OneStopSourcing"),
(["immens.nu"], "Werk.Opdrachten.IMMENS"),
(["circle8.nl"], "Werk.Opdrachten.Circle8"),
(["getthere.nl"], "Werk.Opdrachten.Get There"),
(["huckr.ai"], "Werk.Opdrachten.Huckr"),
(["hey.meetjamie.ai"], "Werk.Opdrachten.Jamie"),
(["futurexl.nl"], "Werk.Opdrachten.FutureXL"),
(["boip.int"], "Werk.Overig.BOIP"),
(["groningen.nl", "groningenbereikbaar.nl"], "Werk.Overig.Gemeente Groningen"),
# Mobiliteit EV
(["eneco-emobility.com", "eneco.nl"], "Mobiliteit.EV.Eneco eMobility"),
(["shellrecharge.com"], "Mobiliteit.EV.Shell Recharge"),
(["fastned.nl"], "Mobiliteit.EV.FastNed"),
(["opngo.com"], "Mobiliteit.EV.OPnGO"),
(["tesla.com"], "Mobiliteit.EV.Tesla"),
(["mmsn.nl"], "Mobiliteit.EV.Mitsubishi"),
(["alfen.com"], "Mobiliteit.EV.Alfen"),
# Mobiliteit OV
(["ns.nl", "email.ns.nl"], "Mobiliteit.OV & Reizen.NS"),
(["booking.com"], "Mobiliteit.OV & Reizen.Booking"),
(["flitsmeister.nl"], "Mobiliteit.OV & Reizen.Flitsmeister"),
# Bestellingen
(["postnl.nl", "edm.postnl.nl", "notificatie.postnl.nl"], "Bestellingen.PostNL"),
(["dhlparcel.nl", "dhlecommerce.nl"], "Bestellingen.DHL"),
(["gls-netherlands.com"], "Bestellingen.GLS"),
(["bol.com", "feedback.bol.com"], "Bestellingen.bol"),
(["coolblue.nl", "coolblue.eu"], "Bestellingen.Coolblue"),
(["allekabels.nl"], "Bestellingen.Allekabels"),
(["vikingdirect.nl", "delivery.vikingdirect.nl",
"online.vikingdirect.nl"], "Bestellingen.Viking"),
(["123inkt.nl", "m.123inkt.nl"], "Bestellingen.123inkt"),
(["officecentre.nl"], "Bestellingen.Office Centre"),
(["sligro.nl"], "Bestellingen.Sligro"),
(["makro.nl"], "Bestellingen.Makro"),
(["smartphonehoesjes.nl"], "Bestellingen.Smartphonehoesjes"),
# Diensten Hosting
(["strato.com", "news.strato.com",
"marketingradar.strato.com"], "Diensten.Hosting.STRATO"),
(["transip.nl"], "Diensten.Hosting.TransIP"),
(["letsencrypt.org"], "Diensten.Hosting.Lets Encrypt"),
(["backblaze.com"], "Diensten.Hosting.Backblaze"),
# Diensten Telecom
(["kpn.com", "campagnes.kpn.com"], "Diensten.Telecom.KPN"),
(["zakelijk.vodafone.nl", "vodafone.nl"], "Diensten.Telecom.Vodafone"),
# Diensten Software
(["cursor.com"], "Diensten.Software.Cursor"),
(["1password.com"], "Diensten.Software.1Password"),
(["home.sophos.com", "cleverbridge.com"], "Diensten.Software.Sophos"),
(["cogsciapps.com"], "Diensten.Software.CogSci Apps"),
(["claude.com"], "Diensten.Software.Claude"),
(["microsoft365.com"], "Diensten.Software.Microsoft Teams"),
(["getmatter.com"], "Diensten.Software.Matter"),
(["mindnode.com"], "Diensten.Software.MindNode"),
(["papersapp.com"], "Diensten.Software.Papers"),
(["team.kahoot.com"], "Diensten.Software.Kahoot"),
(["perplexity.ai"], "Diensten.Software.Perplexity"),
# Nieuwsbrieven Leren
(["coursera.org", "email.coursera.org",
"m.mail.coursera.org", "m.learn.coursera.org"], "Nieuwsbrieven.Leren.Coursera"),
(["technologyreview.com",
"fulfillment.technologyreview.com"], "Nieuwsbrieven.Leren.MIT Technology Review"),
(["mit.edu"], "Nieuwsbrieven.Leren.MIT Open Learning"),
(["storytellingwithdata.com"], "Nieuwsbrieven.Leren.Storytelling with Data"),
(["linkingyourthinking.com"], "Nieuwsbrieven.Leren.Nick Milo"),
(["evakeiffenheim.com"], "Nieuwsbrieven.Leren.Eva Keiffenheim"),
(["tiggelaar.nl"], "Nieuwsbrieven.Leren.Ben Tiggelaar"),
(["briankeating.com"], "Nieuwsbrieven.Leren.Brian Keating"),
(["beehiiv.com", "mail.beehiiv.com"], "Nieuwsbrieven.Leren.AI Report"),
(["elizabethbutlermd.com"], "Nieuwsbrieven.Leren.Elizabeth Butler"),
# Nieuwsbrieven Nieuws & Vakbladen
(["crisismanager.nl"], "Nieuwsbrieven.Nieuws & Vakbladen.Vakblad Crisismanager"),
(["jaarbeurs.nl"], "Nieuwsbrieven.Nieuws & Vakbladen.Computable"),
(["presentation-guru.com"], "Nieuwsbrieven.Nieuws & Vakbladen.The Presentation Guru"),
# Nieuwsbrieven Overig
(["lendahand.com"], "Nieuwsbrieven.Overig.Lendahand"),
(["thegoodroll.nl"], "Nieuwsbrieven.Overig.The Good Roll"),
# Technisch
(["dmarc.postmarkapp.com"], "Technisch.DMARC"),
]
# Bouw snelle lookup: domain → doelmap
DOMAIN_LOOKUP: dict[str, str] = {}
for domains, target in DOMAIN_ROUTES:
for d in domains:
DOMAIN_LOOKUP[d.lower()] = target
FACTUUR_KEYWORDS = [
"factuur", "invoice", "rekening", "nota", "betaalverzoek",
"receipt", "orderbevestiging", "order confirmation",
"betaling ontvangen", "payment received",
"uw factuur", "your invoice", "je factuur",
]
# Mappen die in één keer worden hernoemd/verplaatst (oud → nieuw).
#
# Bewust leeg: oude mappen, archiefmappen en mailinglistmappen zijn bronnen.
@@ -227,17 +42,6 @@ FACTUUR_KEYWORDS = [
MAP_RENAMES: list[tuple[str, str]] = [
]
EXCLUDED_SOURCE_FOLDERS = {
"INBOX.Sent", "Sent",
"INBOX.Drafts", "Drafts",
"INBOX.Trash", "Trash",
"INBOX.Spam", "Spam",
}
ALREADY_SORTED_FOLDERS = {
"INBOX.Facturen - verwerkt",
}
# ── helpers ──────────────────────────────────────────────────────────────────
@@ -257,47 +61,8 @@ def decode_hdr(value) -> str:
return str(value or "")
def sender_domain(from_addr: str) -> str:
m = re.search(r"@([\w.\-]+)", from_addr)
return m.group(1).lower() if m else ""
def quarter_folder(date_str: str) -> str:
"""Geef de doelmap terug op basis van de e-maildatum."""
try:
dt = email.utils.parsedate_to_datetime(date_str)
jaar = dt.year
maand = dt.month
kwartaal = (maand - 1) // 3 + 1
return f"Financieel.Facturen.{jaar}.Q{kwartaal}"
except Exception:
jaar = datetime.now().year
return f"Financieel.Facturen.{jaar}.Q1"
def route_from_subject(from_addr: str, subject: str) -> str | None:
"""Bepaal doelmap op basis van afzender en onderwerp. None = geen match."""
domain = sender_domain(from_addr)
# Facturen hebben prioriteit
subj_lo = subject.lower()
if any(kw in subj_lo for kw in FACTUUR_KEYWORDS):
return "__FACTUUR_DATE__" # behandeld apart (datum nodig)
return DOMAIN_LOOKUP.get(domain)
def is_source_folder(folder: str) -> bool:
"""True als deze map als bron voor routering gebruikt mag worden."""
return folder not in EXCLUDED_SOURCE_FOLDERS and folder not in ALREADY_SORTED_FOLDERS
def list_source_folders(mail) -> list[str]:
status, raw_folders = mail.list()
if status != "OK" or not raw_folders:
return []
folders = [parse_folder_name(item) for item in raw_folders if item]
return sorted(f for f in folders if f and is_source_folder(f))
return [folder for folder in list_folders(mail) if is_source_folder(folder)]
def move_message(mail, uid: bytes, src: str, dst_full: str, dry: bool) -> bool:
@@ -305,8 +70,8 @@ def move_message(mail, uid: bytes, src: str, dst_full: str, dry: bool) -> bool:
if dry:
return True
try:
mail.select(f'"{to_mutf7(src)}"')
res, _ = mail.uid("COPY", uid, f'"{to_mutf7(dst_full)}"')
mail.select(quote_mailbox(src))
res, _ = mail.uid("COPY", uid, quote_mailbox(dst_full))
if res != "OK":
return False
store_res, _ = mail.uid("STORE", uid, "+FLAGS", "\\Deleted")
@@ -329,7 +94,7 @@ def stap1_mapverplaatsingen(mail, log: dict, dry: bool):
continue
# Controleer of bronmap bestaat
status, data = mail.select(f'"{to_mutf7(oud)}"', readonly=True)
status, data = mail.select(quote_mailbox(oud), readonly=True)
if status != "OK":
print(f" bestaat niet: {oud}")
continue
@@ -346,7 +111,7 @@ def stap1_mapverplaatsingen(mail, log: dict, dry: bool):
ids = data[0].split()
gekopieerd = 0
for uid in ids:
res, _ = mail.copy(uid.decode(), f'"{to_mutf7(nieuw_full)}"')
res, _ = mail.copy(uid.decode(), quote_mailbox(nieuw_full))
if res == "OK":
mail.store(uid.decode(), "+FLAGS", "\\Deleted")
gekopieerd += 1
@@ -370,7 +135,7 @@ def stap2_bronmappen_routing(mail, log: dict, dry: bool):
al_verwerkt = set(log.get("stap2_verwerkt", []))
for bron in bron_mappen:
status, data = mail.select(f'"{to_mutf7(bron)}"', readonly=dry)
status, data = mail.select(quote_mailbox(bron), readonly=dry)
if status != "OK":
print(f" Kan map niet openen: {bron}")
continue
@@ -448,7 +213,7 @@ def stap3_facturen(mail, log: dict, dry: bool):
]
for bron in bron_mappen:
status, data = mail.select(f'"{to_mutf7(bron)}"', readonly=dry)
status, data = mail.select(quote_mailbox(bron), readonly=dry)
if status != "OK":
continue