"""Shared mail routing policy for mailcat.""" import email.utils import json import re from datetime import datetime from functools import cache from pathlib import Path PREFIX = "INBOX." MAILINGLIST_ROUTES_FILE = Path(__file__).parent / "mailinglist_routes.json" 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"), (["n26.com"], "Financieel.Bank.N26"), (["update.bunq.com", "hello.bunq.com"], "Financieel.Bank.bunq"), (["snelstart.nl"], "Financieel.Bank.SnelStart"), (["tellow.nl"], "Financieel.Bank.Tellow"), (["mijnkantoorapp.nl", "dijkman-ac.nl"], "Financieel.Boekhouding.Dijkman"), (["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", "notifications.meetjamie.ai"], "Diensten.AI.Jamie"), (["maandag.com"], "Werk.Opdrachten.Maandag"), (["futurexl.nl"], "Werk.Opdrachten.FutureXL"), (["drenthe.nl"], "Werk.Opdrachten.Provincie Drenthe"), (["rdw.nl"], "Werk.Opdrachten.RDW"), (["plasbossinade.nl"], "Werk.Opdrachten.PlasBossinade"), (["boip.int"], "Werk.Overig.BOIP"), (["groningen.nl", "groningenbereikbaar.nl"], "Werk.Overig.Gemeente Groningen"), (["kvk.nl"], "Administratie.KvK"), (["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"), (["cloud86.io", "cloud86.zendesk.com"], "Diensten.Hosting.Cloud86"), (["godaddy.com"], "Diensten.Hosting.GoDaddy"), (["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"), (["nordaccount.com"], "Diensten.Beveiliging.Nord"), (["info.expressvpn.com"], "Diensten.Beveiliging.ExpressVPN"), (["email.remarkable.com"], "Diensten.Apparaten.reMarkable"), (["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", "mail.anthropic.com"], "Diensten.Software.Claude"), (["canva.com", "account.canva.com"], "Diensten.Software.Canva"), (["microsoft.com"], "Diensten.Software.Microsoft"), (["microsoft365.com"], "Diensten.Software.Microsoft Teams"), (["market.envato.com"], "Diensten.Software.Envato"), (["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"), (["notify.orcid.org"], "Nieuwsbrieven.Leren.ORCID"), (["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"), (["tasks.clickup.com"], "Afmelden.Software.ClickUp"), (["gemeenteprojecten.talent-pool.com"], "Afmelden.Werk.Gemeenteprojecten"), (["engage.istockphoto.com"], "Afmelden.Creatief.iStock"), (["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"} @cache def mailinglist_routes() -> list[dict]: """Return normalized mailing-list routing rules.""" if not MAILINGLIST_ROUTES_FILE.exists(): return [] return json.loads(MAILINGLIST_ROUTES_FILE.read_text(encoding="utf-8")) def mailinglist_destination(from_addr: str) -> str | None: """Return a mailing-list destination based on normalized policy rules.""" domain = sender_domain(from_addr) from_lo = from_addr.lower() for rule in mailinglist_routes(): domains = {item.lower() for item in rule.get("domains", [])} if domain and domain in domains: return rule["mailbox"] for needle in rule.get("from_contains", []): if needle.lower() in from_lo: return rule["mailbox"] return None 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__" if destination := mailinglist_destination(from_addr): return destination 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( rule["mailbox"] for rule in mailinglist_routes() if rule.get("mailbox") ) 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)