Initial commit
Claude has done a lot of work, but there are still some errors which cause the mails to be distributed to wrong mail boxes. I am now going to let codex and claude compete. They will have their own branches and merge to main as soon as we are somewhere where we can switch AI.
This commit is contained in:
+217
@@ -0,0 +1,217 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Afmelden van mailinglijsten op basis van decisions.json.
|
||||
Verwerkt List-Unsubscribe headers: HTTP (GET/POST) en mailto.
|
||||
|
||||
Gebruik: python3 unsubscribe.py
|
||||
Log: unsubscribe_log.json (resultaat per lijst)
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
import smtplib
|
||||
import ssl
|
||||
import urllib.request
|
||||
import urllib.parse
|
||||
import urllib.error
|
||||
import config
|
||||
from email.mime.text import MIMEText
|
||||
from pathlib import Path
|
||||
|
||||
DECISIONS_FILE = Path(__file__).parent / "decisions.json"
|
||||
LOG_FILE = Path(__file__).parent / "unsubscribe_log.json"
|
||||
REQUEST_TIMEOUT = 10 # seconden per HTTP-verzoek
|
||||
DELAY_BETWEEN = 1.5 # seconden tussen verzoeken (beleefd)
|
||||
|
||||
|
||||
# ── helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
def parse_unsub_header(raw: str) -> dict[str, list[str]]:
|
||||
"""
|
||||
Parseer List-Unsubscribe header naar {'http': [...], 'mailto': [...]}.
|
||||
Formaat: <url>, <url>, ... (RFC 2369)
|
||||
"""
|
||||
result = {"http": [], "mailto": []}
|
||||
for match in re.finditer(r"<([^>]+)>", raw):
|
||||
url = match.group(1).strip()
|
||||
if url.startswith("http"):
|
||||
result["http"].append(url)
|
||||
elif url.startswith("mailto:"):
|
||||
result["mailto"].append(url)
|
||||
return result
|
||||
|
||||
|
||||
def try_http_unsub(url: str) -> tuple[bool, str]:
|
||||
"""Probeer afmelden via HTTP GET. Geeft (succes, melding)."""
|
||||
try:
|
||||
req = urllib.request.Request(
|
||||
url,
|
||||
headers={"User-Agent": "Mozilla/5.0 (compatible; list-unsubscribe)"},
|
||||
)
|
||||
with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT) as resp:
|
||||
code = resp.getcode()
|
||||
return (code == 200), f"HTTP {code}"
|
||||
except urllib.error.HTTPError as e:
|
||||
return False, f"HTTP {e.code}"
|
||||
except Exception as e:
|
||||
return False, str(e)
|
||||
|
||||
|
||||
def try_mailto_unsub(mailto: str, smtp: smtplib.SMTP, from_addr: str) -> tuple[bool, str]:
|
||||
"""Stuur een lege e-mail naar het afmeldadres."""
|
||||
try:
|
||||
parsed = urllib.parse.urlparse(mailto)
|
||||
to_addr = parsed.path
|
||||
params = urllib.parse.parse_qs(parsed.query)
|
||||
subject = params.get("subject", ["unsubscribe"])[0]
|
||||
|
||||
msg = MIMEText("")
|
||||
msg["From"] = from_addr
|
||||
msg["To"] = to_addr
|
||||
msg["Subject"] = subject
|
||||
|
||||
smtp.sendmail(from_addr, [to_addr], msg.as_string())
|
||||
return True, f"mailto verzonden → {to_addr}"
|
||||
except Exception as e:
|
||||
return False, str(e)
|
||||
|
||||
|
||||
# ── main ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
def run():
|
||||
if not DECISIONS_FILE.exists():
|
||||
print("decisions.json niet gevonden. Draai eerst review_mailinglists.py.")
|
||||
sys.exit(1)
|
||||
|
||||
decisions = json.loads(DECISIONS_FILE.read_text(encoding="utf-8"))
|
||||
te_afmelden = {
|
||||
domain: info
|
||||
for domain, info in decisions.items()
|
||||
if isinstance(info, dict) and info.get("keuze") == "a"
|
||||
and info.get("unsubscribe")
|
||||
}
|
||||
handmatig = {
|
||||
domain: info
|
||||
for domain, info in decisions.items()
|
||||
if isinstance(info, dict) and info.get("keuze") == "a"
|
||||
and not info.get("unsubscribe")
|
||||
}
|
||||
|
||||
print(f"Afmelden: {len(te_afmelden)} via List-Unsubscribe, "
|
||||
f"{len(handmatig)} handmatig (geen header).\n")
|
||||
|
||||
# Laad bestaand log (herstart-safe)
|
||||
log: dict[str, dict] = {}
|
||||
if LOG_FILE.exists():
|
||||
try:
|
||||
log = json.loads(LOG_FILE.read_text(encoding="utf-8"))
|
||||
al_klaar = sum(1 for v in log.values() if v.get("succes"))
|
||||
print(f"Bestaand log geladen: {al_klaar} al verwerkt.\n")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
cfg = config.load()
|
||||
|
||||
# SMTP verbinding voor mailto-afmeldingen
|
||||
smtp = None
|
||||
heeft_mailto = any(
|
||||
any(u.startswith("mailto:") for u in info.get("unsubscribe", []))
|
||||
for info in te_afmelden.values()
|
||||
)
|
||||
if heeft_mailto:
|
||||
print(f"Sommige afmeldingen gaan via e-mail (mailto:). Account: {cfg.username}")
|
||||
try:
|
||||
context = ssl.create_default_context()
|
||||
smtp = smtplib.SMTP(cfg.smtp_host, cfg.smtp_port)
|
||||
smtp.starttls(context=context)
|
||||
smtp.login(cfg.username, cfg.password)
|
||||
print("SMTP verbonden.\n")
|
||||
except Exception as e:
|
||||
print(f"SMTP mislukt: {e}\n mailto-afmeldingen worden overgeslagen.")
|
||||
smtp = None
|
||||
|
||||
# Verwerk afmeldingen
|
||||
total = len(te_afmelden)
|
||||
for idx, (domain, info) in enumerate(te_afmelden.items(), 1):
|
||||
naam = info.get("naam", domain)
|
||||
|
||||
if domain in log and log[domain].get("succes"):
|
||||
print(f"[{idx}/{total}] {naam} — al afgemeld, overgeslagen.")
|
||||
continue
|
||||
|
||||
print(f"[{idx}/{total}] {naam} ({domain})")
|
||||
succes = False
|
||||
methode = ""
|
||||
melding = ""
|
||||
|
||||
for raw_unsub in info.get("unsubscribe", []):
|
||||
parsed = parse_unsub_header(raw_unsub)
|
||||
|
||||
# Probeer HTTP eerst
|
||||
for url in parsed["http"]:
|
||||
ok, msg = try_http_unsub(url)
|
||||
if ok:
|
||||
succes = True
|
||||
methode = "HTTP GET"
|
||||
melding = f"{msg} → {url[:80]}"
|
||||
break
|
||||
else:
|
||||
melding = f"{msg} → {url[:60]}"
|
||||
|
||||
if succes:
|
||||
break
|
||||
|
||||
# Dan mailto
|
||||
if smtp:
|
||||
for mailto in parsed["mailto"]:
|
||||
ok, msg = try_mailto_unsub(mailto, smtp, cfg.username)
|
||||
if ok:
|
||||
succes = True
|
||||
methode = "mailto"
|
||||
melding = msg
|
||||
break
|
||||
|
||||
if succes:
|
||||
break
|
||||
|
||||
status = "✓" if succes else "✗"
|
||||
print(f" {status} {methode or 'mislukt'} {melding}")
|
||||
|
||||
log[domain] = {
|
||||
"naam": naam,
|
||||
"succes": succes,
|
||||
"methode": methode,
|
||||
"melding": melding,
|
||||
}
|
||||
LOG_FILE.write_text(json.dumps(log, indent=2, ensure_ascii=False), encoding="utf-8")
|
||||
|
||||
time.sleep(DELAY_BETWEEN)
|
||||
|
||||
if smtp:
|
||||
smtp.quit()
|
||||
|
||||
# Samenvatting
|
||||
gelukt = sum(1 for v in log.values() if v.get("succes"))
|
||||
mislukt = sum(1 for v in log.values() if not v.get("succes"))
|
||||
|
||||
print(f"\n{'─'*50}")
|
||||
print(f"Klaar. {gelukt} afgemeld, {mislukt} mislukt.")
|
||||
|
||||
if mislukt:
|
||||
print("\nMislukt (handmatig afmelden):")
|
||||
for domain, v in log.items():
|
||||
if not v.get("succes"):
|
||||
print(f" {v['naam']} ({domain})")
|
||||
|
||||
if handmatig:
|
||||
print(f"\nGeen List-Unsubscribe header gevonden ({len(handmatig)}) — handmatig afmelden:")
|
||||
for domain, info in handmatig.items():
|
||||
print(f" {info.get('naam', domain)} ({domain})")
|
||||
|
||||
print(f"\nLog opgeslagen in: {LOG_FILE}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
run()
|
||||
Reference in New Issue
Block a user