Files
mailcat/verplaats_bestaand.py
T
wienen 9c7ba6fa5d Treat only processed invoices as already sorted
Prompts used since the previous commit:

- User said they do not want final destination folders in Archief.
- User clarified: only treat INBOX.Facturen - verwerkt as already sorted; treat all other mailboxes as sources except Sent, Drafts, Trash and Spam.

Special observations:

- Updated source-folder selection accordingly.
- Removed Archief final destination creation and added Technisch destination folders.
- Verified Python compilation and route/folder consistency.
2026-07-04 12:07:27 +02:00

542 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""
Herindeling van bestaande mail naar de nieuwe mappenstructuur.
Stap 1 Vaste mapverplaatsingen:
Uitgeschakeld. Alle niet-uitgesloten mappen worden als bron behandeld.
Stap 2 Per-bericht routing:
Op basis van afzenderdomein en onderwerpkeywords.
Bronmappen: alles behalve Sent, Drafts, Trash, Spam en
INBOX.Facturen - verwerkt.
Stap 3 Facturen per kwartaal sorteren (Facturen - te verwerken).
Standaard: DRY RUN (laat zien wat er zou gebeuren, verplaatst niets).
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 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
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.
# Alleen Sent, Drafts, Trash, Spam en INBOX.Facturen - verwerkt worden
# uitgesloten van routering.
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 ──────────────────────────────────────────────────────────────────
def decode_hdr(value) -> str:
if not value:
return ""
try:
parts = email.header.decode_header(value)
out = []
for raw, cs in parts:
if isinstance(raw, bytes):
out.append(raw.decode(cs or "utf-8", errors="replace"))
else:
out.append(str(raw))
return " ".join(out).strip()
except Exception:
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))
def move_message(mail, uid: bytes, src: str, dst_full: str, dry: bool) -> bool:
"""Kopieer bericht naar dst en markeer als verwijderd in src."""
if dry:
return True
try:
mail.select(f'"{to_mutf7(src)}"')
res, _ = mail.uid("COPY", uid, f'"{to_mutf7(dst_full)}"')
if res != "OK":
return False
store_res, _ = mail.uid("STORE", uid, "+FLAGS", "\\Deleted")
return store_res == "OK"
except Exception as e:
print(f" FOUT bij verplaatsen: {e}")
return False
# ── stap 1: vaste mapverplaatsingen ─────────────────────────────────────────
def stap1_mapverplaatsingen(mail, log: dict, dry: bool):
print("\n── Stap 1: vaste mapverplaatsingen ──")
for oud, nieuw in MAP_RENAMES:
if nieuw == "__FACTUREN__":
continue # wordt in stap 3 verwerkt
if oud in log.get("stap1_klaar", []):
print(f" ✓ al gedaan: {oud}")
continue
# Controleer of bronmap bestaat
status, data = mail.select(f'"{to_mutf7(oud)}"', readonly=True)
if status != "OK":
print(f" bestaat niet: {oud}")
continue
n = int(data[0])
nieuw_full = nieuw if nieuw.startswith("INBOX.") else PREFIX + nieuw
if dry:
print(f" [DRY] {oud}{nieuw} ({n} berichten)")
continue
# Berichten kopiëren
status, data = mail.search(None, "ALL")
ids = data[0].split()
gekopieerd = 0
for uid in ids:
res, _ = mail.copy(uid.decode(), f'"{to_mutf7(nieuw_full)}"')
if res == "OK":
mail.store(uid.decode(), "+FLAGS", "\\Deleted")
gekopieerd += 1
mail.expunge()
print(f" ✓ {oud}{nieuw} ({gekopieerd}/{n})")
log.setdefault("stap1_klaar", []).append(oud)
_save_log(log)
# ── stap 2: per-bericht routing (INBOX) ─────────────────────────────────────
def stap2_bronmappen_routing(mail, log: dict, dry: bool):
print("\n── Stap 2: bronmappen per-bericht routing ──")
bron_mappen = list_source_folders(mail)
print(f" {len(bron_mappen)} bronmappen")
totaal_verplaatst = 0
totaal_geen_match = 0
al_verwerkt = set(log.get("stap2_verwerkt", []))
for bron in bron_mappen:
status, data = mail.select(f'"{to_mutf7(bron)}"', readonly=dry)
if status != "OK":
print(f" Kan map niet openen: {bron}")
continue
n = int(data[0])
if n == 0:
continue
print(f" {bron}: {n} berichten")
status, data = mail.uid("SEARCH", "ALL")
if status != "OK" or not data or not data[0]:
continue
ids = [uid for uid in data[0].split() if uid]
verplaatst = 0
geen_match = 0
for uid in ids:
uid_str = uid.decode()
log_key = f"{bron}:{uid_str}"
if log_key in al_verwerkt:
continue
status, data = mail.uid("FETCH", uid, "(BODY.PEEK[HEADER.FIELDS (FROM SUBJECT DATE)])")
if status != "OK" or not data or not data[0]:
continue
msg = email.message_from_bytes(data[0][1])
from_raw = decode_hdr(msg.get("From", ""))
subject = decode_hdr(msg.get("Subject", ""))
date_str = msg.get("Date", "")
doel = route_from_subject(from_raw, subject)
if doel == "__FACTUUR_DATE__":
doel = PREFIX + quarter_folder(date_str)
elif doel:
doel = PREFIX + doel
else:
geen_match += 1
continue
if bron == doel:
continue
if dry:
print(f" [DRY] {bron}{doel} | {subject[:60]}")
else:
ok = move_message(mail, uid, bron, doel, dry=False)
if ok:
verplaatst += 1
log.setdefault("stap2_verwerkt", []).append(log_key)
if verplaatst % 50 == 0:
mail.expunge()
_save_log(log)
if not dry:
mail.expunge()
_save_log(log)
totaal_verplaatst += verplaatst
totaal_geen_match += geen_match
print(f" Verplaatst: {verplaatst} | Geen match: {geen_match}")
print(f" Totaal verplaatst: {totaal_verplaatst} | Totaal geen match: {totaal_geen_match}")
# ── stap 3: facturen per kwartaal ────────────────────────────────────────────
def stap3_facturen(mail, log: dict, dry: bool):
print("\n── Stap 3: facturen per kwartaal ──")
bron_mappen = [
"INBOX.Facturen - te verwerken",
"INBOX.Financieel.Facturen", # als ze er al in zitten
]
for bron in bron_mappen:
status, data = mail.select(f'"{to_mutf7(bron)}"', readonly=dry)
if status != "OK":
continue
n = int(data[0])
if n == 0:
print(f" {bron}: leeg")
continue
print(f" {bron}: {n} berichten")
status, data = mail.uid("SEARCH", "ALL")
if status != "OK" or not data or not data[0]:
continue
ids = [uid for uid in data[0].split() if uid]
verplaatst = 0
for uid in ids:
status, data = mail.uid("FETCH", uid, "(BODY.PEEK[HEADER.FIELDS (DATE SUBJECT)])")
if status != "OK" or not data or not data[0]:
continue
msg = email.message_from_bytes(data[0][1])
date_str = msg.get("Date", "")
doel = PREFIX + quarter_folder(date_str)
if dry:
dt_kort = date_str[:16] if date_str else "?"
print(f" [DRY] {dt_kort}{doel}")
else:
ok = move_message(mail, uid, bron, doel, dry=False)
if ok:
verplaatst += 1
if not dry:
mail.expunge()
print(f" {verplaatst}/{n} verplaatst")
# ── log ──────────────────────────────────────────────────────────────────────
def _save_log(log: dict):
LOG_FILE.write_text(json.dumps(log, indent=2, ensure_ascii=False), encoding="utf-8")
def load_log() -> dict:
if LOG_FILE.exists():
try:
return json.loads(LOG_FILE.read_text(encoding="utf-8"))
except Exception:
pass
return {}
# ── main ──────────────────────────────────────────────────────────────────────
def run():
if DRY_RUN:
print("DRY RUN — er wordt niets verplaatst.")
print("Voeg --uitvoeren toe om echt uit te voeren.\n")
else:
print("LET OP: dit verplaatst berichten op de echte server.")
bevestig = input("Typ JA om door te gaan: ").strip()
if bevestig != "JA":
print("Afgebroken.")
sys.exit(0)
cfg = config.load()
print(f"Account: {cfg.username}\n")
print(f"Verbinden met {cfg.imap_host}:{cfg.imap_port} ...")
try:
mail = imaplib.IMAP4_SSL(cfg.imap_host, cfg.imap_port)
mail.login(cfg.username, cfg.password)
except Exception as e:
print(f"Verbinding mislukt: {e}")
sys.exit(1)
log = load_log()
stap1_mapverplaatsingen(mail, log, DRY_RUN)
stap2_bronmappen_routing(mail, log, DRY_RUN)
stap3_facturen(mail, log, DRY_RUN)
mail.logout()
print("\nKlaar.")
if DRY_RUN:
print("Draai met --uitvoeren om de wijzigingen door te voeren.")
if __name__ == "__main__":
run()