Initial status for developed files on dagda

mail_routes.py

@@ -0,0 +1,165 @@
+"""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_FILE = Path(__file__).parent / "domain_routes.json"
+
+
+def _normalize_domain(value: str) -> str:
+    domain = value.strip().lower()
+    if not domain or "@" in domain or "/" in domain or any(ch.isspace() for ch in domain):
+        raise ValueError(f"Invalid domain route domain: {value!r}")
+    return domain
+
+
+def _normalize_mailbox(value: str) -> str:
+    mailbox = value.strip()
+    if not mailbox or mailbox.startswith(PREFIX) or mailbox.startswith(".") or mailbox.endswith(".") or ".." in mailbox:
+        raise ValueError(f"Invalid domain route mailbox: {value!r}")
+    return mailbox
+
+
+def normalize_domain_route(rule: dict) -> tuple[list[str], str]:
+    """Return a validated domain route as (domains, mailbox)."""
+    if not isinstance(rule, dict):
+        raise ValueError(f"Invalid domain route entry: {rule!r}")
+
+    raw_domains = rule.get("domains")
+    if not isinstance(raw_domains, list):
+        raise ValueError(f"Domain route domains must be a list: {rule!r}")
+
+    domains = [_normalize_domain(str(item)) for item in raw_domains]
+    domains = list(dict.fromkeys(domains))
+    if not domains:
+        raise ValueError(f"Domain route must contain at least one domain: {rule!r}")
+
+    mailbox = _normalize_mailbox(str(rule.get("mailbox", "")))
+    return domains, mailbox
+
+
+@cache
+def domain_routes() -> list[tuple[list[str], str]]:
+    """Return validated domain routing rules from JSON policy."""
+    raw = json.loads(DOMAIN_ROUTES_FILE.read_text(encoding="utf-8"))
+    if not isinstance(raw, list):
+        raise ValueError("domain_routes.json must contain a list")
+    return [normalize_domain_route(rule) for rule in raw]
+
+
+@cache
+def domain_lookup() -> dict[str, str]:
+    """Return domain-to-mailbox lookup built from domain_routes.json."""
+    lookup: dict[str, str] = {}
+    for domains, target in domain_routes():
+        for domain in domains:
+            lookup[domain] = target
+    return lookup
+
+
+# Compatibility for existing scripts/tests that imported DOMAIN_ROUTES directly.
+DOMAIN_ROUTES = domain_routes()
+DOMAIN_LOOKUP = domain_lookup()
+
+
+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)