Add decisions-based mailing list routing
The prompts used to arrive at this change since the previous commit User said that in the future they will run scripts themselves when instructed, then asked to take the next step after the decisions.json mailing-list routing report and plan. Any special observations that may be relevant for version management for this version. Be brief. Added mailinglist_routes.json as the normalized runtime policy and wired mail_routes.py to use it before legacy domain routes. No mailbox-affecting scripts were run; validation was local only.
This commit is contained in:
@@ -1,10 +1,14 @@
|
||||
"""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"),
|
||||
@@ -104,6 +108,31 @@ EXCLUDED_SOURCE_FOLDERS = {
|
||||
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 ""
|
||||
@@ -126,6 +155,8 @@ def route_from_subject(from_addr: str, subject: str) -> str | 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))
|
||||
|
||||
|
||||
@@ -137,6 +168,11 @@ def is_source_folder(folder: str) -> bool:
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user