mailcat
400 lines
15 KiB
Python
400 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Fase 1: Analyseer lokale mailbox-kopie en genereer overzichten.
|
||
Output: reports/mailinglijsten.md, facturen_mailbox.md, facturen_links.md, prullenbak.md
|
||
"""
|
||
|
||
import email
|
||
import email.header
|
||
import email.utils
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
from collections import defaultdict
|
||
from datetime import datetime, timezone
|
||
|
||
MAILBOX_DIR = Path(__file__).parent / "mailbox"
|
||
REPORTS_DIR = Path(__file__).parent / "reports"
|
||
|
||
INVOICE_SUBJECT_KW = [
|
||
"factuur", "invoice", "rekening", "nota", "kwitantie", "receipt",
|
||
"betaalverzoek", "betalingsverzoek", "betaling ontvangen",
|
||
"payment received", "order confirmation", "orderbevestiging",
|
||
"your order", "je bestelling", "proforma",
|
||
]
|
||
INVOICE_ATTACH_KW = [
|
||
"factuur", "invoice", "rekening", "nota", "receipt", "bill",
|
||
]
|
||
INVOICE_LINK_KW = [
|
||
"bekijk je factuur", "download je factuur", "download your invoice",
|
||
"view your invoice", "your invoice is ready", "je factuur staat klaar",
|
||
"factuur is beschikbaar", "invoice is available", "factuur online",
|
||
"invoice online", "mijn facturen", "my invoices", "portaal", "portal",
|
||
"mijn account", "my account", "log in to view", "inloggen om te bekijken",
|
||
]
|
||
|
||
TRASH_SENDER_RE = re.compile(
|
||
r"(noreply|no-reply|donotreply|do-not-reply|notifications?|newsletter|"
|
||
r"nieuwsbrief|alerts?|updates?|mailer-daemon|postmaster|bounce|"
|
||
r"auto-?reply|automatisch)@",
|
||
re.I,
|
||
)
|
||
TRASH_SUBJECT_RE = re.compile(
|
||
r"(out of office|automatisch antwoord|afwezig|buiten kantoor|"
|
||
r"delivery (status notification|failed|failure)|undeliverable|"
|
||
r"mail delivery|mailer-daemon|bounced|returned mail)",
|
||
re.I,
|
||
)
|
||
|
||
|
||
# ── helpers ─────────────────────────────────────────────────────────────────
|
||
|
||
def decode_hdr(value: str | None) -> 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 parse_dt(date_str: str | None) -> datetime | None:
|
||
if not date_str:
|
||
return None
|
||
try:
|
||
return email.utils.parsedate_to_datetime(date_str)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def sender_domain(from_addr: str) -> str:
|
||
m = re.search(r"@([\w.\-]+)", from_addr)
|
||
return m.group(1).lower() if m else "onbekend"
|
||
|
||
|
||
def get_body(msg) -> str:
|
||
parts = []
|
||
try:
|
||
for part in (msg.walk() if msg.is_multipart() else [msg]):
|
||
if part.get_content_type() in ("text/plain", "text/html"):
|
||
cs = part.get_content_charset() or "utf-8"
|
||
try:
|
||
parts.append(part.get_payload(decode=True).decode(cs, errors="replace"))
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
return "\n".join(parts)[:4000]
|
||
|
||
|
||
def get_attachments(msg) -> list[dict]:
|
||
result = []
|
||
try:
|
||
for part in msg.walk():
|
||
disp = (part.get("Content-Disposition") or "").lower()
|
||
ct = part.get_content_type() or ""
|
||
fn = decode_hdr(part.get_filename() or "")
|
||
if "attachment" in disp or (fn and ct == "application/pdf"):
|
||
result.append({"filename": fn, "type": ct})
|
||
except Exception:
|
||
pass
|
||
return result
|
||
|
||
|
||
def extract_urls(text: str) -> list[str]:
|
||
return list(dict.fromkeys(re.findall(r"https?://[^\s<>\"')]+", text)))
|
||
|
||
|
||
def url_domain(url: str) -> str:
|
||
m = re.match(r"https?://([^/]+)", url)
|
||
return m.group(1).lower() if m else ""
|
||
|
||
|
||
# ── main analysis ────────────────────────────────────────────────────────────
|
||
|
||
def analyze():
|
||
REPORTS_DIR.mkdir(exist_ok=True)
|
||
|
||
eml_files = list(MAILBOX_DIR.rglob("*.eml"))
|
||
total = len(eml_files)
|
||
print(f"Analyseren van {total} e-mails...\n")
|
||
|
||
# mailinglijsten: key → {count, list_id, unsubscribe, senders, dates, folder}
|
||
ml: dict[str, dict] = defaultdict(lambda: {
|
||
"list_id": "", "unsubscribe": [], "senders": set(),
|
||
"count": 0, "dates": [], "folder": "",
|
||
})
|
||
|
||
invoices_in_mb: list[dict] = []
|
||
invoice_links: list[dict] = []
|
||
trash: list[dict] = []
|
||
|
||
now_utc = datetime.now(timezone.utc)
|
||
|
||
for i, path in enumerate(eml_files, 1):
|
||
if i % 1000 == 0:
|
||
print(f" {i}/{total}...", flush=True)
|
||
|
||
try:
|
||
with open(path, "rb") as f:
|
||
msg = email.message_from_bytes(f.read())
|
||
except Exception:
|
||
continue
|
||
|
||
folder = path.parent.name
|
||
from_raw = decode_hdr(msg.get("From", ""))
|
||
subject = decode_hdr(msg.get("Subject", ""))
|
||
date = parse_dt(msg.get("Date"))
|
||
list_id = (msg.get("List-ID") or "").strip()
|
||
list_us = (msg.get("List-Unsubscribe") or "").strip()
|
||
prec = (msg.get("Precedence") or "").lower()
|
||
from_lo = from_raw.lower()
|
||
subj_lo = subject.lower()
|
||
domain = sender_domain(from_raw)
|
||
|
||
attachments = get_attachments(msg)
|
||
body = get_body(msg)
|
||
body_lo = body.lower()
|
||
|
||
# ── mailinglijsten ────────────────────────────────────────────────
|
||
ml_key = None
|
||
if list_id:
|
||
ml_key = re.sub(r"[<>]", "", list_id).strip()
|
||
elif "mailinglists" in folder.lower() or "mailinglist" in folder.lower():
|
||
ml_key = folder # use the subfolder name as key
|
||
|
||
if ml_key:
|
||
entry = ml[ml_key]
|
||
entry["list_id"] = list_id or ml_key
|
||
entry["count"] += 1
|
||
entry["folder"] = folder
|
||
entry["senders"].add(from_raw)
|
||
if date:
|
||
entry["dates"].append(date)
|
||
if list_us and list_us not in entry["unsubscribe"]:
|
||
entry["unsubscribe"].append(list_us)
|
||
|
||
# ── facturen in mailbox ───────────────────────────────────────────
|
||
invoice_reasons: list[str] = []
|
||
|
||
for att in attachments:
|
||
fn_lo = att["filename"].lower()
|
||
if att["type"] == "application/pdf" and any(kw in fn_lo for kw in INVOICE_ATTACH_KW):
|
||
invoice_reasons.append(f"PDF: {att['filename']}")
|
||
|
||
if any(kw in subj_lo for kw in INVOICE_SUBJECT_KW):
|
||
invoice_reasons.append("factuurtermen in onderwerp")
|
||
|
||
# always include what's already in the Facturen folders
|
||
if "facturen" in folder.lower():
|
||
invoice_reasons.append(f"staat in map {folder}")
|
||
|
||
if invoice_reasons:
|
||
invoices_in_mb.append({
|
||
"from": from_raw,
|
||
"domain": domain,
|
||
"subject": subject,
|
||
"date": date,
|
||
"folder": folder,
|
||
"reasons": invoice_reasons,
|
||
"attachments": [a["filename"] for a in attachments if a["type"] == "application/pdf"],
|
||
})
|
||
|
||
# ── factuurlinks ──────────────────────────────────────────────────
|
||
has_link_kw = any(kw in body_lo for kw in INVOICE_LINK_KW)
|
||
has_no_pdf = not any(a["type"] == "application/pdf" for a in attachments)
|
||
|
||
if has_link_kw and has_no_pdf:
|
||
urls = extract_urls(body)
|
||
if urls:
|
||
invoice_links.append({
|
||
"from": from_raw,
|
||
"domain": domain,
|
||
"subject": subject,
|
||
"date": date,
|
||
"folder": folder,
|
||
"urls": urls[:5],
|
||
"url_domains": list(dict.fromkeys(url_domain(u) for u in urls[:5])),
|
||
})
|
||
|
||
# ── prullenbak ────────────────────────────────────────────────────
|
||
trash_reasons: list[str] = []
|
||
|
||
if TRASH_SENDER_RE.search(from_lo):
|
||
trash_reasons.append("automatisch afzenderadres")
|
||
|
||
if TRASH_SUBJECT_RE.search(subj_lo):
|
||
trash_reasons.append("automatisch/fout-onderwerp")
|
||
|
||
if prec in ("bulk", "junk"):
|
||
trash_reasons.append(f"Precedence: {prec}")
|
||
|
||
# Old bulk mail sitting in INBOX
|
||
if folder == "INBOX" and prec in ("bulk", "list", "junk") and date:
|
||
age = (now_utc - date.astimezone(timezone.utc)).days
|
||
if age > 180:
|
||
trash_reasons.append(f"bulk in INBOX, {age} dagen oud")
|
||
|
||
if trash_reasons:
|
||
trash.append({
|
||
"from": from_raw,
|
||
"domain": domain,
|
||
"subject": subject,
|
||
"date": date,
|
||
"folder": folder,
|
||
"reasons": trash_reasons,
|
||
"path": str(path),
|
||
})
|
||
|
||
print(f" {total}/{total} klaar.\n")
|
||
|
||
write_mailinglist_report(ml)
|
||
write_invoice_report(invoices_in_mb)
|
||
write_invoice_links_report(invoice_links)
|
||
write_trash_report(trash)
|
||
|
||
print(f"\nRapporten opgeslagen in: {REPORTS_DIR.resolve()}")
|
||
|
||
|
||
# ── report writers ───────────────────────────────────────────────────────────
|
||
|
||
def fmt_date(d: datetime | None) -> str:
|
||
return d.strftime("%Y-%m-%d") if d else "?"
|
||
|
||
|
||
def write_mailinglist_report(ml: dict):
|
||
lines = [
|
||
"# Mailinglijsten\n",
|
||
f"Totaal gevonden: **{len(ml)}** lijsten.\n",
|
||
"_Advies: > 30 mails = hoog volume, overweeg afmelden._\n",
|
||
]
|
||
|
||
for key, entry in sorted(ml.items(), key=lambda x: -x[1]["count"]):
|
||
dates = sorted(entry["dates"])
|
||
oldest = fmt_date(dates[0]) if dates else "?"
|
||
newest = fmt_date(dates[-1]) if dates else "?"
|
||
senders = sorted(entry["senders"])[:2]
|
||
unsub = entry["unsubscribe"]
|
||
|
||
if entry["count"] > 30:
|
||
advice = "⚠️ Hoog volume – overweeg afmelden"
|
||
elif entry["count"] > 10:
|
||
advice = "Gemiddeld volume – beoordeel zelf"
|
||
else:
|
||
advice = "Laag volume – waarschijnlijk houden"
|
||
|
||
can_unsub = "Ja" if unsub else "Nee (handmatig opzoeken)"
|
||
|
||
lines += [
|
||
f"## {key}",
|
||
f"- **Mails:** {entry['count']} | **Periode:** {oldest} – {newest}",
|
||
f"- **Map:** {entry['folder']}",
|
||
f"- **Afzender:** {', '.join(senders)}",
|
||
f"- **Afmelden mogelijk:** {can_unsub}",
|
||
]
|
||
if unsub:
|
||
lines.append(f"- **Afmeldlink:** `{unsub[0][:120]}`")
|
||
lines += [f"- **Advies:** {advice}", ""]
|
||
|
||
path = REPORTS_DIR / "mailinglijsten.md"
|
||
path.write_text("\n".join(lines), encoding="utf-8")
|
||
print(f"✓ mailinglijsten.md ({len(ml)} lijsten)")
|
||
|
||
|
||
def write_invoice_report(invoices: list[dict]):
|
||
lines = [
|
||
"# Facturen in mailbox\n",
|
||
f"Totaal: **{len(invoices)}** factuurmails gevonden.\n",
|
||
]
|
||
|
||
by_domain: dict[str, list] = defaultdict(list)
|
||
for inv in invoices:
|
||
by_domain[inv["domain"]].append(inv)
|
||
|
||
for domain, items in sorted(by_domain.items(), key=lambda x: -len(x[1])):
|
||
items_sorted = sorted(items, key=lambda x: x["date"] or datetime.min.replace(tzinfo=timezone.utc))
|
||
lines.append(f"## {domain} ({len(items)} mails)")
|
||
for item in items_sorted[:25]:
|
||
atts = f" → {', '.join(item['attachments'])}" if item["attachments"] else ""
|
||
lines.append(f"- {fmt_date(item['date'])} | {item['subject'][:80]}{atts}")
|
||
if len(items) > 25:
|
||
lines.append(f" _...en {len(items) - 25} meer_")
|
||
lines.append("")
|
||
|
||
(REPORTS_DIR / "facturen_mailbox.md").write_text("\n".join(lines), encoding="utf-8")
|
||
print(f"✓ facturen_mailbox.md ({len(invoices)} facturen)")
|
||
|
||
|
||
def write_invoice_links_report(invoice_links: list[dict]):
|
||
lines = [
|
||
"# Factuurlinks – verwijzingen naar portalen\n",
|
||
f"Totaal: **{len(invoice_links)}** mails met factuurverwijzingen (zonder PDF-bijlage).\n",
|
||
"_Dit zijn mails die zeggen 'bekijk je factuur online' – de factuur staat op een website._\n",
|
||
]
|
||
|
||
by_url_domain: dict[str, list] = defaultdict(list)
|
||
for item in invoice_links:
|
||
for ud in (item["url_domains"] or ["onbekend"]):
|
||
by_url_domain[ud].append(item)
|
||
|
||
for ud, items in sorted(by_url_domain.items(), key=lambda x: -len(x[1])):
|
||
unique_senders = sorted({i["domain"] for i in items})
|
||
lines.append(f"## {ud} ({len(items)} mails)")
|
||
lines.append(f" _Afzenders: {', '.join(unique_senders[:5])}_")
|
||
for item in sorted(items, key=lambda x: x["date"] or datetime.min.replace(tzinfo=timezone.utc))[:10]:
|
||
lines.append(f"- {fmt_date(item['date'])} | {item['subject'][:80]}")
|
||
for url in item["urls"][:2]:
|
||
lines.append(f" `{url[:120]}`")
|
||
if len(items) > 10:
|
||
lines.append(f" _...en {len(items) - 10} meer_")
|
||
lines.append("")
|
||
|
||
(REPORTS_DIR / "facturen_links.md").write_text("\n".join(lines), encoding="utf-8")
|
||
print(f"✓ facturen_links.md ({len(invoice_links)} mails)")
|
||
|
||
|
||
def write_trash_report(trash: list[dict]):
|
||
lines = [
|
||
"# Prullenbak-kandidaten\n",
|
||
f"Totaal: **{len(trash)}** kandidaten gevonden.\n",
|
||
]
|
||
|
||
by_reason: dict[str, list] = defaultdict(list)
|
||
for item in trash:
|
||
by_reason[item["reasons"][0]].append(item)
|
||
|
||
for reason, items in sorted(by_reason.items(), key=lambda x: -len(x[1])):
|
||
lines.append(f"## {reason} ({len(items)} mails)")
|
||
|
||
by_domain: dict[str, list] = defaultdict(list)
|
||
for item in items:
|
||
by_domain[item["domain"]].append(item)
|
||
|
||
def _sort_key(x):
|
||
d = x["date"]
|
||
if d is None:
|
||
return datetime.min.replace(tzinfo=timezone.utc)
|
||
return d if d.tzinfo else d.replace(tzinfo=timezone.utc)
|
||
|
||
for domain, ditems in sorted(by_domain.items(), key=lambda x: -len(x[1])):
|
||
lines.append(f"### {domain} ({len(ditems)})")
|
||
for item in sorted(ditems, key=_sort_key)[:10]:
|
||
lines.append(f"- {fmt_date(item['date'])} | {item['folder']} | {item['subject'][:70]}")
|
||
if len(ditems) > 10:
|
||
lines.append(f" _...en {len(ditems) - 10} meer_")
|
||
lines.append("")
|
||
|
||
(REPORTS_DIR / "prullenbak.md").write_text("\n".join(lines), encoding="utf-8")
|
||
print(f"✓ prullenbak.md ({len(trash)} kandidaten)")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
analyze()
|