mailcat/mail_routes.py

"""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)